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

cmake/cmake.define

@@ -46,7 +46,7 @@ ENDIF ()
 
 IF (TD_WINDOWS)
     MESSAGE("${Yellow} set compiler flag for Windows! ${ColourReset}")
-    SET(COMMON_FLAGS "/W3 /D_WIN32")
+    SET(COMMON_FLAGS "/W3 /D_WIN32 /vmg")
 
     # IF (MSVC AND (MSVC_VERSION GREATER_EQUAL 1900))
     #  SET(COMMON_FLAGS "${COMMON_FLAGS} /Wv:18")
@@ -62,7 +62,7 @@ ELSE ()
 	MESSAGE(STATUS "Will compile with Address Sanitizer!")
     ELSE ()
         SET(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -Werror -Werror=return-type -fPIC -gdwarf-2 -g3")
-        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror -Werror=return-type -fPIC -gdwarf-2 -g3")
+        SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-literal-suffix -Werror=return-type -fpermissive -fPIC -gdwarf-2 -g3")
     ENDIF ()
 
     MESSAGE("System processor ID: ${CMAKE_SYSTEM_PROCESSOR}")

example/src/tstream.c

@@ -81,10 +81,10 @@ int32_t create_stream() {
   /*const char* sql = "select min(k), max(k), sum(k) as sum_of_k from st1";*/
   /*const char* sql = "select sum(k) from tu1 interval(10m)";*/
   /*pRes = tmq_create_stream(pConn, "stream1", "out1", sql);*/
-  pRes =
-      taos_query(pConn,
-                 "create stream stream1 trigger window_close as select _wstartts, min(k), max(k), sum(k) as sum_of_k "
-                 "from tu1 interval(10m)");
+  pRes = taos_query(
+      pConn,
+      "create stream stream1 trigger window_close into outstb as select _wstartts, min(k), max(k), sum(k) as sum_of_k "
+      "from tu1 interval(10m)");
   if (taos_errno(pRes) != 0) {
     printf("failed to create stream stream1, reason:%s\n", taos_errstr(pRes));
     return -1;

include/client/taos.h

@@ -195,7 +195,7 @@ DLL_EXPORT void      taos_unsubscribe(TAOS_SUB *tsub, int keepProgress);
 #endif
 
 DLL_EXPORT int       taos_load_table_info(TAOS *taos, const char *tableNameList);
-DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision);
+DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision, bool dataFormat);
 
 /* --------------------------TMQ INTERFACE------------------------------- */
 

include/common/tcommon.h

@@ -194,6 +194,21 @@ typedef struct SExprInfo {
   struct tExprNode*     pExpr;
 } SExprInfo;
 
+typedef struct {
+  const char* key;
+  int32_t keyLen;
+  uint8_t type;
+  int16_t length;
+  union{
+    const char* value;
+    int64_t  i;
+    uint64_t u;
+    double   d;
+    float    f;
+  };
+  int32_t valueLen;
+} SSmlKv;
+
 #define QUERY_ASC_FORWARD_STEP  1
 #define QUERY_DESC_FORWARD_STEP -1
 

include/common/ttypes.h

@@ -186,14 +186,14 @@ typedef struct {
 #define IS_NUMERIC_TYPE(_t) ((IS_SIGNED_NUMERIC_TYPE(_t)) || (IS_UNSIGNED_NUMERIC_TYPE(_t)) || (IS_FLOAT_TYPE(_t)))
 #define IS_MATHABLE_TYPE(_t) (IS_NUMERIC_TYPE(_t) || (_t) == (TSDB_DATA_TYPE_BOOL) || (_t) == (TSDB_DATA_TYPE_TIMESTAMP))
 
-#define IS_VALID_TINYINT(_t)   ((_t) > INT8_MIN && (_t) <= INT8_MAX)
-#define IS_VALID_SMALLINT(_t)  ((_t) > INT16_MIN && (_t) <= INT16_MAX)
-#define IS_VALID_INT(_t)       ((_t) > INT32_MIN && (_t) <= INT32_MAX)
-#define IS_VALID_BIGINT(_t)    ((_t) > INT64_MIN && (_t) <= INT64_MAX)
-#define IS_VALID_UTINYINT(_t)  ((_t) >= 0 && (_t) < UINT8_MAX)
-#define IS_VALID_USMALLINT(_t) ((_t) >= 0 && (_t) < UINT16_MAX)
-#define IS_VALID_UINT(_t)      ((_t) >= 0 && (_t) < UINT32_MAX)
-#define IS_VALID_UBIGINT(_t)   ((_t) >= 0 && (_t) < UINT64_MAX)
+#define IS_VALID_TINYINT(_t)   ((_t) >= INT8_MIN && (_t) <= INT8_MAX)
+#define IS_VALID_SMALLINT(_t)  ((_t) >= INT16_MIN && (_t) <= INT16_MAX)
+#define IS_VALID_INT(_t)       ((_t) >= INT32_MIN && (_t) <= INT32_MAX)
+#define IS_VALID_BIGINT(_t)    ((_t) >= INT64_MIN && (_t) <= INT64_MAX)
+#define IS_VALID_UTINYINT(_t)  ((_t) >= 0 && (_t) <= UINT8_MAX)
+#define IS_VALID_USMALLINT(_t) ((_t) >= 0 && (_t) <= UINT16_MAX)
+#define IS_VALID_UINT(_t)      ((_t) >= 0 && (_t) <= UINT32_MAX)
+#define IS_VALID_UBIGINT(_t)   ((_t) >= 0 && (_t) <= UINT64_MAX)
 #define IS_VALID_FLOAT(_t)     ((_t) >= -FLT_MAX && (_t) <= FLT_MAX)
 #define IS_VALID_DOUBLE(_t)    ((_t) >= -DBL_MAX && (_t) <= DBL_MAX)
 

include/libs/parser/parser.h

@@ -91,17 +91,23 @@ int32_t qCloneStmtDataBlock(void** pDst, void* pSrc);
 void    qFreeStmtDataBlock(void* pDataBlock);
 int32_t qRebuildStmtDataBlock(void** pDst, void* pSrc);
 void    qDestroyStmtDataBlock(void* pBlock);
+
 int32_t qBindStmtColsValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen);
 int32_t qBindStmtSingleColValue(void* pBlock, TAOS_MULTI_BIND* bind, char* msgBuf, int32_t msgBufLen, int32_t colIdx,
                                 int32_t rowNum);
 int32_t qBuildStmtColFields(void* pDataBlock, int32_t* fieldNum, TAOS_FIELD** fields);
 int32_t qBuildStmtTagFields(void* pBlock, void* boundTags, int32_t* fieldNum, TAOS_FIELD** fields);
-int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, SName* pName, TAOS_MULTI_BIND* bind,
+int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, char* tName, TAOS_MULTI_BIND* bind,
                            char* msgBuf, int32_t msgBufLen);
 void    destroyBoundColumnInfo(void* pBoundInfo);
 int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char* dbName, char* msgBuf,
                      int32_t msgBufLen);
 
+void*   smlInitHandle(SQuery *pQuery);
+void    smlDestroyHandle(void *pHandle);
+int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SHashObj *colsHash, SArray *cols, bool format, STableMeta *pTableMeta, char *tableName, char *msgBuf, int16_t msgBufLen);
+int32_t smlBuildOutput(void* handle, SHashObj* pVgHash);
+
 #ifdef __cplusplus
 }
 #endif

include/util/taoserror.h

@@ -642,6 +642,10 @@ int32_t* taosGetErrno();
 #define TSDB_CODE_FUNC_FUNTION_PARA_VALUE       TAOS_DEF_ERROR_CODE(0, 0x2803)
 #define TSDB_CODE_FUNC_INVALID_FUNTION          TAOS_DEF_ERROR_CODE(0, 0x2604)
 
+#define TSDB_CODE_SML_INVALID_PROTOCOL_TYPE     TAOS_DEF_ERROR_CODE(0, 0x3000)
+#define TSDB_CODE_SML_INVALID_PRECISION_TYPE    TAOS_DEF_ERROR_CODE(0, 0x3001)
+#define TSDB_CODE_SML_INVALID_DATA              TAOS_DEF_ERROR_CODE(0, 0x3002)
+
 #ifdef __cplusplus
 }
 #endif

include/util/tdef.h

@@ -86,9 +86,13 @@ extern const int32_t TYPE_BYTES[15];
 #define TS_PATH_DELIMITER "."
 #define TS_ESCAPE_CHAR    '`'
 
+
 #define TSDB_TIME_PRECISION_MILLI 0
 #define TSDB_TIME_PRECISION_MICRO 1
 #define TSDB_TIME_PRECISION_NANO  2
+#define TSDB_TIME_PRECISION_HOURS   3
+#define TSDB_TIME_PRECISION_MINUTES 4
+#define TSDB_TIME_PRECISION_SECONDS 5
 
 #define TSDB_TIME_PRECISION_MILLI_STR "ms"
 #define TSDB_TIME_PRECISION_MICRO_STR "us"

source/client/src/clientMain.c

@@ -787,8 +787,3 @@ int taos_stmt_close(TAOS_STMT *stmt) {
   return stmtClose(stmt);
 }
 
-TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) {
-  // TODO
-  return NULL;
-}
-

source/client/src/clientStmt.c

@@ -458,7 +458,7 @@ int stmtSetTbTags(TAOS_STMT *stmt, TAOS_MULTI_BIND *tags) {
     STMT_ERR_RET(TSDB_CODE_QRY_APP_ERROR);
   }
   
-  STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, &pStmt->bInfo.sname, tags, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
+  STMT_ERR_RET(qBindStmtTagsValue(*pDataBlock, pStmt->bInfo.boundTags, pStmt->bInfo.tbSuid, pStmt->bInfo.sname.tname, tags, pStmt->exec.pRequest->msgBuf, pStmt->exec.pRequest->msgBufLen));
 
   return TSDB_CODE_SUCCESS;
 }

source/client/test/CMakeLists.txt

@@ -17,14 +17,27 @@ TARGET_LINK_LIBRARIES(
         PUBLIC os util common transport parser catalog scheduler function gtest taos_static qcom
 )
 
+ADD_EXECUTABLE(smlTest smlTest.cpp)
+TARGET_LINK_LIBRARIES(
+        smlTest
+        PUBLIC os util common transport parser catalog scheduler function gtest taos_static qcom
+)
+
 TARGET_INCLUDE_DIRECTORIES(
         clientTest
-        PUBLIC "${TD_SOURCE_DIR}/include/libs/client/"
-        PRIVATE "${TD_SOURCE_DIR}/source/libs/client/inc"
+        PUBLIC "${TD_SOURCE_DIR}/include/client/"
+        PRIVATE "${TD_SOURCE_DIR}/source/client/inc"
 )
 
 TARGET_INCLUDE_DIRECTORIES(
         tmqTest
-        PUBLIC "${TD_SOURCE_DIR}/include/libs/client/"
-        PRIVATE "${TD_SOURCE_DIR}/source/libs/client/inc"
+        PUBLIC "${TD_SOURCE_DIR}/include/client/"
+        PRIVATE "${TD_SOURCE_DIR}/source/client/inc"
 )
+
+TARGET_INCLUDE_DIRECTORIES(
+        smlTest
+        PUBLIC "${TD_SOURCE_DIR}/include/client/"
+        PRIVATE "${TD_SOURCE_DIR}/source/client/inc"
+)
+

source/client/test/smlTest.cpp

