From f7c21c45023100a7556a0dd152229c073b1292b9 Mon Sep 17 00:00:00 2001
From: wangmm0220 <wangmm0220@gmail.com>
Date: Mon, 30 May 2022 18:00:03 +0800
Subject: [PATCH] feat:add schemaless db paras

---
 include/common/tmsg.h                   |  2 ++
 include/libs/parser/parser.h            |  1 +
 include/util/taoserror.h                |  1 +
 source/client/inc/clientInt.h           |  1 +
 source/client/src/clientEnv.c           |  1 +
 source/client/src/clientImpl.c          |  1 +
 source/client/src/clientSml.c           | 33 +++++++++++++++++++++++++
 source/common/src/tmsg.c                |  2 ++
 source/dnode/mnode/impl/inc/mndDef.h    |  1 +
 source/dnode/mnode/impl/src/mndDb.c     |  7 ++++++
 source/dnode/mnode/impl/src/mndVgroup.c |  1 +
 source/libs/parser/inc/parInt.h         |  1 +
 source/libs/parser/src/parInsert.c      | 24 +++++++++++++++++-
 source/libs/parser/src/parTranslater.c  | 11 +++++++++
 source/util/src/terror.c                |  1 +
 tests/test/c/sdbDump.c                  |  1 +
 16 files changed, 88 insertions(+), 1 deletion(-)

diff --git a/include/common/tmsg.h b/include/common/tmsg.h
index faf4addb4b..68480e4ffd 100644
--- a/include/common/tmsg.h
+++ b/include/common/tmsg.h
@@ -779,6 +779,7 @@ typedef struct {
   int8_t  cacheLastRow;
   int32_t numOfRetensions;
   SArray* pRetensions;
+  int8_t  schemaless;
 } SDbCfgRsp;
 
 int32_t tSerializeSDbCfgRsp(void* buf, int32_t bufLen, const SDbCfgRsp* pRsp);
@@ -1009,6 +1010,7 @@ typedef struct {
   SReplica replicas[TSDB_MAX_REPLICA];
   int32_t  numOfRetensions;
   SArray*  pRetensions;  // SRetention
+  int8_t   schemaless;
 
   // for tsma
   int8_t isTsma;
diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h
index 1466bb7400..774a23ff4b 100644
--- a/include/libs/parser/parser.h
+++ b/include/libs/parser/parser.h
@@ -49,6 +49,7 @@ typedef struct SParseContext {
   const char*      pUser;
   bool             isSuperUser;
   bool             async;
+  int8_t           schemalessType;
 } SParseContext;
 
 int32_t qParseSql(SParseContext* pCxt, SQuery** pQuery);
diff --git a/include/util/taoserror.h b/include/util/taoserror.h
index 65cfe8de0b..b9eae6962c 100644
--- a/include/util/taoserror.h
+++ b/include/util/taoserror.h
@@ -673,6 +673,7 @@ int32_t* taosGetErrno();
 #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)
+#define TSDB_CODE_SML_INVALID_DB_CONF           TAOS_DEF_ERROR_CODE(0, 0x3003)
 
 #ifdef __cplusplus
 }
