diff --git a/include/common/tmsg.h b/include/common/tmsg.h index 5cf9e9855def12dc355e809e796495f9ce063a16..aa57607d6f9cf582d3269ce31a53a6cbd5b4171a 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); diff --git a/include/libs/parser/parser.h b/include/libs/parser/parser.h index ca825b9e2fb460b6aa35110c89535071a50cac52..6abd1ffa6d57834b2d36b72071001019276f5e99 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 c3d27888971732f8b6c8ccd732d0063d93aec487..636435ef1607a523383352167b5520177d706c9d 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -674,6 +674,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 c5fa377fea704eefc2fbcb8ddd4d8eed9e3f5c69..74a828f027a8d82f5cbe9ba21323114bb2e74e25 100644 --- a/source/client/inc/clientInt.h +++ b/source/client/inc/clientInt.h @@ -151,6 +151,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 669b2bc97eb3e6fab04701aebbf80402432b44c1..35e354e952d32ad39c796475e98187dc033b0020 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 ee6cea79a71a0a5342bd96fef1324260b1650827..8805c68401c0770e52fa7e716d5274136e229bc2 100644 --- a/source/client/src/clientImpl.c +++ b/source/client/src/clientImpl.c @@ -176,6 +176,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 7d623072d664a4b9f1d77251812032f8c4fa4de1..8d1924b1c9143a4594339c2b630c7c8c73a316cd 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,19 @@ 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){ + request->code = TSDB_CODE_SML_INVALID_DB_CONF; + smlBuildInvalidDataMsg(&info->msgBuf, "Cannot write data to a non schemaless database", NULL); + goto end; + } + if (!lines) { request->code = TSDB_CODE_SML_INVALID_DATA; smlBuildInvalidDataMsg(&info->msgBuf, "lines is null", NULL); diff --git a/source/client/test/smlTest.cpp b/source/client/test/smlTest.cpp index 217699e36071e1e4c5e93e391e77a95c4f857af8..4684df8418559812d285db8b0401e21deeaf514e 100644 --- a/source/client/test/smlTest.cpp +++ b/source/client/test/smlTest.cpp @@ -1258,4 +1258,26 @@ TEST(testCase, sml_TD15742_Test) { destroyRequest(request); smlDestroyInfo(info); -} \ No newline at end of file +} + +TEST(testCase, sml_params_Test) { + TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0); + ASSERT_NE(taos, nullptr); + + TAOS_RES* pRes = taos_query(taos, "create database if not exists param"); + taos_free_result(pRes); + + const char *sql[] = { + "test_ms,t0=t c0=f 1626006833641", + }; + TAOS_RES* res = taos_schemaless_insert(taos, (char**)sql, 1, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_MILLI_SECONDS); + ASSERT_EQ(taos_errno(res), TSDB_CODE_PAR_DB_NOT_SPECIFIED); + taos_free_result(pRes); + + pRes = taos_query(taos, "use param"); + taos_free_result(pRes); + + res = taos_schemaless_insert(taos, (char**)sql, 1, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_MILLI_SECONDS); + ASSERT_EQ(taos_errno(res), TSDB_CODE_SML_INVALID_DB_CONF); + taos_free_result(pRes); +} diff --git a/source/common/src/systable.c b/source/common/src/systable.c index 8207ffb22f42d865c54898530154d317af0ea19d..d5999411aea72b79f9c409af97bb1d81e8cad72e 100644 --- a/source/common/src/systable.c +++ b/source/common/src/systable.c @@ -89,7 +89,9 @@ static const SSysDbTableSchema userDBSchema[] = { {.name = "cachelast", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT}, {.name = "precision", .bytes = 2 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, {.name = "single_stable", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT}, + {.name = "schemaless", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT}, {.name = "status", .bytes = 10 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR}, + // {.name = "update", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT}, // disable update }; diff --git a/source/common/src/tmsg.c b/source/common/src/tmsg.c index 2962216b0cebe4841be58f2c52fd109629416d95..2ceee6da04e2493d30b7b3dfaf2de14d8304b68a 100644 --- a/source/common/src/tmsg.c +++ b/source/common/src/tmsg.c @@ -2277,6 +2277,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; @@ -2325,6 +2326,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 d0c737ae5a7518cf864f55ab3f5c9702b7f60073..3ef4a34a5c91b3d498d227d1a370e17da7943f4b 100644 --- a/source/dnode/mnode/impl/inc/mndDef.h +++ b/source/dnode/mnode/impl/inc/mndDef.h @@ -255,6 +255,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 c062a2c5523dd42ac5bcae27dc2863c058f8e374..9bba1a1b2ff2fed39cab7ed8373af9d89cdf606e 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; @@ -899,6 +905,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); @@ -1542,8 +1549,12 @@ static void dumpDbInfoData(SSDataBlock *pBlock, SDbObj *pDb, SShowObj *pShow, in pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.numOfStables, false); + pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); + colDataAppend(pColInfo, rows, (const char *)&pDb->cfg.schemaless, false); + pColInfo = taosArrayGet(pBlock->pDataBlock, cols); colDataAppend(pColInfo, rows, (const char *)statusB, false); + } } diff --git a/source/libs/parser/inc/parInt.h b/source/libs/parser/inc/parInt.h index 3efe6700d2339b2234d33f868c0b42fa993d1b64..bdef3becf7272ba476bffcc0c8693d5976ec152e 100644 --- a/source/libs/parser/inc/parInt.h +++ b/source/libs/parser/inc/parInt.h @@ -32,6 +32,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 8003c011a17af1b81468571054f3fb1be1ed101e..7729d4c415b8ca548077faab8bae3381773630db 100644 --- a/source/libs/parser/src/parInsert.c +++ b/source/libs/parser/src/parInsert.c @@ -1404,6 +1404,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, ...)] @@ -1451,6 +1468,11 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) { (*pQuery)->msgType = TDMT_VND_SUBMIT; (*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) { @@ -1460,7 +1482,7 @@ int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery) { context.pOutput->payloadType = PAYLOAD_TYPE_KV; - int32_t code = skipInsertInto(&context.pSql, &context.msg); + code = skipInsertInto(&context.pSql, &context.msg); 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 55a473982b97dc67308f6851e66d44898cd25fb7..18d64905c983002d466929357bcca526115379ac 100644 --- a/source/libs/parser/src/parTranslater.c +++ b/source/libs/parser/src/parTranslater.c @@ -2647,6 +2647,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; } @@ -4425,6 +4430,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); @@ -4845,6 +4853,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 b81d81c736b177952c10cd722cacdca59c3e37bc..e8edd142d6a530e2d4406576ab997ea122a11d61 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -470,6 +470,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 1781dd31836376cb3e502de0bd05bde50e66288a..e5986cf4dddaa7191e7047e76b51305baefc55c1 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); }