+/*
+ * 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/>.
+ */
+
+#include <gtest/gtest.h>
+
+#include <taoserror.h>
+#include <tglobal.h>
+#include <iostream>
+
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wwrite-strings"
+#pragma GCC diagnostic ignored "-Wunused-function"
+#pragma GCC diagnostic ignored "-Wunused-variable"
+#pragma GCC diagnostic ignored "-Wsign-compare"
+
+#include "../src/clientSml.c"
+#include "taos.h"
+
+int main(int argc, char **argv) {
+  testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}
+
+TEST(testCase, smlParseString_Test) {
+  char       msg[256] = {0};
+  SSmlMsgBuf msgBuf;
+  msgBuf.buf = msg;
+  msgBuf.len = 256;
+  SSmlLineInfo elements = {0};
+
+  // case 1
+  char *sql = "st,t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000    ,32,c=3";
+  int   ret = smlParseString(sql, &elements, &msgBuf);
+  ASSERT_EQ(ret, 0);
+  ASSERT_EQ(elements.measure, sql);
+  ASSERT_EQ(elements.measureLen, strlen("st"));
+  ASSERT_EQ(elements.measureTagsLen, strlen("st,t1=3,t2=4,t3=t3"));
+
+  ASSERT_EQ(elements.tags, sql + elements.measureLen + 1);
+  ASSERT_EQ(elements.tagsLen, strlen("t1=3,t2=4,t3=t3"));
+
+  ASSERT_EQ(elements.cols, sql + elements.measureTagsLen + 1);
+  ASSERT_EQ(elements.colsLen, strlen("c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64"));
+
+  ASSERT_EQ(elements.timestamp, sql + elements.measureTagsLen + 1 + elements.colsLen + 1);
+  ASSERT_EQ(elements.timestampLen, strlen("1626006833639000000"));
+
+  // case 2  false
+  sql = "st,t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2,c2=false,c4=4f64 1626006833639000000";
+  memset(&elements, 0, sizeof(SSmlLineInfo));
+  ret = smlParseString(sql, &elements, &msgBuf);
+  ASSERT_NE(ret, 0);
+
+  // case 3  false
+  sql = "st, t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2,c2=false,c4=4f64 1626006833639000000";
+  memset(&elements, 0, sizeof(SSmlLineInfo));
+  ret = smlParseString(sql, &elements, &msgBuf);
+  ASSERT_EQ(ret, 0);
+  ASSERT_EQ(elements.cols, sql + elements.measureTagsLen + 2);
+  ASSERT_EQ(elements.colsLen, strlen("t1=3,t2=4,t3=t3"));
+
+  // case 4  tag is null
+  sql = "st, c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000";
+  memset(&elements, 0, sizeof(SSmlLineInfo));
+  ret = smlParseString(sql, &elements, &msgBuf);
+  ASSERT_EQ(ret, 0);
+  ASSERT_EQ(elements.measure, sql);
+  ASSERT_EQ(elements.measureLen, strlen("st"));
+  ASSERT_EQ(elements.measureTagsLen, strlen("st"));
+
+  ASSERT_EQ(elements.tags, sql + elements.measureLen + 1);
+  ASSERT_EQ(elements.tagsLen, 0);
+
+  ASSERT_EQ(elements.cols, sql + elements.measureTagsLen + 2);
+  ASSERT_EQ(elements.colsLen, strlen("c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64"));
+
+  ASSERT_EQ(elements.timestamp, sql + elements.measureTagsLen + 2 + elements.colsLen + 1);
+  ASSERT_EQ(elements.timestampLen, strlen("1626006833639000000"));
+
+  // case 5 tag is null
+  sql = " st   c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64  1626006833639000000 ";
+  memset(&elements, 0, sizeof(SSmlLineInfo));
+  ret = smlParseString(sql, &elements, &msgBuf);
+  sql++;
+  ASSERT_EQ(ret, 0);
+  ASSERT_EQ(elements.measure, sql);
+  ASSERT_EQ(elements.measureLen, strlen("st"));
+  ASSERT_EQ(elements.measureTagsLen, strlen("st"));
+
+  ASSERT_EQ(elements.tags, sql + elements.measureLen);
+  ASSERT_EQ(elements.tagsLen, 0);
+
+  ASSERT_EQ(elements.cols, sql + elements.measureTagsLen + 3);
+  ASSERT_EQ(elements.colsLen, strlen("c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64"));
+
+  ASSERT_EQ(elements.timestamp, sql + elements.measureTagsLen + 3 + elements.colsLen + 2);
+  ASSERT_EQ(elements.timestampLen, strlen("1626006833639000000"));
+
+  // case 6
+  sql = " st   c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64   ";
+  memset(&elements, 0, sizeof(SSmlLineInfo));
+  ret = smlParseString(sql, &elements, &msgBuf);
+  ASSERT_EQ(ret, 0);
+
+  // case 7
+  sql = " st   ,   ";
+  memset(&elements, 0, sizeof(SSmlLineInfo));
+  ret = smlParseString(sql, &elements, &msgBuf);
+  sql++;
+  ASSERT_EQ(ret, 0);
+  ASSERT_EQ(elements.cols, sql + elements.measureTagsLen + 3);
+  ASSERT_EQ(elements.colsLen, strlen(","));
+
+  // case 8 false
+  sql = ", st   ,   ";
+  memset(&elements, 0, sizeof(SSmlLineInfo));
+  ret = smlParseString(sql, &elements, &msgBuf);
+  ASSERT_NE(ret, 0);
+}
+
+TEST(testCase, smlParseCols_Error_Test) {
+  const char *data[] = {
+    "c=\"89sd",           // binary, nchar
+    "c=j\"89sd\"",
+    "c=\"89sd\"k",
+    "c=u",                // bool
+    "c=truet",
+    "c=f64",              // double
+    "c=8f64f",
+    "c=8ef64",
+    "c=1.7976931348623158e+390f64",
+    "c=f32",              // float
+    "c=8f32f",
+    "c=8wef32",
+    "c=-3.402823466e+39f32",
+    "c=",                 // float
+    "c=8f",
+    "c=8we",
+    "c=3.402823466e+39",
+    "c=i8",              // tiny int
+    "c=-8i8f",
+    "c=8wei8",
+    "c=-999i8",
+    "c=u8",              // u tiny int
+    "c=8fu8",
+    "c=8weu8",
+    "c=999u8",
+    "c=-8u8",
+    "c=i16",              // small int
+    "c=8fi16u",
+    "c=8wei16",
+    "c=-67787i16",
+    "c=u16",              // u small int
+    "c=8u16f",
+    "c=8weu16",
+    "c=-9u16",
+    "c=67787u16",
+    "c=i32",                // int
+    "c=8i32f",
+    "c=8wei32",
+    "c=2147483649i32",
+    "c=u32",                // u int
+    "c=8u32f",
+    "c=8weu32",
+    "c=-4u32",
+    "c=42949672958u32",
+    "c=i64",                // big int
+    "c=8i64i",
+    "c=8wei64",
+    "c=-9223372036854775809i64",
+    "c=i",                  // big int
+    "c=8fi",
+    "c=8wei",
+    "c=9223372036854775808i",
+    "c=u64",                // u big int
+    "c=8u64f",
+    "c=8weu64",
+    "c=-3.402823466e+39u64",
+    "c=-339u64",
+    "c=18446744073709551616u64",
+  };
+
+  for(int i = 0; i < sizeof(data)/sizeof(data[0]); i++){
+    char       msg[256] = {0};
+    SSmlMsgBuf msgBuf;
+    msgBuf.buf = msg;
+    msgBuf.len = 256;
+    int32_t len = strlen(data[i]);
+    int32_t ret = smlParseCols(data[i], len, NULL, false, &msgBuf);
+    ASSERT_NE(ret, TSDB_CODE_SUCCESS);
+  }
+}
+
+TEST(testCase, smlParseCols_tag_Test) {
+  char       msg[256] = {0};
+  SSmlMsgBuf msgBuf;
+  msgBuf.buf = msg;
+  msgBuf.len = 256;
+
+  SArray *cols = taosArrayInit(16, POINTER_BYTES);
+  ASSERT_NE(cols, NULL);
+
+  const char *data =
+      "cbin=\"passit hello,c=2\",cnch=L\"iisdfsf\",cbool=false,cf64=4.31f64,cf32_=8.32,cf32=8.23f32,ci8=-34i8,cu8=89u8,ci16=233i16,cu16=898u16,ci32=98289i32,cu32=12323u32,ci64=-89238i64,ci=989i,cu64=8989323u64,cbooltrue=true,cboolt=t,cboolf=f,cnch_=l\"iuwq\"";
+  int32_t len = strlen(data);
+  int32_t ret = smlParseCols(data, len, cols, true, &msgBuf);
+  ASSERT_EQ(ret, TSDB_CODE_SUCCESS);
+  int32_t size = taosArrayGetSize(cols);
+  ASSERT_EQ(size, 19);
+
+  // nchar
+  SSmlKv *kv = taosArrayGetP(cols, 0);
+  ASSERT_EQ(strncasecmp(kv->key, "cbin", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
+  ASSERT_EQ(kv->valueLen, 18);
+  ASSERT_EQ(strncasecmp(kv->value, "\"passit", 7), 0);
+  taosMemoryFree(kv);
+
+  // nchar
+  kv = taosArrayGetP(cols, 3);
+  ASSERT_EQ(strncasecmp(kv->key, "cf64", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
+  ASSERT_EQ(kv->valueLen, 7);
+  ASSERT_EQ(strncasecmp(kv->value, "4.31f64", 7), 0);
+  taosMemoryFree(kv);
+
+  taosArrayClear(cols);
+  data = "t=3e";
+  len = 0;
+  memset(msgBuf.buf, 0, msgBuf.len);
+  ret = smlParseCols(data, len, cols, true, &msgBuf);
+  ASSERT_EQ(ret, TSDB_CODE_SUCCESS);
+  size = taosArrayGetSize(cols);
+  ASSERT_EQ(size, 1);
+
+  // nchar
+  kv = taosArrayGetP(cols, 0);
+  ASSERT_EQ(strncasecmp(kv->key, TAG, strlen(TAG)), 0);
+  ASSERT_EQ(kv->keyLen, strlen(TAG));
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
+  ASSERT_EQ(kv->valueLen, strlen(TAG));
+  ASSERT_EQ(strncasecmp(kv->value, TAG, strlen(TAG)), 0);
+  taosMemoryFree(kv);
+}
+
+TEST(testCase, smlParseCols_Test) {
+  char       msg[256] = {0};
+  SSmlMsgBuf msgBuf;
+  msgBuf.buf = msg;
+  msgBuf.len = 256;
+
+  SArray *cols = taosArrayInit(16, POINTER_BYTES);
+  ASSERT_NE(cols, NULL);
+
+  const char *data = "cbin=\"passit hello,c=2\",cnch=L\"iisdfsf\",cbool=false,cf64=4.31f64,cf32_=8.32,cf32=8.23f32,ci8=-34i8,cu8=89u8,ci16=233i16,cu16=898u16,ci32=98289i32,cu32=12323u32,ci64=-89238i64,ci=989i,cu64=8989323u64,cbooltrue=true,cboolt=t,cboolf=f,cnch_=l\"iuwq\"";
+  int32_t len = strlen(data);
+  int32_t ret = smlParseCols(data, len, cols, false, &msgBuf);
+  ASSERT_EQ(ret, TSDB_CODE_SUCCESS);
+  int32_t size = taosArrayGetSize(cols);
+  ASSERT_EQ(size, 19);
+
+  // binary
+  SSmlKv *kv = taosArrayGetP(cols, 0);
+  ASSERT_EQ(strncasecmp(kv->key, "cbin", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BINARY);
+  ASSERT_EQ(kv->length, 16);
+  ASSERT_EQ(strncasecmp(kv->value, "passit", 6), 0);
+  taosMemoryFree(kv);
+
+  // nchar
+  kv = taosArrayGetP(cols, 1);
+  ASSERT_EQ(strncasecmp(kv->key, "cnch", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
+  ASSERT_EQ(kv->length, 7);
+  ASSERT_EQ(strncasecmp(kv->value, "iisd", 4), 0);
+  taosMemoryFree(kv);
+
+  // bool
+  kv = taosArrayGetP(cols, 2);
+  ASSERT_EQ(strncasecmp(kv->key, "cbool", 5), 0);
+  ASSERT_EQ(kv->keyLen, 5);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BOOL);
+  ASSERT_EQ(kv->length, 1);
+  ASSERT_EQ(kv->i, false);
+  taosMemoryFree(kv);
+
+  // double
+  kv = taosArrayGetP(cols, 3);
+  ASSERT_EQ(strncasecmp(kv->key, "cf64", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_DOUBLE);
+  ASSERT_EQ(kv->length, 8);
+  //ASSERT_EQ(kv->d, 4.31);
+  printf("4.31 = kv->f:%f\n", kv->d);
+  taosMemoryFree(kv);
+
+  // float
+  kv = taosArrayGetP(cols, 4);
+  ASSERT_EQ(strncasecmp(kv->key, "cf32_", 5), 0);
+  ASSERT_EQ(kv->keyLen, 5);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_FLOAT);
+  ASSERT_EQ(kv->length, 4);
+  //ASSERT_EQ(kv->f, 8.32);
+  printf("8.32 = kv->f:%f\n", kv->f);
+  taosMemoryFree(kv);
+
+  // float
+  kv = taosArrayGetP(cols, 5);
+  ASSERT_EQ(strncasecmp(kv->key, "cf32", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_FLOAT);
+  ASSERT_EQ(kv->length, 4);
+  //ASSERT_EQ(kv->f, 8.23);
+  printf("8.23 = kv->f:%f\n", kv->f);
+  taosMemoryFree(kv);
+
+  // tiny int
+  kv = taosArrayGetP(cols, 6);
+  ASSERT_EQ(strncasecmp(kv->key, "ci8", 3), 0);
+  ASSERT_EQ(kv->keyLen, 3);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_TINYINT);
+  ASSERT_EQ(kv->length, 1);
+  ASSERT_EQ(kv->i, -34);
+  taosMemoryFree(kv);
+
+  // unsigned tiny int
+  kv = taosArrayGetP(cols, 7);
+  ASSERT_EQ(strncasecmp(kv->key, "cu8", 3), 0);
+  ASSERT_EQ(kv->keyLen, 3);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_UTINYINT);
+  ASSERT_EQ(kv->length, 1);
+  ASSERT_EQ(kv->u, 89);
+  taosMemoryFree(kv);
+
+  // small int
+  kv = taosArrayGetP(cols, 8);
+  ASSERT_EQ(strncasecmp(kv->key, "ci16", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_SMALLINT);
+  ASSERT_EQ(kv->length, 2);
+  ASSERT_EQ(kv->u, 233);
+  taosMemoryFree(kv);
+
+  // unsigned smallint
+  kv = taosArrayGetP(cols, 9);
+  ASSERT_EQ(strncasecmp(kv->key, "cu16", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_USMALLINT);
+  ASSERT_EQ(kv->length, 2);
+  ASSERT_EQ(kv->u, 898);
+  taosMemoryFree(kv);
+
+  // int
+  kv = taosArrayGetP(cols, 10);
+  ASSERT_EQ(strncasecmp(kv->key, "ci32", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_INT);
+  ASSERT_EQ(kv->length, 4);
+  ASSERT_EQ(kv->u, 98289);
+  taosMemoryFree(kv);
+
+  // unsigned int
+  kv = taosArrayGetP(cols, 11);
+  ASSERT_EQ(strncasecmp(kv->key, "cu32", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_UINT);
+  ASSERT_EQ(kv->length, 4);
+  ASSERT_EQ(kv->u, 12323);
+  taosMemoryFree(kv);
+
+
+  // bigint
+  kv = taosArrayGetP(cols, 12);
+  ASSERT_EQ(strncasecmp(kv->key, "ci64", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BIGINT);
+  ASSERT_EQ(kv->length, 8);
+  ASSERT_EQ(kv->i, -89238);
+  taosMemoryFree(kv);
+
+  // bigint
+  kv = taosArrayGetP(cols, 13);
+  ASSERT_EQ(strncasecmp(kv->key, "ci", 2), 0);
+  ASSERT_EQ(kv->keyLen, 2);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BIGINT);
+  ASSERT_EQ(kv->length, 8);
+  ASSERT_EQ(kv->i, 989);
+  taosMemoryFree(kv);
+
+  // unsigned bigint
+  kv = taosArrayGetP(cols, 14);
+  ASSERT_EQ(strncasecmp(kv->key, "cu64", 4), 0);
+  ASSERT_EQ(kv->keyLen, 4);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_UBIGINT);
+  ASSERT_EQ(kv->length, 8);
+  ASSERT_EQ(kv->u, 8989323);
+  taosMemoryFree(kv);
+
+  // bool
+  kv = taosArrayGetP(cols, 15);
+  ASSERT_EQ(strncasecmp(kv->key, "cbooltrue", 9), 0);
+  ASSERT_EQ(kv->keyLen, 9);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BOOL);
+  ASSERT_EQ(kv->length, 1);
+  ASSERT_EQ(kv->i, true);
+  taosMemoryFree(kv);
+
+
+  // bool
+  kv = taosArrayGetP(cols, 16);
+  ASSERT_EQ(strncasecmp(kv->key, "cboolt", 6), 0);
+  ASSERT_EQ(kv->keyLen, 6);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BOOL);
+  ASSERT_EQ(kv->length, 1);
+  ASSERT_EQ(kv->i, true);
+  taosMemoryFree(kv);
+
+  // bool
+  kv = taosArrayGetP(cols, 17);
+  ASSERT_EQ(strncasecmp(kv->key, "cboolf", 6), 0);
+  ASSERT_EQ(kv->keyLen, 6);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BOOL);
+  ASSERT_EQ(kv->length, 1);
+  ASSERT_EQ(kv->i, false);
+  taosMemoryFree(kv);
+
+  // nchar
+  kv = taosArrayGetP(cols, 18);
+  ASSERT_EQ(strncasecmp(kv->key, "cnch_", 5), 0);
+  ASSERT_EQ(kv->keyLen, 5);
+  ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
+  ASSERT_EQ(kv->length, 4);
+  ASSERT_EQ(strncasecmp(kv->value, "iuwq", 4), 0);
+  taosMemoryFree(kv);
+
+  taosArrayDestroy(cols);
+}
+
+TEST(testCase, smlParseLine_Test) {
+  TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
+  ASSERT_NE(taos, NULL);
+
+  TAOS_RES* pRes = taos_query(taos, "create database if not exists sml_db");
+  taos_free_result(pRes);
+
+  pRes = taos_query(taos, "use sml_db");
+  taos_free_result(pRes);
+
+  SRequestObj *request = createRequest(taos, NULL, NULL, TSDB_SQL_INSERT);
+  ASSERT_NE(request, NULL);
+
+  SSmlHandle *info = smlBuildSmlInfo(taos, request, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS, true);
+  ASSERT_NE(info, NULL);
+
+  const char *sql[3] = {
+    "readings,name=truck_0,fleet=South,driver=Trish,model=H-2,device_version=v2.3 load_capacity=1500,fuel_capacity=150,nominal_fuel_consumption=12,latitude=52.31854,longitude=4.72037,elevation=124,velocity=0,heading=221,grade=0,fuel_consumption=25 1451606400000000000",
+    "readings,name=truck_1,fleet=South,driver=Albert,model=F-150,device_version=v1.5 load_capacity=2000,fuel_capacity=200,nominal_fuel_consumption=15,latitude=72.45258,longitude=68.83761,elevation=255,velocity=0,heading=181,grade=0,fuel_consumption=25 1451606400000000000",
+    "readings,name=truck_2,fleet=North,driver=Derek,model=F-150,device_version=v1.5 load_capacity=2000,fuel_capacity=200,nominal_fuel_consumption=15,latitude=24.5208,longitude=28.09377,elevation=428,velocity=0,heading=304,grade=0,fuel_consumption=25 1451606400000000000"
+  };
+  smlInsertLines(info, sql, 3);
+//  for (int i = 0; i < 3; i++) {
+//    smlParseLine(info, sql[i]);
+//  }
+}
+
+// TEST(testCase, smlParseTS_Test) {
+//   char       msg[256] = {0};
+//   SSmlMsgBuf msgBuf;
+//   msgBuf.buf = msg;
+//   msgBuf.len = 256;
+//   SSmlLineInfo elements = {0};
+//
+//   SSmlHandle* info = smlBuildSmlInfo(taos, request, protocol, precision, dataFormat);
+//   if(!info){
+//     return (TAOS_RES*)request;
+//   }
+//   ret = smlParseTS(info, elements.timestamp, elements.timestampLen, cols);
+//   if(ret != TSDB_CODE_SUCCESS){
+//     uError("SML:0x%"PRIx64" smlParseTS failed", info->id);
+//     return ret;
+//   }
+// }

source/dnode/mgmt/mgmt_vnode/src/vmHandle.c

@@ -293,14 +293,10 @@ void vmInitMsgHandle(SMgmtWrapper *pWrapper) {
   dmSetMsgHandle(pWrapper, TDMT_VND_TASK_MERGE_EXEC, vmProcessMergeMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_TASK_WRITE_EXEC, vmProcessWriteMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_STREAM_TRIGGER, vmProcessFetchMsg, DEFAULT_HANDLE);
-
   dmSetMsgHandle(pWrapper, TDMT_VND_ALTER_VNODE, vmProcessWriteMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_VND_COMPACT_VNODE, vmProcessWriteMsg, DEFAULT_HANDLE);
-
   dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
   dmSetMsgHandle(pWrapper, TDMT_DND_DROP_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
-  // dmSetMsgHandle(pWrapper, TDMT_DND_SYNC_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
-  // dmSetMsgHandle(pWrapper, TDMT_DND_COMPACT_VNODE, vmProcessMgmtMsg, DEFAULT_HANDLE);
 
   // sync integration
   dmSetMsgHandle(pWrapper, TDMT_VND_SYNC_TIMEOUT, (NodeMsgFp)vmProcessSyncMsg, DEFAULT_HANDLE);

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

@@ -31,6 +31,7 @@ int32_t mndCheckDropUserAuth(SUserObj *pOperUser);
 
 int32_t mndCheckNodeAuth(SUserObj *pOperUser);
 int32_t mndCheckFuncAuth(SUserObj *pOperUser);
+int32_t mndCheckTransAuth(SUserObj *pOperUser);
 
 int32_t mndCheckCreateDbAuth(SUserObj *pOperUser);
 int32_t mndCheckAlterDropCompactDbAuth(SUserObj *pOperUser, SDbObj *pDb);

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

@@ -72,7 +72,7 @@ typedef enum {
   TRN_TYPE_DROP_USER = 1003,
   TRN_TYPE_CREATE_FUNC = 1004,
   TRN_TYPE_DROP_FUNC = 1005,
-  
+
   TRN_TYPE_CREATE_SNODE = 1006,
   TRN_TYPE_DROP_SNODE = 1007,
   TRN_TYPE_CREATE_QNODE = 1008,
@@ -601,7 +601,6 @@ typedef struct {
   int32_t        triggerParam;
   int64_t        waterMark;
   char*          sql;
-  char*          logicalPlan;
   char*          physicalPlan;
   SArray*        tasks;  // SArray<SArray<SStreamTask>>
   SSchemaWrapper outputSchema;

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

@@ -27,10 +27,14 @@ void     mndCleanupVgroup(SMnode *pMnode);
 SVgObj  *mndAcquireVgroup(SMnode *pMnode, int32_t vgId);
 void     mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup);
 SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup);
-int32_t  mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups);
 SEpSet   mndGetVgroupEpset(SMnode *pMnode, const SVgObj *pVgroup);
 int32_t  mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId);
 
+int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups);
+SArray *mndBuildDnodesArray(SMnode *pMnode);
+int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray);
+int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray, SVnodeGid *del1, SVnodeGid *del2);
+
 void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
 void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);
 void *mndBuildAlterVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen);

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