diff --git a/source/client/inc/clientInt.h b/source/client/inc/clientInt.h
index d9f3351008..4c5bac71e4 100644
--- a/source/client/inc/clientInt.h
+++ b/source/client/inc/clientInt.h
@@ -149,6 +149,7 @@ typedef struct STscObj {
   int32_t       numOfReqs;  // number of sqlObj bound to this connection
   SAppInstInfo* pAppInfo;
   SHashObj*     pRequests;
+  int8_t        schemalessType;
 } STscObj;
 
 typedef struct SResultColumn {
diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c
index 669b2bc97e..35e354e952 100644
--- a/source/client/src/clientEnv.c
+++ b/source/client/src/clientEnv.c
@@ -161,6 +161,7 @@ void *createTscObj(const char *user, const char *auth, const char *db, int32_t c
 
   taosThreadMutexInit(&pObj->mutex, NULL);
   pObj->id = taosAddRef(clientConnRefPool, pObj);
+  pObj->schemalessType = 0;
 
   tscDebug("connObj created, 0x%" PRIx64, pObj->id);
   return pObj;
diff --git a/source/client/src/clientImpl.c b/source/client/src/clientImpl.c
index eb4c4cb59f..ba25f2f4ca 100644
--- a/source/client/src/clientImpl.c
+++ b/source/client/src/clientImpl.c
@@ -175,6 +175,7 @@ int32_t parseSql(SRequestObj* pRequest, bool topicQuery, SQuery** pQuery, SStmtC
                        .pTransporter = pTscObj->pAppInfo->pTransporter,
                        .pStmtCb = pStmtCb,
                        .pUser = pTscObj->user,
+                       .schemalessType = pTscObj->schemalessType,
                        .isSuperUser = (0 == strcmp(pTscObj->user, TSDB_DEFAULT_USER))};
 
   cxt.mgmtEpSet = getEpSet_s(&pTscObj->pAppInfo->mgmtEp);
diff --git a/source/client/src/clientSml.c b/source/client/src/clientSml.c
index 7d623072d6..29c7608126 100644
--- a/source/client/src/clientSml.c
+++ b/source/client/src/clientSml.c
@@ -2318,6 +2318,28 @@ cleanup:
   return code;
 }
 
+static int32_t isSchemalessDb(SSmlHandle* info){
+  SName          name;
+  tNameSetDbName(&name, info->taos->acctId, info->taos->db, strlen(info->taos->db));
+  char dbFname[TSDB_DB_FNAME_LEN] = {0};
+  tNameGetFullDbName(&name, dbFname);
+  SDbCfgInfo pInfo = {0};
+  SEpSet ep = getEpSet_s(&info->taos->pAppInfo->mgmtEp);
+
+  int32_t code = catalogGetDBCfg(info->pCatalog, info->taos->pAppInfo->pTransporter, &ep, dbFname, &pInfo);
+  if (code != TSDB_CODE_SUCCESS) {
+    info->pRequest->code = code;
+    smlBuildInvalidDataMsg(&info->msgBuf, "catalogGetDBCfg error, code:", tstrerror(code));
+    return code;
+  }
+  if (!pInfo.schemaless){
+    info->pRequest->code = TSDB_CODE_SML_INVALID_DB_CONF;
+    smlBuildInvalidDataMsg(&info->msgBuf, "can not insert into schemaless db:", dbFname);
+    return TSDB_CODE_SML_INVALID_DB_CONF;
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 /**
  * taos_schemaless_insert() parse and insert data points into database according to
  * different protocol.
@@ -2351,6 +2373,17 @@ TAOS_RES* taos_schemaless_insert(TAOS* taos, char* lines[], int numLines, int pr
     return (TAOS_RES*)request;
   }
 
+  info->taos->schemalessType = 1;
+  if(request->pDb == NULL){
+    request->code = TSDB_CODE_PAR_DB_NOT_SPECIFIED;
+    smlBuildInvalidDataMsg(&info->msgBuf, "Database not specified", NULL);
+    goto end;
+  }
+
+  if(isSchemalessDb(info) != TSDB_CODE_SUCCESS){
+    goto end;
+  }
+
   if (!lines) {
     request->code = TSDB_CODE_SML_INVALID_DATA;
     smlBuildInvalidDataMsg(&info->msgBuf, "lines is null", NULL);
diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c
index 7615f7b070..b001953ee9 100644
--- a/source/common/src/tmsg.c
+++ b/source/common/src/tmsg.c
@@ -2226,6 +2226,7 @@ int32_t tSerializeSDbCfgRsp(void *buf, int32_t bufLen, const SDbCfgRsp *pRsp) {
     if (tEncodeI8(&encoder, pRetension->freqUnit) < 0) return -1;
     if (tEncodeI8(&encoder, pRetension->keepUnit) < 0) return -1;
   }
+  if (tEncodeI8(&encoder, pRsp->schemaless) < 0) return -1;
   tEndEncode(&encoder);
 
   int32_t tlen = encoder.pos;
@@ -2274,6 +2275,7 @@ int32_t tDeserializeSDbCfgRsp(void *buf, int32_t bufLen, SDbCfgRsp *pRsp) {
       return -1;
     }
   }
+  if (tDecodeI8(&decoder, &pRsp->schemaless) < 0) return -1;
   tEndDecode(&decoder);
 
   tDecoderClear(&decoder);
diff --git a/source/dnode/mnode/impl/inc/mndDef.h b/source/dnode/mnode/impl/inc/mndDef.h
index 9e036d7b25..6d839ea622 100644
--- a/source/dnode/mnode/impl/inc/mndDef.h
+++ b/source/dnode/mnode/impl/inc/mndDef.h
@@ -302,6 +302,7 @@ typedef struct {
   int8_t  hashMethod;  // default is 1
   int32_t numOfRetensions;
   SArray* pRetensions;
+  int8_t  schemaless;
 } SDbCfg;
 
 typedef struct {
diff --git a/source/dnode/mnode/impl/src/mndDb.c b/source/dnode/mnode/impl/src/mndDb.c
index 95d3383ee1..cc884fb18b 100644
--- a/source/dnode/mnode/impl/src/mndDb.c
+++ b/source/dnode/mnode/impl/src/mndDb.c
@@ -115,6 +115,7 @@ static SSdbRaw *mndDbActionEncode(SDbObj *pDb) {
     SDB_SET_INT8(pRaw, dataPos, pRetension->freqUnit, _OVER)
     SDB_SET_INT8(pRaw, dataPos, pRetension->keepUnit, _OVER)
   }
+  SDB_SET_INT8(pRaw, dataPos, pDb->cfg.schemaless, _OVER)
 
   SDB_SET_RESERVE(pRaw, dataPos, DB_RESERVE_SIZE, _OVER)
   SDB_SET_DATALEN(pRaw, dataPos, _OVER)
@@ -192,6 +193,7 @@ static SSdbRow *mndDbActionDecode(SSdbRaw *pRaw) {
       }
     }
   }
+  SDB_GET_INT8(pRaw, dataPos, &pDb->cfg.schemaless, _OVER)
 
   SDB_GET_RESERVE(pRaw, dataPos, DB_RESERVE_SIZE, _OVER)
   taosInitRWLatch(&pDb->lock);
@@ -380,6 +382,7 @@ static int32_t mndCheckDbCfg(SMnode *pMnode, SDbCfg *pCfg) {
   if (pCfg->replications < TSDB_MIN_DB_REPLICA || pCfg->replications > TSDB_MAX_DB_REPLICA) 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->schemaless < TSDB_DB_SCHEMALESS_OFF || pCfg->schemaless > TSDB_DB_SCHEMALESS_ON) 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;
   if (pCfg->replications > mndGetDnodeSize(pMnode)) {
@@ -411,6 +414,8 @@ static void mndSetDefaultDbCfg(SDbCfg *pCfg) {
   if (pCfg->strict < 0) pCfg->strict = TSDB_DEFAULT_DB_STRICT;
   if (pCfg->cacheLastRow < 0) pCfg->cacheLastRow = TSDB_DEFAULT_CACHE_LAST_ROW;
   if (pCfg->numOfRetensions < 0) pCfg->numOfRetensions = 0;
+  if (pCfg->schemaless < 0) pCfg->schemaless = TSDB_DB_SCHEMALESS_OFF;
+
 }
 
 static int32_t mndSetCreateDbRedoLogs(SMnode *pMnode, STrans *pTrans, SDbObj *pDb, SVgObj *pVgroups) {
@@ -521,6 +526,7 @@ static int32_t mndCreateDb(SMnode *pMnode, SRpcMsg *pReq, SCreateDbReq *pCreate,
       .strict = pCreate->strict,
       .cacheLastRow = pCreate->cacheLastRow,
       .hashMethod = 1,
+      .schemaless = pCreate->schemaless,
   };
 
   dbObj.cfg.numOfRetensions = pCreate->numOfRetensions;
@@ -890,6 +896,7 @@ static int32_t mndProcessGetDbCfgReq(SRpcMsg *pReq) {
   cfgRsp.cacheLastRow = pDb->cfg.cacheLastRow;
   cfgRsp.numOfRetensions = pDb->cfg.numOfRetensions;
   cfgRsp.pRetensions = pDb->cfg.pRetensions;
+  cfgRsp.schemaless = pDb->cfg.schemaless;
 
   int32_t contLen = tSerializeSDbCfgRsp(NULL, 0, &cfgRsp);
   void   *pRsp = rpcMallocCont(contLen);
diff --git a/source/dnode/mnode/impl/src/mndVgroup.c b/source/dnode/mnode/impl/src/mndVgroup.c
index e05b38a7c0..a00841f911 100644
--- a/source/dnode/mnode/impl/src/mndVgroup.c
+++ b/source/dnode/mnode/impl/src/mndVgroup.c
@@ -218,6 +218,7 @@ void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVg
   createReq.hashMethod = pDb->cfg.hashMethod;
   createReq.numOfRetensions = pDb->cfg.numOfRetensions;
   createReq.pRetensions = pDb->cfg.pRetensions;
+  createReq.schemaless = pDb->cfg.schemaless;
 
   for (int32_t v = 0; v < pVgroup->replica; ++v) {
     SReplica  *pReplica = &createReq.replicas[v];
diff --git a/source/libs/parser/inc/parInt.h b/source/libs/parser/inc/parInt.h
index 184ed7d8b2..c74567f1a0 100644
--- a/source/libs/parser/inc/parInt.h
+++ b/source/libs/parser/inc/parInt.h
@@ -31,6 +31,7 @@ int32_t authenticate(SParseContext* pParseCxt, SQuery* pQuery);
 int32_t translate(SParseContext* pParseCxt, SQuery* pQuery);
 int32_t extractResultSchema(const SNode* pRoot, int32_t* numOfCols, SSchema** pSchema);
 int32_t calculateConstant(SParseContext* pParseCxt, SQuery* pQuery);
+int32_t isNotSchemalessDb(SParseContext* pContext);
 
 #ifdef __cplusplus
 }
diff --git a/source/libs/parser/src/parInsert.c b/source/libs/parser/src/parInsert.c
index 58a6d1483f..c78733c3c4 100644
--- a/source/libs/parser/src/parInsert.c
+++ b/source/libs/parser/src/parInsert.c
@@ -1204,6 +1204,23 @@ static int32_t parseInsertBody(SInsertParseContext* pCxt) {
   return buildOutput(pCxt);
 }
 
+int32_t isNotSchemalessDb(SParseContext* pContext){
+  SName          name;
+  tNameSetDbName(&name, pContext->acctId, pContext->db, strlen(pContext->db));
+  char dbFname[TSDB_DB_FNAME_LEN] = {0};
+  tNameGetFullDbName(&name, dbFname);
+  SDbCfgInfo pInfo = {0};
+  int32_t code = catalogGetDBCfg(pContext->pCatalog, pContext->pTransporter, &pContext->mgmtEpSet, dbFname, &pInfo);
+  if (code != TSDB_CODE_SUCCESS) {
+    parserError("catalogGetDBCfg error, code:%s, dbFName:%s", tstrerror(code), dbFname);
+    return code;
+  }
+  if (pInfo.schemaless){
+    parserError("can not insert into schemaless db:%s", dbFname);
+    return TSDB_CODE_SML_INVALID_DB_CONF;
+  }
+  return TSDB_CODE_SUCCESS;
+}
 // INSERT INTO
 //   tb_name
 //       [USING stb_name [(tag1_name, ...)] TAGS (tag1_value, ...)]
@@ -1252,6 +1269,11 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
     (*pQuery)->pRoot = (SNode*)context.pOutput;
   }
 
+  int32_t code = isNotSchemalessDb(pContext);
+  if(code != TSDB_CODE_SUCCESS){
+    return code;
+  }
+
   if (NULL == (*pQuery)->pTableList) {
     (*pQuery)->pTableList = taosArrayInit(taosHashGetSize(context.pTableNameHashObj), sizeof(SName));
     if (NULL == (*pQuery)->pTableList) {
@@ -1261,7 +1283,7 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) {
 
   context.pOutput->payloadType = PAYLOAD_TYPE_KV;
 
-  int32_t code = skipInsertInto(&context);
+  code = skipInsertInto(&context);
   if (TSDB_CODE_SUCCESS == code) {
     code = parseInsertBody(&context);
   }
diff --git a/source/libs/parser/src/parTranslater.c b/source/libs/parser/src/parTranslater.c
index d84b005f7f..31363401fe 100644
--- a/source/libs/parser/src/parTranslater.c
+++ b/source/libs/parser/src/parTranslater.c
@@ -2619,6 +2619,11 @@ static int32_t checkCreateTable(STranslateContext* pCxt, SCreateTableStmt* pStmt
   if (TSDB_CODE_SUCCESS == code) {
     code = checkTableSchema(pCxt, pStmt);
   }
+  if (TSDB_CODE_SUCCESS == code) {
+    if(pCxt->pParseCxt->schemalessType == 0 && isNotSchemalessDb(pCxt->pParseCxt) != TSDB_CODE_SUCCESS){
+      code = TSDB_CODE_SML_INVALID_DB_CONF;
+    }
+  }
   return code;
 }
 
@@ -4350,6 +4355,9 @@ static SArray* serializeVgroupsCreateTableBatch(int32_t acctId, SHashObj* pVgrou
 }
 
 static int32_t rewriteCreateMultiTable(STranslateContext* pCxt, SQuery* pQuery) {
+  if(pCxt->pParseCxt->schemalessType == 0 && isNotSchemalessDb(pCxt->pParseCxt) != TSDB_CODE_SUCCESS){
+    return TSDB_CODE_SML_INVALID_DB_CONF;
+  }
   SCreateMultiTableStmt* pStmt = (SCreateMultiTableStmt*)pQuery->pRoot;
 
   SHashObj* pVgroupHashmap = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), false, HASH_NO_LOCK);
@@ -4766,6 +4774,9 @@ static int32_t buildModifyVnodeArray(STranslateContext* pCxt, SAlterTableStmt* p
 }
 
 static int32_t rewriteAlterTable(STranslateContext* pCxt, SQuery* pQuery) {
+  if(pCxt->pParseCxt->schemalessType == 0 && isNotSchemalessDb(pCxt->pParseCxt) != TSDB_CODE_SUCCESS){
+    return TSDB_CODE_SML_INVALID_DB_CONF;
+  }
   SAlterTableStmt* pStmt = (SAlterTableStmt*)pQuery->pRoot;
 
   STableMeta* pTableMeta = NULL;
diff --git a/source/util/src/terror.c b/source/util/src/terror.c
index 178d6e8d2b..0ffcebca5b 100644
--- a/source/util/src/terror.c
+++ b/source/util/src/terror.c
@@ -469,6 +469,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_UDF_INVALID_OUTPUT_TYPE,      "udf invalid output ty
 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")
+TAOS_DEFINE_ERROR(TSDB_CODE_SML_INVALID_DB_CONF,          "Invalid schemaless db config")
 
 #ifdef TAOS_ERROR_C
 };
diff --git a/tests/test/c/sdbDump.c b/tests/test/c/sdbDump.c
index 8be2822c0a..6a7255b33c 100644
--- a/tests/test/c/sdbDump.c
+++ b/tests/test/c/sdbDump.c
@@ -86,6 +86,7 @@ void dumpDb(SSdb *pSdb, SJson *json) {
     tjsonAddIntegerToObject(item, "cacheLastRow", pObj->cfg.cacheLastRow);
     tjsonAddIntegerToObject(item, "hashMethod", pObj->cfg.hashMethod);
     tjsonAddIntegerToObject(item, "numOfRetensions", pObj->cfg.numOfRetensions);
+    tjsonAddIntegerToObject(item, "schemaless", pObj->cfg.schemaless);
 
     sdbRelease(pSdb, pObj);
   }