xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat:add schemaless db paras

This commit is contained in:
wangmm0220 2022-05-30 18:00:03 +08:00
parent d7a946e27d
commit f7c21c4502
16 changed files with 88 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/tmsg.h
View File

@ -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;

1
include/libs/parser/parser.h
View File

@ -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);

1
include/util/taoserror.h
View File

@ -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
}

1
source/client/inc/clientInt.h
View File

@ -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 {

1
source/client/src/clientEnv.c
View File

@ -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;

1
source/client/src/clientImpl.c
View File

@ -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);

33
source/client/src/clientSml.c
View File

@ -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);

2
source/common/src/tmsg.c
View File

@ -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);

1
source/dnode/mnode/impl/inc/mndDef.h
View File

@ -302,6 +302,7 @@ typedef struct {
int8_t hashMethod; // default is 1
int32_t numOfRetensions;
SArray* pRetensions;
int8_t schemaless;
} SDbCfg;
typedef struct {

7
source/dnode/mnode/impl/src/mndDb.c
View File

@ -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);

1
source/dnode/mnode/impl/src/mndVgroup.c
View File

@ -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];

1
source/libs/parser/inc/parInt.h
View File

@ -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
}

24
source/libs/parser/src/parInsert.c
View File

@ -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);
}

11
source/libs/parser/src/parTranslater.c
View File

@ -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;

1
source/util/src/terror.c
View File

@ -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
};

1
tests/test/c/sdbDump.c
View File

@ -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);
}