@@ -74,10 +74,7 @@ static int32_t mndProcessAuthReq(SNodeMsg *pReq) {
 }
 
 int32_t mndCheckCreateUserAuth(SUserObj *pOperUser) {
-  if (pOperUser->superUser) {
-    return 0;
-  }
-
+  if (pOperUser->superUser) return 0;
   terrno = TSDB_CODE_MND_NO_RIGHTS;
   return -1;
 }
@@ -118,28 +115,25 @@ int32_t mndCheckAlterUserAuth(SUserObj *pOperUser, SUserObj *pUser, SDbObj *pDb,
 }
 
 int32_t mndCheckDropUserAuth(SUserObj *pOperUser) {
-  if (pOperUser->superUser) {
-    return 0;
-  }
-
+  if (pOperUser->superUser) return 0;
   terrno = TSDB_CODE_MND_NO_RIGHTS;
   return -1;
 }
 
 int32_t mndCheckNodeAuth(SUserObj *pOperUser) {
-  if (pOperUser->superUser) {
-    return 0;
-  }
-
+  if (pOperUser->superUser) return 0;
   terrno = TSDB_CODE_MND_NO_RIGHTS;
   return -1;
 }
 
 int32_t mndCheckFuncAuth(SUserObj *pOperUser) {
-  if (pOperUser->superUser) {
-    return 0;
-  }
+  if (pOperUser->superUser) return 0;
+  terrno = TSDB_CODE_MND_NO_RIGHTS;
+  return -1;
+}
 
+int32_t mndCheckTransAuth(SUserObj *pOperUser) {
+  if (pOperUser->superUser) return 0;
   terrno = TSDB_CODE_MND_NO_RIGHTS;
   return -1;
 }

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

@@ -261,6 +261,104 @@ void mndReleaseDb(SMnode *pMnode, SDbObj *pDb) {
   sdbRelease(pSdb, pDb);
 }
 
+static int32_t mndAddCreateVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid,
+                                       bool isRedo) {
+  STransAction action = {0};
+
+  SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
+  if (pDnode == NULL) return -1;
+  action.epSet = mndGetDnodeEpset(pDnode);
+  mndReleaseDnode(pMnode, pDnode);
+
+  int32_t contLen = 0;
+  void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
+  if (pReq == NULL) return -1;
+
+  action.pCont = pReq;
+  action.contLen = contLen;
+  action.msgType = TDMT_DND_CREATE_VNODE;
+  action.acceptableCode = TSDB_CODE_NODE_ALREADY_DEPLOYED;
+
+  if (isRedo) {
+    if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+      taosMemoryFree(pReq);
+      return -1;
+    }
+  } else {
+    if (mndTransAppendUndoAction(pTrans, &action) != 0) {
+      taosMemoryFree(pReq);
+      return -1;
+    }
+  }
+
+  return 0;
+}
+
+static int32_t mndAddAlterVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid,
+                                      bool isRedo) {
+  STransAction action = {0};
+
+  SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
+  if (pDnode == NULL) return -1;
+  action.epSet = mndGetDnodeEpset(pDnode);
+  mndReleaseDnode(pMnode, pDnode);
+
+  int32_t contLen = 0;
+  void   *pReq = mndBuildAlterVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
+  if (pReq == NULL) return -1;
+
+  action.pCont = pReq;
+  action.contLen = contLen;
+  action.msgType = TDMT_VND_ALTER_VNODE;
+
+  if (isRedo) {
+    if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+      taosMemoryFree(pReq);
+      return -1;
+    }
+  } else {
+    if (mndTransAppendUndoAction(pTrans, &action) != 0) {
+      taosMemoryFree(pReq);
+      return -1;
+    }
+  }
+
+  return 0;
+}
+
+static int32_t mndAddDropVnodeAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SVnodeGid *pVgid,
+                                     bool isRedo) {
+  STransAction action = {0};
+
+  SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
+  if (pDnode == NULL) return -1;
+  action.epSet = mndGetDnodeEpset(pDnode);
+  mndReleaseDnode(pMnode, pDnode);
+
+  int32_t contLen = 0;
+  void   *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
+  if (pReq == NULL) return -1;
+
+  action.pCont = pReq;
+  action.contLen = contLen;
+  action.msgType = TDMT_DND_DROP_VNODE;
+  action.acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED;
+
+  if (isRedo) {
+    if (mndTransAppendRedoAction(pTrans, &action) != 0) {
+      taosMemoryFree(pReq);
+      return -1;
+    }
+  } else {
+    if (mndTransAppendUndoAction(pTrans, &action) != 0) {
+      taosMemoryFree(pReq);
+      return -1;
+    }
+  }
+
+  return 0;
+}
+
 static int32_t mndCheckDbName(const char *dbName, SUserObj *pUser) {
   char *pos = strstr(dbName, TS_PATH_DELIMITER);
   if (pos == NULL) {
@@ -278,6 +376,8 @@ static int32_t mndCheckDbName(const char *dbName, SUserObj *pUser) {
 }
 
 static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
+  terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
+
   if (pCfg->numOfVgroups < TSDB_MIN_VNODES_PER_DB || pCfg->numOfVgroups > TSDB_MAX_VNODES_PER_DB) return -1;
   if (pCfg->numOfStables < TSDB_DB_STREAM_MODE_OFF || pCfg->numOfStables > TSDB_DB_STREAM_MODE_ON) return -1;
   if (pCfg->buffer < TSDB_MIN_BUFFER_PER_VNODE || pCfg->buffer > TSDB_MAX_BUFFER_PER_VNODE) return -1;
@@ -299,10 +399,12 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
   if (pCfg->compression < TSDB_MIN_COMP_LEVEL || pCfg->compression > TSDB_MAX_COMP_LEVEL) return -1;
   if (pCfg->replications < TSDB_MIN_DB_REPLICA || pCfg->replications > TSDB_MAX_DB_REPLICA) return -1;
   if (pCfg->replications > mndGetDnodeSize(pMnode)) return -1;
+  if (pCfg->replications != 1 && pCfg->replications != 3) return -1;
   if (pCfg->strict < TSDB_DB_STRICT_OFF || pCfg->strict > TSDB_DB_STRICT_ON) return -1;
-  if (pCfg->strict > pCfg->replications) return -1;
   if (pCfg->cacheLastRow < TSDB_MIN_DB_CACHE_LAST_ROW || pCfg->cacheLastRow > TSDB_MAX_DB_CACHE_LAST_ROW) return -1;
   if (pCfg->hashMethod != 1) return -1;
+
+  terrno = 0;
   return TSDB_CODE_SUCCESS;
 }
 
@@ -381,24 +483,8 @@ static int32_t mndSetCreateDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj
     SVgObj *pVgroup = pVgroups + vg;
 
     for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
-      STransAction action = {0};
-      SVnodeGid   *pVgid = pVgroup->vnodeGid + vn;
-
-      SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
-      if (pDnode == NULL) return -1;
-      action.epSet = mndGetDnodeEpset(pDnode);
-      mndReleaseDnode(pMnode, pDnode);
-
-      int32_t contLen = 0;
-      void   *pReq = mndBuildCreateVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
-      if (pReq == NULL) return -1;
-
-      action.pCont = pReq;
-      action.contLen = contLen;
-      action.msgType = TDMT_DND_CREATE_VNODE;
-      action.acceptableCode = TSDB_CODE_NODE_ALREADY_DEPLOYED;
-      if (mndTransAppendRedoAction(pTrans, &action) != 0) {
-        taosMemoryFree(pReq);
+      SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
+      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, true) != 0) {
         return -1;
       }
     }
@@ -412,24 +498,8 @@ static int32_t mndSetCreateDbUndoActions(SMnode *pMnode, STrans *pTrans, SDbObj
     SVgObj *pVgroup = pVgroups + vg;
 
     for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
-      STransAction action = {0};
-      SVnodeGid   *pVgid = pVgroup->vnodeGid + vn;
-
-      SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
-      if (pDnode == NULL) return -1;
-      action.epSet = mndGetDnodeEpset(pDnode);
-      mndReleaseDnode(pMnode, pDnode);
-
-      int32_t contLen = 0;
-      void   *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
-      if (pReq == NULL) return -1;
-
-      action.pCont = pReq;
-      action.contLen = contLen;
-      action.msgType = TDMT_DND_DROP_VNODE;
-      action.acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED;
-      if (mndTransAppendUndoAction(pTrans, &action) != 0) {
-        taosMemoryFree(pReq);
+      SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
+      if (mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, false) != 0) {
         return -1;
       }
     }
@@ -482,7 +552,6 @@ static int32_t mndCreateDb(SMnode *pMnode, SNodeMsg *pReq, SCreateDbReq *pCreate
   }
 
   if (mndCheckDbCfg(pMnode, &dbObj.cfg) != 0) {
-    terrno = TSDB_CODE_MND_INVALID_DB_OPTION;
     mError("db:%s, failed to create since %s", pCreate->db, terrstr());
     return -1;
   }
@@ -570,37 +639,37 @@ _OVER:
 static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
   terrno = TSDB_CODE_MND_DB_OPTION_UNCHANGED;
 
-  if (pAlter->buffer >= 0 && pAlter->buffer != pDb->cfg.buffer) {
+  if (pAlter->buffer > 0 && pAlter->buffer != pDb->cfg.buffer) {
     pDb->cfg.buffer = pAlter->buffer;
     terrno = 0;
   }
 
-  if (pAlter->pages >= 0 && pAlter->pages != pDb->cfg.pages) {
+  if (pAlter->pages > 0 && pAlter->pages != pDb->cfg.pages) {
     pDb->cfg.pages = pAlter->pages;
     terrno = 0;
   }
 
-  if (pAlter->pageSize >= 0 && pAlter->pageSize != pDb->cfg.pageSize) {
+  if (pAlter->pageSize > 0 && pAlter->pageSize != pDb->cfg.pageSize) {
     pDb->cfg.pageSize = pAlter->pageSize;
     terrno = 0;
   }
 
-  if (pAlter->daysPerFile >= 0 && pAlter->daysPerFile != pDb->cfg.daysPerFile) {
+  if (pAlter->daysPerFile > 0 && pAlter->daysPerFile != pDb->cfg.daysPerFile) {
     pDb->cfg.daysPerFile = pAlter->daysPerFile;
     terrno = 0;
   }
 
-  if (pAlter->daysToKeep0 >= 0 && pAlter->daysToKeep0 != pDb->cfg.daysToKeep0) {
+  if (pAlter->daysToKeep0 > 0 && pAlter->daysToKeep0 != pDb->cfg.daysToKeep0) {
     pDb->cfg.daysToKeep0 = pAlter->daysToKeep0;
     terrno = 0;
   }
 
-  if (pAlter->daysToKeep1 >= 0 && pAlter->daysToKeep1 != pDb->cfg.daysToKeep1) {
+  if (pAlter->daysToKeep1 > 0 && pAlter->daysToKeep1 != pDb->cfg.daysToKeep1) {
     pDb->cfg.daysToKeep1 = pAlter->daysToKeep1;
     terrno = 0;
   }
 
-  if (pAlter->daysToKeep2 >= 0 && pAlter->daysToKeep2 != pDb->cfg.daysToKeep2) {
+  if (pAlter->daysToKeep2 > 0 && pAlter->daysToKeep2 != pDb->cfg.daysToKeep2) {
     pDb->cfg.daysToKeep2 = pAlter->daysToKeep2;
     terrno = 0;
   }
@@ -625,8 +694,9 @@ static int32_t mndSetDbCfgFromAlterDbReq(SDbObj *pDb, SAlterDbReq *pAlter) {
     terrno = 0;
   }
 
-  if (pAlter->replications >= 0 && pAlter->replications != pDb->cfg.replications) {
+  if (pAlter->replications > 0 && pAlter->replications != pDb->cfg.replications) {
     pDb->cfg.replications = pAlter->replications;
+    pDb->vgVersion++;
     terrno = 0;
   }
 
@@ -651,35 +721,57 @@ static int32_t mndSetAlterDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *p
   return 0;
 }
 
-static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup) {
-  for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
-    STransAction action = {0};
-    SVnodeGid   *pVgid = pVgroup->vnodeGid + vn;
-
-    SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
-    if (pDnode == NULL) return -1;
-    action.epSet = mndGetDnodeEpset(pDnode);
-    mndReleaseDnode(pMnode, pDnode);
-
-    int32_t contLen = 0;
-    void   *pReq = mndBuildAlterVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
-    if (pReq == NULL) return -1;
+static int32_t mndBuildAlterVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup, SArray *pArray) {
+  if (pVgroup->replica <= 0 || pVgroup->replica == pDb->cfg.replications) {
+    for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
+      SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
+      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, true) != 0) {
+        return -1;
+      }
+    }
+  } else {
+    SVgObj newVgroup = {0};
+    memcpy(&newVgroup, pVgroup, sizeof(SVgObj));
+    if (newVgroup.replica < pDb->cfg.replications) {
+      mInfo("db:%s, vgId:%d, will add 2 vnodes, vn:0 dnode:%d", pVgroup->dbName, pVgroup->vgId,
+            pVgroup->vnodeGid[0].dnodeId);
+
+      if (mndAddVnodeToVgroup(pMnode, &newVgroup, pArray) != 0) {
+        mError("db:%s, failed to add vnode to vgId:%d since %s", pDb->name, newVgroup.vgId, terrstr());
+        return -1;
+      }
+      newVgroup.replica = pDb->cfg.replications;
+      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[0], true) != 0) return -1;
+      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[1], true) != 0) return -1;
+      if (mndAddCreateVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[2], true) != 0) return -1;
+    } else {
+      mInfo("db:%s, vgId:%d, will remove 2 vnodes", pVgroup->dbName, pVgroup->vgId);
 
-    action.pCont = pReq;
-    action.contLen = contLen;
-    action.msgType = TDMT_VND_ALTER_VNODE;
-    if (mndTransAppendRedoAction(pTrans, &action) != 0) {
-      taosMemoryFree(pReq);
-      return -1;
+      SVnodeGid del1 = {0};
+      SVnodeGid del2 = {0};
+      if (mndRemoveVnodeFromVgroup(pMnode, &newVgroup, pArray, &del1, &del2) != 0) {
+        mError("db:%s, failed to remove vnode from vgId:%d since %s", pDb->name, newVgroup.vgId, terrstr());
+        return -1;
+      }
+      newVgroup.replica = pDb->cfg.replications;
+      if (mndAddAlterVnodeAction(pMnode, pTrans, pDb, &newVgroup, &newVgroup.vnodeGid[0], true) != 0) return -1;
+      if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del1, true) != 0) return -1;
+      if (mndAddDropVnodeAction(pMnode, pTrans, pDb, &newVgroup, &del2, true) != 0) return -1;
     }
+
+    SSdbRaw *pVgRaw = mndVgroupActionEncode(&newVgroup);
+    if (pVgRaw == NULL) return -1;
+    if (mndTransAppendCommitlog(pTrans, pVgRaw) != 0) return -1;
+    if (sdbSetRawStatus(pVgRaw, SDB_STATUS_READY) != 0) return -1;
   }
 
   return 0;
 }
 
 static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *pOld, SDbObj *pNew) {
-  SSdb *pSdb = pMnode->pSdb;
-  void *pIter = NULL;
+  SSdb   *pSdb = pMnode->pSdb;
+  void   *pIter = NULL;
+  SArray *pArray = mndBuildDnodesArray(pMnode);
 
   while (1) {
     SVgObj *pVgroup = NULL;
@@ -687,9 +779,10 @@ static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
     if (pIter == NULL) break;
 
     if (pVgroup->dbUid == pNew->uid) {
-      if (mndBuildAlterVgroupAction(pMnode, pTrans, pNew, pVgroup) != 0) {
+      if (mndBuildAlterVgroupAction(pMnode, pTrans, pNew, pVgroup, pArray) != 0) {
         sdbCancelFetch(pSdb, pIter);
         sdbRelease(pSdb, pVgroup);
+        taosArrayDestroy(pArray);
         return -1;
       }
     }
@@ -697,6 +790,7 @@ static int32_t mndSetAlterDbRedoActions(SMnode *pMnode, STrans *pTrans, SDbObj *
     sdbRelease(pSdb, pVgroup);
   }
 
+  taosArrayDestroy(pArray);
   return 0;
 }
 
@@ -726,6 +820,7 @@ static int32_t mndProcessAlterDbReq(SNodeMsg *pReq) {
   SDbObj     *pDb = NULL;
   SUserObj   *pUser = NULL;
   SAlterDbReq alterReq = {0};
+  SDbObj      dbObj = {0};
 
   if (tDeserializeSAlterDbReq(pReq->rpcMsg.pCont, pReq->rpcMsg.contLen, &alterReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
@@ -749,15 +844,17 @@ static int32_t mndProcessAlterDbReq(SNodeMsg *pReq) {
     goto _OVER;
   }
 
-  SDbObj dbObj = {0};
   memcpy(&dbObj, pDb, sizeof(SDbObj));
-  dbObj.cfg.numOfRetensions = 0;
-  dbObj.cfg.pRetensions = NULL;
+  if (dbObj.cfg.pRetensions != NULL) {
+    dbObj.cfg.pRetensions = taosArrayDup(pDb->cfg.pRetensions);
+    if (dbObj.cfg.pRetensions == NULL) goto _OVER;
+  }
 
   code = mndSetDbCfgFromAlterDbReq(&dbObj, &alterReq);
-  if (code != 0) {
-    goto _OVER;
-  }
+  if (code != 0) goto _OVER;
+
+  code = mndCheckDbCfg(pMnode, &dbObj.cfg);
+  if (code != 0) goto _OVER;
 
   dbObj.cfgVersion++;
   dbObj.updateTime = taosGetTimestampMs();
@@ -771,6 +868,7 @@ _OVER:
 
   mndReleaseDb(pMnode, pDb);
   mndReleaseUser(pMnode, pUser);
+  taosArrayDestroy(dbObj.cfg.pRetensions);
 
   return code;
 }
@@ -899,24 +997,8 @@ static int32_t mndSetDropDbCommitLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pD
 
 static int32_t mndBuildDropVgroupAction(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroup) {
   for (int32_t vn = 0; vn < pVgroup->replica; ++vn) {
-    STransAction action = {0};
-    SVnodeGid   *pVgid = pVgroup->vnodeGid + vn;
-
-    SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
-    if (pDnode == NULL) return -1;
-    action.epSet = mndGetDnodeEpset(pDnode);
-    mndReleaseDnode(pMnode, pDnode);
-
-    int32_t contLen = 0;
-    void   *pReq = mndBuildDropVnodeReq(pMnode, pDnode, pDb, pVgroup, &contLen);
-    if (pReq == NULL) return -1;
-
-    action.pCont = pReq;
-    action.contLen = contLen;
-    action.msgType = TDMT_DND_DROP_VNODE;
-    action.acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED;
-    if (mndTransAppendRedoAction(pTrans, &action) != 0) {
-      taosMemoryFree(pReq);
+    SVnodeGid *pVgid = pVgroup->vnodeGid + vn;
+    if (mndAddDropVnodeAction(pMnode, pTrans, pDb, pVgroup, pVgid, true) != 0) {
       return -1;
     }
   }
@@ -988,6 +1070,17 @@ static int32_t mndDropDb(SMnode *pMnode, SNodeMsg *pReq, SDbObj *pDb) {
   /*if (mndDropTopicByDB(pMnode, pTrans, pDb) != 0) goto _OVER;*/
   if (mndSetDropDbRedoActions(pMnode, pTrans, pDb) != 0) goto _OVER;
 
+  SUserObj *pUser = mndAcquireUser(pMnode, pDb->createUser);
+  if (pUser != NULL) {
+    pUser->authVersion++;
+    SSdbRaw *pCommitRaw = mndUserActionEncode(pUser);
+    if (pCommitRaw == NULL || mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) {
+      mError("trans:%d, failed to append redo log since %s", pTrans->id, terrstr());
+      goto _OVER;
+    }
+    sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY);
+  }
+
   int32_t rspLen = 0;
   void   *pRsp = NULL;
   if (mndBuildDropDbRsp(pDb, &rspLen, &pRsp, false) < 0) goto _OVER;

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

@@ -21,7 +21,7 @@
 #include "mndInfoSchema.h"
 #include "mndMnode.h"
 #include "mndShow.h"
-#include "mndStb.c"
+#include "mndStb.h"
 #include "mndStream.h"
 #include "mndTrans.h"
 #include "mndUser.h"
@@ -419,7 +419,6 @@ static int32_t mndCreateSma(SMnode *pMnode, SNodeMsg *pReq, SMCreateSmaReq *pCre
   streamObj.fixedSinkVgId = smaObj.dstVgId;
   streamObj.smaId = smaObj.uid;
   /*streamObj.physicalPlan = "";*/
-  streamObj.logicalPlan = "not implemented";
 
   int32_t code = -1;
   STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_TYPE_CREATE_SMA, &pReq->rpcMsg);

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

@@ -13,6 +13,7 @@
  * along with this program. If not, see <http://www.gnu.org/licenses/>.
  */
 
+#define _DEFAULT_SOURCE
 #include "mndStb.h"
 #include "mndAuth.h"
 #include "mndDb.h"

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

@@ -290,7 +290,7 @@ int32_t mndAddStreamToTrans(SMnode *pMnode, SStreamObj *pStream, const char *ast
   return 0;
 }
 
-static SStbObj *mndCreateStbForStream(SMnode *pMnode, STrans *pTrans, const SStreamObj *pStream, const char *user) {
+static int32_t mndCreateStbForStream(SMnode *pMnode, STrans *pTrans, const SStreamObj *pStream, const char *user) {
   SStbObj  *pStb = NULL;
   SDbObj   *pDb = NULL;
   SUserObj *pUser = NULL;
@@ -301,7 +301,22 @@ static SStbObj *mndCreateStbForStream(SMnode *pMnode, STrans *pTrans, const SStr
   createReq.numOfTags = 1;  // group id
   createReq.pColumns = taosArrayInit(createReq.numOfColumns, sizeof(SField));
   // build fields
+  taosArraySetSize(createReq.pColumns, createReq.numOfColumns);
+  for (int32_t i = 0; i < createReq.numOfColumns; i++) {
+    SField *pField = taosArrayGet(createReq.pColumns, i);
+    tstrncpy(pField->name, pStream->outputSchema.pSchema[i].name, TSDB_COL_NAME_LEN);
+    pField->flags = pStream->outputSchema.pSchema[i].flags;
+    pField->type = pStream->outputSchema.pSchema[i].type;
+    pField->bytes = pStream->outputSchema.pSchema[i].bytes;
+  }
+  createReq.pTags = taosArrayInit(createReq.numOfTags, sizeof(SField));
+  taosArraySetSize(createReq.pTags, 1);
   // build tags
+  SField *pField = taosArrayGet(createReq.pTags, 0);
+  strcpy(pField->name, "group_id");
+  pField->type = TSDB_DATA_TYPE_UBIGINT;
+  pField->flags = 0;
+  pField->bytes = 8;
 
   if (mndCheckCreateStbReq(&createReq) != 0) {
     terrno = TSDB_CODE_INVALID_MSG;
@@ -342,18 +357,14 @@ static SStbObj *mndCreateStbForStream(SMnode *pMnode, STrans *pTrans, const SStr
     goto _OVER;
   }
 
-  if (mndBuildStbFromReq(pMnode, pStb, &createReq, pDb) != 0) {
-    goto _OVER;
-  }
-
   if (mndAddStbToTrans(pMnode, pTrans, pDb, &stbObj) < 0) goto _OVER;
 
-  return pStb;
+  return 0;
 _OVER:
   mndReleaseStb(pMnode, pStb);
   mndReleaseDb(pMnode, pDb);
   mndReleaseUser(pMnode, pUser);
-  return NULL;
+  return -1;
 }
 
 static int32_t mndCreateStream(SMnode *pMnode, SNodeMsg *pReq, SCMCreateStreamReq *pCreate, SDbObj *pDb) {
@@ -373,7 +384,6 @@ static int32_t mndCreateStream(SMnode *pMnode, SNodeMsg *pReq, SCMCreateStreamRe
   streamObj.fixedSinkVgId = 0;
   streamObj.smaId = 0;
   /*streamObj.physicalPlan = "";*/
-  streamObj.logicalPlan = "not implemented";
   streamObj.trigger = pCreate->triggerType;
   streamObj.waterMark = pCreate->watermark;
 
@@ -384,16 +394,14 @@ static int32_t mndCreateStream(SMnode *pMnode, SNodeMsg *pReq, SCMCreateStreamRe
   }
   mDebug("trans:%d, used to create stream:%s", pTrans->id, pCreate->name);
 
-#if 0
-  if (mndCreateStbForStream(pMnode, pTrans, &streamObj, pReq->user) < 0) {
+  if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, pCreate->triggerType, pCreate->watermark, pTrans) != 0) {
     mError("trans:%d, failed to add stream since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }
-#endif
 
-  if (mndAddStreamToTrans(pMnode, &streamObj, pCreate->ast, pCreate->triggerType, pCreate->watermark, pTrans) != 0) {
-    mError("trans:%d, failed to add stream since %s", pTrans->id, terrstr());
+  if (streamObj.targetSTbName[0] && mndCreateStbForStream(pMnode, pTrans, &streamObj, pReq->user) < 0) {
+    mError("trans:%d, failed to create stb for stream since %s", pTrans->id, terrstr());
     mndTransDrop(pTrans);
     return -1;
   }

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

@@ -1336,8 +1336,7 @@ static int32_t mndProcessKillTransReq(SNodeMsg *pReq) {
     goto _OVER;
   }
 
-  if (!pUser->superUser) {
-    terrno = TSDB_CODE_MND_NO_RIGHTS;
+  if (mndCheckTransAuth(pUser) != 0) {
     goto _OVER;
   }
 

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

@@ -106,6 +106,7 @@ SSdbRaw *mndUserActionEncode(SUserObj *pUser) {
   SDB_SET_INT64(pRaw, dataPos, pUser->createdTime, _OVER)
   SDB_SET_INT64(pRaw, dataPos, pUser->updateTime, _OVER)
   SDB_SET_INT8(pRaw, dataPos, pUser->superUser, _OVER)
+  SDB_SET_INT32(pRaw, dataPos, pUser->authVersion, _OVER)
   SDB_SET_INT32(pRaw, dataPos, numOfReadDbs, _OVER)
   SDB_SET_INT32(pRaw, dataPos, numOfWriteDbs, _OVER)
 
@@ -161,6 +162,7 @@ static SSdbRow *mndUserActionDecode(SSdbRaw *pRaw) {
   SDB_GET_INT64(pRaw, dataPos, &pUser->createdTime, _OVER)
   SDB_GET_INT64(pRaw, dataPos, &pUser->updateTime, _OVER)
   SDB_GET_INT8(pRaw, dataPos, &pUser->superUser, _OVER)
+  SDB_GET_INT32(pRaw, dataPos, &pUser->authVersion, _OVER)
 
   int32_t numOfReadDbs = 0;
   int32_t numOfWriteDbs = 0;
@@ -588,7 +590,7 @@ _OVER:
   return code;
 }
 
-static int32_t mndSetUserAuthRsp(SMnode       *pMnode, SUserObj *pUser, SGetUserAuthRsp *pRsp) {
+static int32_t mndSetUserAuthRsp(SMnode *pMnode, SUserObj *pUser, SGetUserAuthRsp *pRsp) {
   memcpy(pRsp->user, pUser->user, TSDB_USER_LEN);
   pRsp->superAuth = pUser->superUser;
   pRsp->version = pUser->authVersion;
@@ -601,7 +603,7 @@ static int32_t mndSetUserAuthRsp(SMnode       *pMnode, SUserObj *pUser, SGetUser
     terrno = TSDB_CODE_OUT_OF_MEMORY;
     return -1;
   }
-  
+
   SSdb *pSdb = pMnode->pSdb;
   void *pIter = NULL;
   while (1) {
@@ -659,7 +661,7 @@ static int32_t mndProcessGetUserAuthReq(SNodeMsg *pReq) {
   code = 0;
 
 _OVER:
-  
+
   mndReleaseUser(pMnode, pUser);
   tFreeSGetUserAuthRsp(&authRsp);
 
@@ -711,7 +713,8 @@ static void mndCancelGetNextUser(SMnode *pMnode, void *pIter) {
   sdbCancelFetch(pSdb, pIter);
 }
 
-int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_t numOfUses, void **ppRsp, int32_t *pRspLen) {
+int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_t numOfUses, void **ppRsp,
+                                int32_t *pRspLen) {
   SUserAuthBatchRsp batchRsp = {0};
   batchRsp.pArray = taosArrayInit(numOfUses, sizeof(SGetUserAuthRsp));
   if (batchRsp.pArray == NULL) {
@@ -731,7 +734,7 @@ int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_
       mndReleaseUser(pMnode, pUser);
       continue;
     }
-    
+
     SGetUserAuthRsp rsp = {0};
     code = mndSetUserAuthRsp(pMnode, pUser, &rsp);
     if (code) {
@@ -740,7 +743,6 @@ int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_
       goto _OVER;
     }
 
-
     taosArrayPush(batchRsp.pArray, &rsp);
     mndReleaseUser(pMnode, pUser);
   }
@@ -748,7 +750,7 @@ int32_t mndValidateUserAuthInfo(SMnode *pMnode, SUserAuthVersion *pUsers, int32_
   if (taosArrayGetSize(batchRsp.pArray) <= 0) {
     *ppRsp = NULL;
     *pRspLen = 0;
-    
+
     tFreeSUserAuthBatchRsp(&batchRsp);
     return 0;
   }
@@ -772,10 +774,7 @@ _OVER:
 
   *ppRsp = NULL;
   *pRspLen = 0;
-  
+
   tFreeSUserAuthBatchRsp(&batchRsp);
   return code;
 }
-
-
-

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

@@ -370,7 +370,7 @@ static bool mndBuildDnodesArrayFp(SMnode *pMnode, void *pObj, void *p1, void *p2
   return true;
 }
 
-static SArray *mndBuildDnodesArray(SMnode *pMnode) {
+SArray *mndBuildDnodesArray(SMnode *pMnode) {
   SSdb   *pSdb = pMnode->pSdb;
   int32_t numOfDnodes = mndGetDnodeSize(pMnode);
 
@@ -421,7 +421,7 @@ static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup, SArray *pAr
       pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
     }
 
-    mDebug("db:%s, vgId:%d, vn:%d dnode:%d is alloced", pVgroup->dbName, pVgroup->vgId, v, pVgid->dnodeId);
+    mInfo("db:%s, vgId:%d, vn:%d dnode:%d is alloced", pVgroup->dbName, pVgroup->vgId, v, pVgid->dnodeId);
     pDnode->numOfVnodes++;
   }
 
@@ -440,12 +440,10 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
   }
 
   pArray = mndBuildDnodesArray(pMnode);
-  if (pArray == NULL) {
-    goto _OVER;
-  }
+  if (pArray == NULL) goto _OVER;
 
-  mDebug("db:%s, total %d dnodes used to create %d vgroups (%d vnodes)", pDb->name, (int32_t)taosArrayGetSize(pArray),
-         pDb->cfg.numOfVgroups, pDb->cfg.numOfVgroups * pDb->cfg.replications);
+  mInfo("db:%s, total %d dnodes used to create %d vgroups (%d vnodes)", pDb->name, (int32_t)taosArrayGetSize(pArray),
+        pDb->cfg.numOfVgroups, pDb->cfg.numOfVgroups * pDb->cfg.replications);
 
   int32_t  allocedVgroups = 0;
   int32_t  maxVgId = sdbGetMaxId(pMnode->pSdb, SDB_VGROUP);
@@ -483,7 +481,7 @@ int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
   *ppVgroups = pVgroups;
   code = 0;
 
-  mDebug("db:%s, %d vgroups is alloced, replica:%d", pDb->name, pDb->cfg.numOfVgroups, pDb->cfg.replications);
+  mInfo("db:%s, %d vgroups is alloced, replica:%d", pDb->name, pDb->cfg.numOfVgroups, pDb->cfg.replications);
 
 _OVER:
   if (code != 0) taosMemoryFree(pVgroups);
@@ -491,6 +489,88 @@ _OVER:
   return code;
 }
 
+int32_t mndAddVnodeToVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
+  taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
+  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
+    SDnodeObj *pDnode = taosArrayGet(pArray, i);
+    mDebug("dnode:%d, equivalent vnodes:%d", pDnode->id, pDnode->numOfVnodes);
+  }
+
+  int32_t maxPos = 1;
+  for (int32_t d = 0; d < taosArrayGetSize(pArray); ++d) {
+    SDnodeObj *pDnode = taosArrayGet(pArray, d);
+
+    bool used = false;
+    for (int32_t vn = 0; vn < maxPos; ++vn) {
+      if (pDnode->id == pVgroup->vnodeGid[vn].dnodeId) {
+        used = true;
+        break;
+      }
+    }
+    if (used) continue;
+
+    if (pDnode == NULL || pDnode->numOfVnodes > pDnode->numOfSupportVnodes) {
+      terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
+      return -1;
+    }
+
+    SVnodeGid *pVgid = &pVgroup->vnodeGid[maxPos];
+    pVgid->dnodeId = pDnode->id;
+    pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
+    pDnode->numOfVnodes++;
+
+    mInfo("db:%s, vgId:%d, vn:%d dnode:%d is added", pVgroup->dbName, pVgroup->vgId, maxPos, pVgid->dnodeId);
+    maxPos++;
+    if (maxPos == 3) return 0;
+  }
+
+  terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
+  return -1;
+}
+
+int32_t mndRemoveVnodeFromVgroup(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray, SVnodeGid *del1, SVnodeGid *del2) {
+  int32_t removedNum = 0;
+
+  taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);
+  for (int32_t i = 0; i < taosArrayGetSize(pArray); ++i) {
+    SDnodeObj *pDnode = taosArrayGet(pArray, i);
+    mDebug("dnode:%d, equivalent vnodes:%d", pDnode->id, pDnode->numOfVnodes);
+  }
+
+  for (int32_t d = taosArrayGetSize(pArray) - 1; d >= 0; --d) {
+    SDnodeObj *pDnode = taosArrayGet(pArray, d);
+
+    for (int32_t vn = 0; vn < TSDB_MAX_REPLICA; ++vn) {
+      SVnodeGid *pVgid = &pVgroup->vnodeGid[vn];
+      if (pVgid->dnodeId == pDnode->id) {
+        if (removedNum == 0) *del1 = *pVgid;
+        if (removedNum == 1) *del2 = *pVgid;
+
+        mInfo("db:%s, vgId:%d, vn:%d dnode:%d is removed", pVgroup->dbName, pVgroup->vgId, vn, pVgid->dnodeId);
+        memset(pVgid, 0, sizeof(SVnodeGid));
+        removedNum++;
+        pDnode->numOfVnodes--;
+
+        if (removedNum == 2) goto _OVER;
+      }
+    }
+  }
+
+_OVER:
+  if (removedNum != 2) return -1;
+
+  for (int32_t vn = 1; vn < TSDB_MAX_REPLICA; ++vn) {
+    SVnodeGid *pVgid = &pVgroup->vnodeGid[vn];
+    if (pVgid->dnodeId != 0) {
+      memcpy(&pVgroup->vnodeGid[0], pVgid, sizeof(SVnodeGid));
+      memset(pVgid, 0, sizeof(SVnodeGid));
+    }
+  }
+
+  mInfo("db:%s, vgId:%d, dnode:%d is keeped", pVgroup->dbName, pVgroup->vgId, pVgroup->vnodeGid[0].dnodeId);
+  return 0;
+}
+
 SEpSet mndGetVgroupEpset(SMnode *pMnode, const SVgObj *pVgroup) {
   SEpSet epset = {0};
 

source/dnode/mnode/impl/test/db/db.cpp

@@ -73,14 +73,19 @@ TEST_F(MndTestDb, 02_Create_Alter_Drop_Db) {
   {
     SAlterDbReq alterdbReq = {0};
     strcpy(alterdbReq.db, "1.d1");
+
     alterdbReq.buffer = 12;
-    alterdbReq.daysToKeep0 = 300;
-    alterdbReq.daysToKeep1 = 400;
-    alterdbReq.daysToKeep2 = 500;
+    alterdbReq.pageSize = -1;
+    alterdbReq.pages = -1;
+    alterdbReq.daysPerFile = -1;
+    alterdbReq.daysToKeep0 = -1;
+    alterdbReq.daysToKeep1 = -1;
+    alterdbReq.daysToKeep2 = -1;
     alterdbReq.fsyncPeriod = 4000;
     alterdbReq.walLevel = 2;
-    alterdbReq.strict = 2;
+    alterdbReq.strict = 1;
     alterdbReq.cacheLastRow = 1;
+    alterdbReq.replications = 1;
 
     int32_t contLen = tSerializeSAlterDbReq(NULL, 0, &alterdbReq);
     void*   pReq = rpcMallocCont(contLen);

source/dnode/mnode/impl/test/dnode/mdnode.cpp

@@ -266,4 +266,96 @@ TEST_F(MndTestDnode, 05_Create_Drop_Restart_Dnode) {
   taosMsleep(1300);
   test.SendShowReq(TSDB_MGMT_TABLE_DNODE, "dnodes", "");
   EXPECT_EQ(test.GetShowRows(), 4);
+
+  // alter replica
+#if 0
+  {
+    SCreateDbReq createReq = {0};
+    strcpy(createReq.db, "1.d2");
+    createReq.numOfVgroups = 2;
+    createReq.buffer = -1;
+    createReq.pageSize = -1;
+    createReq.pages = -1;
+    createReq.daysPerFile = 1000;
+    createReq.daysToKeep0 = 3650;
+    createReq.daysToKeep1 = 3650;
+    createReq.daysToKeep2 = 3650;
+    createReq.minRows = 100;
+    createReq.maxRows = 4096;
+    createReq.fsyncPeriod = 3000;
+    createReq.walLevel = 1;
+    createReq.precision = 0;
+    createReq.compression = 2;
+    createReq.replications = 1;
+    createReq.strict = 1;
+    createReq.cacheLastRow = 0;
+    createReq.ignoreExist = 1;
+    createReq.numOfStables = 0;
+    createReq.numOfRetensions = 0;
+
+    int32_t contLen = tSerializeSCreateDbReq(NULL, 0, &createReq);
+    void*   pReq = rpcMallocCont(contLen);
+    tSerializeSCreateDbReq(pReq, contLen, &createReq);
+
+    SRpcMsg* pRsp = test.SendReq(TDMT_MND_CREATE_DB, pReq, contLen);
+    ASSERT_NE(pRsp, nullptr);
+    ASSERT_EQ(pRsp->code, 0);
+
+    test.SendShowReq(TSDB_MGMT_TABLE_DB, "user_databases", "");
+    EXPECT_EQ(test.GetShowRows(), 3);
+  }
+
+  {
+    SAlterDbReq alterdbReq = {0};
+    strcpy(alterdbReq.db, "1.d2");
+
+    alterdbReq.buffer = 12;
+    alterdbReq.pageSize = -1;
+    alterdbReq.pages = -1;
+    alterdbReq.daysPerFile = -1;
+    alterdbReq.daysToKeep0 = -1;
+    alterdbReq.daysToKeep1 = -1;
+    alterdbReq.daysToKeep2 = -1;
+    alterdbReq.fsyncPeriod = 4000;
+    alterdbReq.walLevel = 2;
+    alterdbReq.strict = 1;
+    alterdbReq.cacheLastRow = 1;
+    alterdbReq.replications = 3;
+
+    int32_t contLen = tSerializeSAlterDbReq(NULL, 0, &alterdbReq);
+    void*   pReq = rpcMallocCont(contLen);
+    tSerializeSAlterDbReq(pReq, contLen, &alterdbReq);
+
+    SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_DB, pReq, contLen);
+    ASSERT_NE(pRsp, nullptr);
+    ASSERT_EQ(pRsp->code, 0);
+  }
+
+  {
+    SAlterDbReq alterdbReq = {0};
+    strcpy(alterdbReq.db, "1.d2");
+
+    alterdbReq.buffer = 12;
+    alterdbReq.pageSize = -1;
+    alterdbReq.pages = -1;
+    alterdbReq.daysPerFile = -1;
+    alterdbReq.daysToKeep0 = -1;
+    alterdbReq.daysToKeep1 = -1;
+    alterdbReq.daysToKeep2 = -1;
+    alterdbReq.fsyncPeriod = 4000;
+    alterdbReq.walLevel = 2;
+    alterdbReq.strict = 1;
+    alterdbReq.cacheLastRow = 1;
+    alterdbReq.replications = 1;
+
+    int32_t contLen = tSerializeSAlterDbReq(NULL, 0, &alterdbReq);
+    void*   pReq = rpcMallocCont(contLen);
+    tSerializeSAlterDbReq(pReq, contLen, &alterdbReq);
+
+    SRpcMsg* pRsp = test.SendReq(TDMT_MND_ALTER_DB, pReq, contLen);
+    ASSERT_NE(pRsp, nullptr);
+    ASSERT_EQ(pRsp->code, 0);
+  }
+
+#endif
 }

source/libs/CMakeLists.txt

@@ -17,4 +17,4 @@ add_subdirectory(tfs)
 add_subdirectory(monitor)
 add_subdirectory(nodes)
 add_subdirectory(scalar)
-add_subdirectory(command)
+add_subdirectory(command)
\ No newline at end of file

source/libs/parser/inc/parInsertData.h

@@ -133,15 +133,15 @@ static FORCE_INLINE int32_t setBlockInfo(SSubmitBlk *pBlocks, STableDataBlocks *
 int32_t schemaIdxCompar(const void *lhs, const void *rhs);
 int32_t boundIdxCompar(const void *lhs, const void *rhs);
 void    setBoundColumnInfo(SParsedDataColInfo *pColList, SSchema *pSchema, col_id_t numOfCols);
-void    destroyBlockArrayList(SArray *pDataBlockList);
-void    destroyBlockHashmap(SHashObj *pDataBlockHash);
-int     initRowBuilder(SRowBuilder *pBuilder, int16_t schemaVer, SParsedDataColInfo *pColInfo);
-int32_t allocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int32_t *numOfRows);
-int32_t getDataBlockFromList(SHashObj *pHashList, int64_t id, int32_t size, int32_t startOffset, int32_t rowSize,
-                             const STableMeta *pTableMeta, STableDataBlocks **dataBlocks, SArray *pBlockList,
-                             SVCreateTbReq *pCreateTbReq);
-int32_t mergeTableDataBlocks(SHashObj *pHashObj, uint8_t payloadType, SArray **pVgDataBlocks);
-int32_t buildCreateTbMsg(STableDataBlocks *pBlocks, SVCreateTbReq *pCreateTbReq);
+void destroyBlockArrayList(SArray* pDataBlockList);
+void destroyBlockHashmap(SHashObj* pDataBlockHash);
+int  initRowBuilder(SRowBuilder *pBuilder, int16_t schemaVer, SParsedDataColInfo *pColInfo);
+int32_t allocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize, int32_t * numOfRows);
+int32_t getDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, int32_t startOffset, int32_t rowSize,
+     STableMeta* pTableMeta, STableDataBlocks** dataBlocks, SArray* pBlockList, SVCreateTbReq* pCreateTbReq);
+int32_t mergeTableDataBlocks(SHashObj* pHashObj, uint8_t payloadType, SArray** pVgDataBlocks);
+int32_t buildCreateTbMsg(STableDataBlocks* pBlocks, SVCreateTbReq* pCreateTbReq);
+
 int32_t allocateMemForSize(STableDataBlocks *pDataBlock, int32_t allSize);
 
 #endif  // TDENGINE_DATABLOCKMGT_H

source/libs/parser/inc/parUtil.h

@@ -41,7 +41,6 @@ int32_t generateSyntaxErrMsg(SMsgBuf* pBuf, int32_t errCode, ...);
 int32_t buildInvalidOperationMsg(SMsgBuf* pMsgBuf, const char* msg);
 int32_t buildSyntaxErrMsg(SMsgBuf* pBuf, const char* additionalInfo, const char* sourceStr);
 
-STableMeta*   tableMetaDup(const STableMeta* pTableMeta);
 SSchema*      getTableColumnSchema(const STableMeta* pTableMeta);
 SSchema*      getTableTagSchema(const STableMeta* pTableMeta);
 int32_t       getNumOfColumns(const STableMeta* pTableMeta);

source/libs/parser/src/parInsert.c

@@ -762,11 +762,9 @@ static int32_t KvRowAppend(SMsgBuf* pMsgBuf, const void* value, int32_t len, voi
   return TSDB_CODE_SUCCESS;
 }
 
-static int32_t buildCreateTbReq(SVCreateTbReq* pTbReq, const SName* pName, SKVRow row, int64_t suid) {
-  char dbFName[TSDB_DB_FNAME_LEN] = {0};
-  tNameGetFullDbName(pName, dbFName);
+static int32_t buildCreateTbReq(SVCreateTbReq *pTbReq, const char* tname, SKVRow row, int64_t suid) {
   pTbReq->type = TD_CHILD_TABLE;
-  pTbReq->name = strdup(pName->tname);
+  pTbReq->name = strdup(tname);
   pTbReq->ctb.suid = suid;
   pTbReq->ctb.pTag = row;
 
@@ -774,7 +772,7 @@ static int32_t buildCreateTbReq(SVCreateTbReq* pTbReq, const SName* pName, SKVRo
 }
 
 // pSql -> tag1_value, ...)
-static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint8_t precision, const SName* pName) {
+static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint8_t precision, const char* tName) {
   if (tdInitKVRowBuilder(&pCxt->tagsBuilder) < 0) {
     return TSDB_CODE_TSC_OUT_OF_MEMORY;
   }
@@ -815,7 +813,7 @@ static int32_t parseTagsClause(SInsertParseContext* pCxt, SSchema* pSchema, uint
   }
   tdSortKVRowByColIdx(row);
 
-  return buildCreateTbReq(&pCxt->createTblReq, pName, row, pCxt->pTableMeta->suid);
+  return buildCreateTbReq(&pCxt->createTblReq, tName, row, pCxt->pTableMeta->suid);
 }
 
 static int32_t cloneTableMeta(STableMeta* pSrc, STableMeta** pDst) {
@@ -884,7 +882,7 @@ static int32_t parseUsingClause(SInsertParseContext* pCxt, SToken* pTbnameToken)
   if (TK_NK_LP != sToken.type) {
     return buildSyntaxErrMsg(&pCxt->msg, "( is expected", sToken.z);
   }
-  CHECK_CODE(parseTagsClause(pCxt, pCxt->pTableMeta->schema, getTableInfo(pCxt->pTableMeta).precision, &name));
+  CHECK_CODE(parseTagsClause(pCxt, pCxt->pTableMeta->schema, getTableInfo(pCxt->pTableMeta).precision, name.tname));
   NEXT_TOKEN(pCxt->pSql, sToken);
   if (TK_NK_RP != sToken.type) {
     return buildSyntaxErrMsg(&pCxt->msg, ") is expected", sToken.z);
@@ -1038,11 +1036,6 @@ static void destroyDataBlock(STableDataBlocks* pDataBlock) {
 
   taosMemoryFreeClear(pDataBlock->pData);
   if (!pDataBlock->cloned) {
-    // free the refcount for metermeta
-    if (pDataBlock->pTableMeta != NULL) {
-      taosMemoryFreeClear(pDataBlock->pTableMeta);
-    }
-
     destroyBoundColumnInfo(&pDataBlock->boundColumnInfo);
   }
   taosMemoryFreeClear(pDataBlock);
@@ -1270,10 +1263,9 @@ int32_t qBuildStmtOutput(SQuery* pQuery, SHashObj* pVgHash, SHashObj* pBlockHash
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, SName* pName, TAOS_MULTI_BIND* bind,
-                           char* msgBuf, int32_t msgBufLen) {
-  STableDataBlocks*   pDataBlock = (STableDataBlocks*)pBlock;
-  SMsgBuf             pBuf = {.buf = msgBuf, .len = msgBufLen};
+int32_t qBindStmtTagsValue(void *pBlock, void *boundTags, int64_t suid, char *tName, TAOS_MULTI_BIND *bind, char *msgBuf, int32_t msgBufLen){
+  STableDataBlocks *pDataBlock = (STableDataBlocks *)pBlock;
+  SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};
   SParsedDataColInfo* tags = (SParsedDataColInfo*)boundTags;
   if (NULL == tags) {
     return TSDB_CODE_QRY_APP_ERROR;
@@ -1312,7 +1304,7 @@ int32_t qBindStmtTagsValue(void* pBlock, void* boundTags, int64_t suid, SName* p
   tdSortKVRowByColIdx(row);
 
   SVCreateTbReq tbReq = {0};
-  CHECK_CODE(buildCreateTbReq(&tbReq, pName, row, suid));
+  CHECK_CODE(buildCreateTbReq(&tbReq, tName, row, suid));
   CHECK_CODE(buildCreateTbMsg(pDataBlock, &tbReq));
 
   destroyCreateSubTbReq(&tbReq);
@@ -1544,3 +1536,265 @@ int32_t qBuildStmtColFields(void* pBlock, int32_t* fieldNum, TAOS_FIELD** fields
 
   return TSDB_CODE_SUCCESS;
 }
+
+// schemaless logic start
+
+typedef struct SmlExecHandle {
+  SHashObj*    pBlockHash;
+
+  SParsedDataColInfo tags;      // each table
+  SKVRowBuilder tagsBuilder;    // each table
+  SVCreateTbReq createTblReq;   // each table
+
+  SQuery* pQuery;
+} SSmlExecHandle;
+
+static int32_t smlBoundColumns(SArray *cols, SParsedDataColInfo* pColList, SSchema* pSchema) {
+  col_id_t nCols = pColList->numOfCols;
+
+  pColList->numOfBound = 0;
+  pColList->boundNullLen = 0;
+  memset(pColList->boundColumns, 0, sizeof(col_id_t) * nCols);
+  for (col_id_t i = 0; i < nCols; ++i) {
+    pColList->cols[i].valStat = VAL_STAT_NONE;
+  }
+
+  bool     isOrdered = true;
+  col_id_t lastColIdx = -1;  // last column found
+  for (int i = 0; i < taosArrayGetSize(cols); ++i) {
+    SSmlKv *kv = taosArrayGetP(cols, i);
+    SToken sToken = {.n=kv->keyLen, .z=(char*)kv->key};
+    col_id_t t = lastColIdx + 1;
+    col_id_t index = findCol(&sToken, t, nCols, pSchema);
+    if (index < 0 && t > 0) {
+      index = findCol(&sToken, 0, t, pSchema);
+      isOrdered = false;
+    }
+    if (index < 0) {
+      return TSDB_CODE_SML_INVALID_DATA;
+    }
+    if (pColList->cols[index].valStat == VAL_STAT_HAS) {
+      return TSDB_CODE_SML_INVALID_DATA;
+    }
+    lastColIdx = index;
+    pColList->cols[index].valStat = VAL_STAT_HAS;
+    pColList->boundColumns[pColList->numOfBound] = index + PRIMARYKEY_TIMESTAMP_COL_ID;
+    ++pColList->numOfBound;
+    switch (pSchema[t].type) {
+      case TSDB_DATA_TYPE_BINARY:
+        pColList->boundNullLen += (sizeof(VarDataOffsetT) + VARSTR_HEADER_SIZE + CHAR_BYTES);
+        break;
+      case TSDB_DATA_TYPE_NCHAR:
+        pColList->boundNullLen += (sizeof(VarDataOffsetT) + VARSTR_HEADER_SIZE + TSDB_NCHAR_SIZE);
+        break;
+      default:
+        pColList->boundNullLen += TYPE_BYTES[pSchema[t].type];
+        break;
+    }
+  }
+
+  pColList->orderStatus = isOrdered ? ORDER_STATUS_ORDERED : ORDER_STATUS_DISORDERED;
+
+  if (!isOrdered) {
+    pColList->colIdxInfo = taosMemoryCalloc(pColList->numOfBound, sizeof(SBoundIdxInfo));
+    if (NULL == pColList->colIdxInfo) {
+      return TSDB_CODE_TSC_OUT_OF_MEMORY;
+    }
+    SBoundIdxInfo* pColIdx = pColList->colIdxInfo;
+    for (col_id_t i = 0; i < pColList->numOfBound; ++i) {
+      pColIdx[i].schemaColIdx = pColList->boundColumns[i];
+      pColIdx[i].boundIdx = i;
+    }
+    qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), schemaIdxCompar);
+    for (col_id_t i = 0; i < pColList->numOfBound; ++i) {
+      pColIdx[i].finalIdx = i;
+    }
+    qsort(pColIdx, pColList->numOfBound, sizeof(SBoundIdxInfo), boundIdxCompar);
+  }
+
+  if(pColList->numOfCols > pColList->numOfBound){
+    memset(&pColList->boundColumns[pColList->numOfBound], 0,
+           sizeof(col_id_t) * (pColList->numOfCols - pColList->numOfBound));
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+static int32_t smlBoundTags(SArray *cols, SKVRowBuilder *tagsBuilder, SParsedDataColInfo* tags, SSchema* pSchema, SKVRow *row, SMsgBuf *msg) {
+  if (tdInitKVRowBuilder(tagsBuilder) < 0) {
+    return TSDB_CODE_TSC_OUT_OF_MEMORY;
+  }
+
+  SKvParam param = {.builder = tagsBuilder};
+  for (int i = 0; i < tags->numOfBound; ++i) {
+    SSchema* pTagSchema = &pSchema[tags->boundColumns[i] - 1]; // colId starts with 1
+    param.schema = pTagSchema;
+    SSmlKv *kv = taosArrayGetP(cols, i);
+    KvRowAppend(msg, kv->value, kv->valueLen, &param) ;
+  }
+
+
+  *row = tdGetKVRowFromBuilder(tagsBuilder);
+  if(*row == NULL){
+    return TSDB_CODE_SML_INVALID_DATA;
+  }
+  tdSortKVRowByColIdx(*row);
+  return TSDB_CODE_SUCCESS;
+}
+
+int32_t smlBindData(void *handle, SArray *tags, SArray *colsFormat, SHashObj *colsHash, SArray *cols, bool format,
+                    STableMeta *pTableMeta, char *tableName, char *msgBuf, int16_t msgBufLen) {
+  SMsgBuf pBuf = {.buf = msgBuf, .len = msgBufLen};
+
+  SSmlExecHandle *smlHandle = (SSmlExecHandle *)handle;
+  SSchema* pTagsSchema = getTableTagSchema(pTableMeta);
+  setBoundColumnInfo(&smlHandle->tags, pTagsSchema, getNumOfTags(pTableMeta));
+  int ret = smlBoundColumns(tags, &smlHandle->tags, pTagsSchema);
+  if(ret != TSDB_CODE_SUCCESS){
+    buildInvalidOperationMsg(&pBuf, "bound tags error");
+    return ret;
+  }
+  SKVRow row = NULL;
+  ret = smlBoundTags(tags, &smlHandle->tagsBuilder, &smlHandle->tags, pTagsSchema, &row, &pBuf);
+  if(ret != TSDB_CODE_SUCCESS){
+    return ret;
+  }
+
+  buildCreateTbReq(&smlHandle->createTblReq, tableName, row, pTableMeta->suid);
+
+  STableDataBlocks* pDataBlock = NULL;
+  ret = getDataBlockFromList(smlHandle->pBlockHash, pTableMeta->uid, TSDB_DEFAULT_PAYLOAD_SIZE,
+                                  sizeof(SSubmitBlk), getTableInfo(pTableMeta).rowSize, pTableMeta,
+                                  &pDataBlock, NULL, &smlHandle->createTblReq);
+  if(ret != TSDB_CODE_SUCCESS){
+    buildInvalidOperationMsg(&pBuf, "create data block error");
+    return ret;
+  }
+
+  SSchema* pSchema = getTableColumnSchema(pTableMeta);
+
+
+  if(format){
+    ret = smlBoundColumns(taosArrayGetP(colsFormat, 0), &pDataBlock->boundColumnInfo, pSchema);
+  }else{
+    SArray *columns = taosArrayInit(16, POINTER_BYTES);
+    void **p1 = taosHashIterate(colsHash, NULL);
+    while (p1) {
+      SSmlKv* kv = *p1;
+      taosArrayPush(columns, &kv);
+      p1 = taosHashIterate(colsHash, p1);
+    }
+    ret = smlBoundColumns(columns, &pDataBlock->boundColumnInfo, pSchema);
+    taosArrayDestroy(columns);
+  }
+
+  if(ret != TSDB_CODE_SUCCESS){
+    buildInvalidOperationMsg(&pBuf, "bound cols error");
+    return ret;
+  }
+  int32_t extendedRowSize = getExtendedRowSize(pDataBlock);
+  SParsedDataColInfo* spd = &pDataBlock->boundColumnInfo;
+  SRowBuilder*        pBuilder = &pDataBlock->rowBuilder;
+  SMemParam param = {.rb = pBuilder};
+
+  initRowBuilder(&pDataBlock->rowBuilder, pDataBlock->pTableMeta->sversion, &pDataBlock->boundColumnInfo);
+
+  int32_t rowNum = format ? taosArrayGetSize(colsFormat) : taosArrayGetSize(cols);
+  if(rowNum <= 0) {
+    return buildInvalidOperationMsg(&pBuf, "cols size <= 0");
+  }
+  ret = allocateMemForSize(pDataBlock, extendedRowSize * rowNum);
+  if(ret != TSDB_CODE_SUCCESS){
+    buildInvalidOperationMsg(&pBuf, "allocate memory error");
+    return ret;
+  }
+  for (int32_t r = 0; r < rowNum; ++r) {
+    STSRow* row = (STSRow*)(pDataBlock->pData + pDataBlock->size);  // skip the SSubmitBlk header
+    tdSRowResetBuf(pBuilder, row);
+    void *rowData = NULL;
+    if(format){
+      rowData = taosArrayGetP(colsFormat, r);
+    }else{
+      rowData = taosArrayGetP(cols, r);
+    }
+
+    // 1. set the parsed value from sql string
+    for (int c = 0; c < spd->numOfBound; ++c) {
+      SSchema* pColSchema = &pSchema[spd->boundColumns[c] - 1];
+
+      param.schema = pColSchema;
+      getSTSRowAppendInfo(pBuilder->rowType, spd, c, &param.toffset, &param.colIdx);
+
+      SSmlKv *kv = NULL;
+      if(format){
+        kv = taosArrayGetP(rowData, c);
+        if (!kv){
+          char msg[64] = {0};
+          sprintf(msg, "cols num not the same like before:%d", r);
+          return buildInvalidOperationMsg(&pBuf, msg);
+        }
+      }else{
+        void **p =taosHashGet(rowData, pColSchema->name, strlen(pColSchema->name));
+        kv = *p;
+      }
+
+      if (kv->length == 0) {
+        MemRowAppend(&pBuf, NULL, 0, &param);
+      } else {
+        int32_t colLen = pColSchema->bytes;
+        if (IS_VAR_DATA_TYPE(pColSchema->type)) {
+          colLen = kv->length;
+        }
+
+        MemRowAppend(&pBuf, &(kv->value), colLen, &param);
+      }
+
+      if (PRIMARYKEY_TIMESTAMP_COL_ID == pColSchema->colId) {
+        TSKEY tsKey = TD_ROW_KEY(row);
+        checkTimestamp(pDataBlock, (const char *)&tsKey);
+      }
+    }
+
+    // set the null value for the columns that do not assign values
+    if ((spd->numOfBound < spd->numOfCols) && TD_IS_TP_ROW(row)) {
+      for (int32_t i = 0; i < spd->numOfCols; ++i) {
+        if (spd->cols[i].valStat == VAL_STAT_NONE) {  // the primary TS key is not VAL_STAT_NONE
+          tdAppendColValToTpRow(pBuilder, TD_VTYPE_NONE, getNullValue(pSchema[i].type), true, pSchema[i].type, i,
+                                spd->cols[i].toffset);
+        }
+      }
+    }
+
+    pDataBlock->size += extendedRowSize;
+  }
+
+  SSubmitBlk *pBlocks = (SSubmitBlk *)(pDataBlock->pData);
+  if (TSDB_CODE_SUCCESS != setBlockInfo(pBlocks, pDataBlock, rowNum)) {
+    return buildInvalidOperationMsg(&pBuf, "too many rows in sql, total number of rows should be less than 32767");
+  }
+
+  return TSDB_CODE_SUCCESS;
+}
+
+void* smlInitHandle(SQuery *pQuery){
+  SSmlExecHandle *handle = taosMemoryCalloc(1, sizeof(SSmlExecHandle));
+  if(!handle) return NULL;
+  handle->pBlockHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, false);
+  handle->pQuery = pQuery;
+
+  return handle;
+}
+
+void smlDestroyHandle(void *pHandle){
+  if(!pHandle) return;
+  SSmlExecHandle *handle = (SSmlExecHandle *)pHandle;
+  destroyBlockHashmap(handle->pBlockHash);
+  taosMemoryFree(handle);
+}
+
+int32_t smlBuildOutput(void* handle, SHashObj* pVgHash) {
+  SSmlExecHandle *smlHandle = (SSmlExecHandle *)handle;
+  return qBuildStmtOutput(smlHandle->pQuery, pVgHash, smlHandle->pBlockHash);
+}
+// schemaless logic end
+

source/libs/parser/src/parInsertData.c

@@ -115,7 +115,7 @@ void destroyBoundColumnInfo(void* pBoundInfo) {
   taosMemoryFreeClear(pColList->colIdxInfo);
 }
 
-static int32_t createDataBlock(size_t defaultSize, int32_t rowSize, int32_t startOffset, const STableMeta* pTableMeta,
+static int32_t createDataBlock(size_t defaultSize, int32_t rowSize, int32_t startOffset, STableMeta* pTableMeta,
                                STableDataBlocks** dataBlocks) {
   STableDataBlocks* dataBuf = (STableDataBlocks*)taosMemoryCalloc(1, sizeof(STableDataBlocks));
   if (dataBuf == NULL) {
@@ -137,8 +137,7 @@ static int32_t createDataBlock(size_t defaultSize, int32_t rowSize, int32_t star
   }
   memset(dataBuf->pData, 0, sizeof(SSubmitBlk));
 
-  // Here we keep the tableMeta to avoid it to be remove by other threads.
-  dataBuf->pTableMeta = tableMetaDup(pTableMeta);
+  dataBuf->pTableMeta = pTableMeta;
 
   SParsedDataColInfo* pColInfo = &dataBuf->boundColumnInfo;
   SSchema*            pSchema = getTableColumnSchema(dataBuf->pTableMeta);
@@ -187,7 +186,7 @@ int32_t buildCreateTbMsg(STableDataBlocks* pBlocks, SVCreateTbReq* pCreateTbReq)
 }
 
 int32_t getDataBlockFromList(SHashObj* pHashList, int64_t id, int32_t size, int32_t startOffset, int32_t rowSize,
-                             const STableMeta* pTableMeta, STableDataBlocks** dataBlocks, SArray* pBlockList,
+                             STableMeta* pTableMeta, STableDataBlocks** dataBlocks, SArray* pBlockList,
                              SVCreateTbReq* pCreateTbReq) {
   *dataBlocks = NULL;
   STableDataBlocks** t1 = (STableDataBlocks**)taosHashGet(pHashList, (const char*)&id, sizeof(id));
@@ -238,9 +237,9 @@ static void destroyDataBlock(STableDataBlocks* pDataBlock) {
   taosMemoryFreeClear(pDataBlock->pData);
   if (!pDataBlock->cloned) {
     // free the refcount for metermeta
-    if (pDataBlock->pTableMeta != NULL) {
-      taosMemoryFreeClear(pDataBlock->pTableMeta);
-    }
+//    if (pDataBlock->pTableMeta != NULL) {
+//      taosMemoryFreeClear(pDataBlock->pTableMeta);
+//    }
 
     destroyBoundColumnInfo(&pDataBlock->boundColumnInfo);
   }
@@ -456,6 +455,7 @@ int32_t mergeTableDataBlocks(SHashObj* pHashObj, uint8_t payloadType, SArray** p
     SSubmitBlk* pBlocks = (SSubmitBlk*)pOneTableBlock->pData;
     if (pBlocks->numOfRows > 0) {
       STableDataBlocks* dataBuf = NULL;
+      pOneTableBlock->pTableMeta->vgId = pOneTableBlock->vgId;    // for schemaless, restore origin vgId
       int32_t           ret =
           getDataBlockFromList(pVnodeDataBlockHashList, pOneTableBlock->vgId, TSDB_PAYLOAD_SIZE, INSERT_HEAD_SIZE, 0,
                                pOneTableBlock->pTableMeta, &dataBuf, pVnodeDataBlockList, NULL);

source/libs/parser/src/parUtil.c

@@ -191,27 +191,7 @@ int32_t buildSyntaxErrMsg(SMsgBuf* pBuf, const char* additionalInfo, const char*
   return TSDB_CODE_TSC_SQL_SYNTAX_ERROR;
 }
 
-static uint32_t getTableMetaSize(const STableMeta* pTableMeta) {
-  assert(pTableMeta != NULL);
-
-  int32_t totalCols = 0;
-  if (pTableMeta->tableInfo.numOfColumns >= 0) {
-    totalCols = pTableMeta->tableInfo.numOfColumns + pTableMeta->tableInfo.numOfTags;
-  }
-
-  return sizeof(STableMeta) + totalCols * sizeof(SSchema);
-}
-
-STableMeta* tableMetaDup(const STableMeta* pTableMeta) {
-  assert(pTableMeta != NULL);
-  size_t size = getTableMetaSize(pTableMeta);
-
-  STableMeta* p = taosMemoryMalloc(size);
-  memcpy(p, pTableMeta, size);
-  return p;
-}
-
-SSchema* getTableColumnSchema(const STableMeta* pTableMeta) {
+SSchema *getTableColumnSchema(const STableMeta *pTableMeta) {
   assert(pTableMeta != NULL);
   return (SSchema*)pTableMeta->schema;
 }

source/libs/scalar/test/scalar/scalarTests.cpp

@@ -166,7 +166,6 @@ void scltMakeColumnNode(SNode **pNode, SSDataBlock **block, int32_t dataType, in
     idata.pData = (char *)taosMemoryCalloc(1, size);
     colInfoDataEnsureCapacity(&idata, 0, rowNum);
     taosArrayPush(res->pDataBlock, &idata);
-    
     SColumnInfoData *pColumn = (SColumnInfoData *)taosArrayGetLast(res->pDataBlock);
     for (int32_t i = 0; i < rowNum; ++i) {
       colDataAppend(pColumn, i, (const char *)value, false);

source/libs/wal/src/walRead.c

@@ -49,7 +49,10 @@ void walCloseReadHandle(SWalReadHandle *pRead) {
   taosMemoryFree(pRead);
 }
 
-int32_t walRegisterRead(SWalReadHandle *pRead, int64_t ver) { return 0; }
+int32_t walRegisterRead(SWalReadHandle *pRead, int64_t ver) {
+  // TODO
+  return 0;
+}
 
 static int32_t walReadSeekFilePos(SWalReadHandle *pRead, int64_t fileFirstVer, int64_t ver) {
   int code = 0;

source/util/src/terror.c

@@ -449,6 +449,11 @@ TAOS_DEFINE_ERROR(TSDB_CODE_PAR_TABLE_NOT_EXIST,          "Table does not exist"
 //planner
 TAOS_DEFINE_ERROR(TSDB_CODE_PLAN_INTERNAL_ERROR,          "planner internal error")
 
+//schemaless
+TAOS_DEFINE_ERROR(TSDB_CODE_SML_INVALID_PROTOCOL_TYPE,    "Invalid line protocol type")
+TAOS_DEFINE_ERROR(TSDB_CODE_SML_INVALID_PRECISION_TYPE,   "Invalid timestamp precision type")
+TAOS_DEFINE_ERROR(TSDB_CODE_SML_INVALID_DATA,             "Invalid data type")
+
 #ifdef TAOS_ERROR_C
 };
 #endif

tools/shell/src/shellCommand.c

@@ -151,8 +151,8 @@ void shellGetPrevCharSize(const char *str, int32_t pos, int32_t *size, int32_t *
     if (str[pos] > 0 || shellCountPrefixOnes((uint8_t)str[pos]) > 1) break;
   }
 
-  int32_t rc = taosMbToWchar(&wc, str + pos, MB_CUR_MAX);
-  assert(rc == *size);
+  taosMbToWchar(&wc, str + pos, MB_CUR_MAX);
+  // assert(rc == *size); // it will be core, if str is encode by utf8 and taos charset is gbk
 
   *width = taosWcharWidth(wc);
 }

tools/shell/src/shellEngine.c

@@ -118,66 +118,27 @@ int32_t shellRunCommand(char *command) {
     }
   }
 
-  bool esc = false;
-  char quote = 0, *cmd = command, *p = command;
+  char quote = 0, *cmd = command;
   for (char c = *command++; c != 0; c = *command++) {
-    if (esc) {
-      switch (c) {
-        case 'n':
-          c = '\n';
-          break;
-        case 'r':
-          c = '\r';
-          break;
-        case 't':
-          c = '\t';
-          break;
-        case 'G':
-          *p++ = '\\';
-          break;
-        case '\'':
-        case '"':
-          if (quote) {
-            *p++ = '\\';
-          }
-          break;
-        default:
-          *p++ = '\\';
-          break;
-      }
-      *p++ = c;
-      esc = false;
+    if (c == '\\' && (*command == '\'' || *command == '"' || *command == '`')) {
+      command ++;
       continue;
     }
 
-    if (c == '\\') {
-      if (quote != 0 && (*command == '_' || *command == '\\')) {
-        // DO nothing
-      } else {
-        esc = true;
-        continue;
-      }
-    }
-
     if (quote == c) {
       quote = 0;
-    } else if (quote == 0 && (c == '\'' || c == '"')) {
+    } else if (quote == 0 && (c == '\'' || c == '"' || c == '`')) {
       quote = c;
-    }
-
-    *p++ = c;
-    if (c == ';' && quote == 0) {
-      c = *p;
-      *p = 0;
+    } else if (c == ';' && quote == 0) {
+      c = *command;
+      *command = 0;
       if (shellRunSingleCommand(cmd) < 0) {
         return -1;
       }
-      *p = c;
-      p = cmd;
+      *command = c;
+      cmd = command;
     }
   }
-
-  *p = 0;
   return shellRunSingleCommand(cmd);
 }
 

taos-tools @ 2f3dfddd

-Subproject commit 59e0ebaf4905e4cb6d95a01c58b3fa507abc5a20
+Subproject commit 2f3dfddd4d9a869e706ba3cf98fb6d769404cd7c