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

feat: parser adapts asynchronous interface

This commit is contained in:
Xiaoyu Wang 2022-05-28 15:11:37 +08:00
parent 1de2045cb8
commit b5db96b717
13 changed files with 920 additions and 150 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

293
source/libs/nodes/src/nodesCodeFuncs.c
View File

@ -2534,7 +2534,7 @@ static const char* jkSessionWindowTsPrimaryKey = "TsPrimaryKey";
static const char* jkSessionWindowGap = "Gap";
static int32_t sessionWindowNodeToJson(const void* pObj, SJson* pJson) {
const SSessionWindowNode * pNode = (const SSessionWindowNode*)pObj;
const SSessionWindowNode* pNode = (const SSessionWindowNode*)pObj;
int32_t code = tjsonAddObject(pJson, jkSessionWindowTsPrimaryKey, nodeToJson, pNode->pCol);
if (TSDB_CODE_SUCCESS == code) {
@ -2546,9 +2546,9 @@ static int32_t sessionWindowNodeToJson(const void* pObj, SJson* pJson) {
static int32_t jsonToSessionWindowNode(const SJson* pJson, void* pObj) {
SSessionWindowNode* pNode = (SSessionWindowNode*)pObj;
int32_t code = jsonToNodeObject(pJson, jkSessionWindowTsPrimaryKey, (SNode **)&pNode->pCol);
int32_t code = jsonToNodeObject(pJson, jkSessionWindowTsPrimaryKey, (SNode**)&pNode->pCol);
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkSessionWindowGap, (SNode **)&pNode->pGap);
code = jsonToNodeObject(pJson, jkSessionWindowGap, (SNode**)&pNode->pGap);
}
return code;
}
@ -2775,6 +2775,150 @@ static int32_t jsonToDownstreamSourceNode(const SJson* pJson, void* pObj) {
return code;
}
static const char* jkDatabaseOptionsBuffer = "Buffer";
static const char* jkDatabaseOptionsCachelast = "Cachelast";
static const char* jkDatabaseOptionsCompressionLevel = "CompressionLevel";
static const char* jkDatabaseOptionsDaysPerFileNode = "DaysPerFileNode";
static const char* jkDatabaseOptionsDaysPerFile = "DaysPerFile";
static const char* jkDatabaseOptionsFsyncPeriod = "FsyncPeriod";
static const char* jkDatabaseOptionsMaxRowsPerBlock = "MaxRowsPerBlock";
static const char* jkDatabaseOptionsMinRowsPerBlock = "MinRowsPerBlock";
static const char* jkDatabaseOptionsKeep = "Keep";
static const char* jkDatabaseOptionsPages = "Pages";
static const char* jkDatabaseOptionsPagesize = "Pagesize";
static const char* jkDatabaseOptionsPrecision = "Precision";
static const char* jkDatabaseOptionsReplica = "Replica";
static const char* jkDatabaseOptionsStrict = "Strict";
static const char* jkDatabaseOptionsWalLevel = "WalLevel";
static const char* jkDatabaseOptionsNumOfVgroups = "NumOfVgroups";
static const char* jkDatabaseOptionsSingleStable = "SingleStable";
static const char* jkDatabaseOptionsRetentions = "Retentions";
static const char* jkDatabaseOptionsSchemaless = "Schemaless";
static int32_t databaseOptionsToJson(const void* pObj, SJson* pJson) {
const SDatabaseOptions* pNode = (const SDatabaseOptions*)pObj;
int32_t code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsBuffer, pNode->buffer);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsCachelast, pNode->cachelast);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsCompressionLevel, pNode->compressionLevel);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkDatabaseOptionsDaysPerFileNode, nodeToJson, pNode->pDaysPerFile);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsDaysPerFile, pNode->daysPerFile);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsFsyncPeriod, pNode->fsyncPeriod);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsMaxRowsPerBlock, pNode->maxRowsPerBlock);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsMinRowsPerBlock, pNode->minRowsPerBlock);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkDatabaseOptionsKeep, pNode->pKeep);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsPages, pNode->pages);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsPagesize, pNode->pagesize);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddStringToObject(pJson, jkDatabaseOptionsPrecision, pNode->precisionStr);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsReplica, pNode->replica);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsStrict, pNode->strict);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsWalLevel, pNode->walLevel);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsNumOfVgroups, pNode->numOfVgroups);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsSingleStable, pNode->singleStable);
}
if (TSDB_CODE_SUCCESS == code) {
code = nodeListToJson(pJson, jkDatabaseOptionsRetentions, pNode->pRetentions);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkDatabaseOptionsSchemaless, pNode->schemaless);
}
return code;
}
static int32_t jsonToDatabaseOptions(const SJson* pJson, void* pObj) {
SDatabaseOptions* pNode = (SDatabaseOptions*)pObj;
int32_t code = tjsonGetIntValue(pJson, jkDatabaseOptionsBuffer, &pNode->buffer);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkDatabaseOptionsCachelast, &pNode->cachelast);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkDatabaseOptionsCompressionLevel, &pNode->compressionLevel);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkDatabaseOptionsDaysPerFileNode, (SNode**)&pNode->pDaysPerFile);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetIntValue(pJson, jkDatabaseOptionsDaysPerFile, &pNode->daysPerFile);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetIntValue(pJson, jkDatabaseOptionsFsyncPeriod, &pNode->fsyncPeriod);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetIntValue(pJson, jkDatabaseOptionsMaxRowsPerBlock, &pNode->maxRowsPerBlock);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetIntValue(pJson, jkDatabaseOptionsMinRowsPerBlock, &pNode->minRowsPerBlock);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkDatabaseOptionsKeep, &pNode->pKeep);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetIntValue(pJson, jkDatabaseOptionsPages, &pNode->pages);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetIntValue(pJson, jkDatabaseOptionsPagesize, &pNode->pagesize);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetStringValue(pJson, jkDatabaseOptionsPrecision, pNode->precisionStr);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkDatabaseOptionsReplica, &pNode->replica);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkDatabaseOptionsStrict, &pNode->strict);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkDatabaseOptionsWalLevel, &pNode->walLevel);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetIntValue(pJson, jkDatabaseOptionsNumOfVgroups, &pNode->numOfVgroups);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkDatabaseOptionsSingleStable, &pNode->singleStable);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeList(pJson, jkDatabaseOptionsRetentions, &pNode->pRetentions);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkDatabaseOptionsSchemaless, &pNode->schemaless);
}
return code;
}
static const char* jkDataBlockDescDataBlockId = "DataBlockId";
static const char* jkDataBlockDescSlots = "Slots";
static const char* jkDataBlockTotalRowSize = "TotalRowSize";
@ -2977,6 +3121,130 @@ static int32_t jsonToSelectStmt(const SJson* pJson, void* pObj) {
return code;
}
static const char* jkAlterDatabaseStmtDbName = "DbName";
static const char* jkAlterDatabaseStmtOptions = "Options";
static int32_t alterDatabaseStmtToJson(const void* pObj, SJson* pJson) {
const SAlterDatabaseStmt* pNode = (const SAlterDatabaseStmt*)pObj;
int32_t code = tjsonAddStringToObject(pJson, jkAlterDatabaseStmtDbName, pNode->dbName);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkAlterDatabaseStmtOptions, nodeToJson, pNode->pOptions);
}
return code;
}
static int32_t jsonToAlterDatabaseStmt(const SJson* pJson, void* pObj) {
SAlterDatabaseStmt* pNode = (SAlterDatabaseStmt*)pObj;
int32_t code = tjsonGetStringValue(pJson, jkAlterDatabaseStmtDbName, pNode->dbName);
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkAlterDatabaseStmtOptions, (SNode**)&pNode->pOptions);
}
return code;
}
static const char* jkAlterTableStmtDbName = "DbName";
static const char* jkAlterTableStmtTableName = "TableName";
static const char* jkAlterTableStmtAlterType = "AlterType";
static const char* jkAlterTableStmtColName = "ColName";
static const char* jkAlterTableStmtNewColName = "NewColName";
static const char* jkAlterTableStmtOptions = "Options";
static const char* jkAlterTableStmtNewDataType = "NewDataType";
static const char* jkAlterTableStmtNewTagVal = "NewTagVal";
static int32_t alterTableStmtToJson(const void* pObj, SJson* pJson) {
const SAlterTableStmt* pNode = (const SAlterTableStmt*)pObj;
int32_t code = tjsonAddStringToObject(pJson, jkAlterTableStmtDbName, pNode->dbName);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddStringToObject(pJson, jkAlterTableStmtTableName, pNode->tableName);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddIntegerToObject(pJson, jkAlterTableStmtAlterType, pNode->alterType);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddStringToObject(pJson, jkAlterTableStmtColName, pNode->colName);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddStringToObject(pJson, jkAlterTableStmtNewColName, pNode->newColName);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkAlterTableStmtOptions, nodeToJson, pNode->pOptions);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkAlterTableStmtNewDataType, dataTypeToJson, &pNode->dataType);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddObject(pJson, jkAlterTableStmtOptions, nodeToJson, pNode->pVal);
}
return code;
}
static int32_t jsonToAlterTableStmt(const SJson* pJson, void* pObj) {
SAlterTableStmt* pNode = (SAlterTableStmt*)pObj;
int32_t code = tjsonGetStringValue(pJson, jkAlterTableStmtDbName, pNode->dbName);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetStringValue(pJson, jkAlterTableStmtTableName, pNode->tableName);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetTinyIntValue(pJson, jkAlterTableStmtAlterType, &pNode->alterType);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetStringValue(pJson, jkAlterTableStmtColName, pNode->colName);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetStringValue(pJson, jkAlterTableStmtNewColName, pNode->newColName);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkAlterTableStmtOptions, (SNode**)&pNode->pOptions);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonToObject(pJson, jkAlterTableStmtNewDataType, jsonToDataType, &pNode->dataType);
}
if (TSDB_CODE_SUCCESS == code) {
code = jsonToNodeObject(pJson, jkAlterTableStmtOptions, (SNode**)&pNode->pVal);
}
return code;
}
static const char* jkAlterDnodeStmtDnodeId = "DnodeId";
static const char* jkAlterDnodeStmtConfig = "Config";
static const char* jkAlterDnodeStmtValue = "Value";
static int32_t alterDnodeStmtToJson(const void* pObj, SJson* pJson) {
const SAlterDnodeStmt* pNode = (const SAlterDnodeStmt*)pObj;
int32_t code = tjsonAddIntegerToObject(pJson, jkAlterDnodeStmtDnodeId, pNode->dnodeId);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddStringToObject(pJson, jkAlterDnodeStmtConfig, pNode->config);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonAddStringToObject(pJson, jkAlterDnodeStmtValue, pNode->value);
}
return code;
}
static int32_t jsonToAlterDnodeStmt(const SJson* pJson, void* pObj) {
SAlterDnodeStmt* pNode = (SAlterDnodeStmt*)pObj;
int32_t code = tjsonGetIntValue(pJson, jkAlterDnodeStmtDnodeId, &pNode->dnodeId);
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetStringValue(pJson, jkAlterDnodeStmtConfig, pNode->config);
}
if (TSDB_CODE_SUCCESS == code) {
code = tjsonGetStringValue(pJson, jkAlterDnodeStmtValue, pNode->value);
}
return code;
}
static const char* jkCreateTopicStmtTopicName = "TopicName";
static const char* jkCreateTopicStmtSubscribeDbName = "SubscribeDbName";
static const char* jkCreateTopicStmtIgnoreExists = "IgnoreExists";
@ -3061,6 +3329,8 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
break;
case QUERY_NODE_DOWNSTREAM_SOURCE:
return downstreamSourceNodeToJson(pObj, pJson);
case QUERY_NODE_DATABASE_OPTIONS:
return databaseOptionsToJson(pObj, pJson);
case QUERY_NODE_LEFT_VALUE:
return TSDB_CODE_SUCCESS; // SLeftValueNode has no fields to serialize.
case QUERY_NODE_SET_OPERATOR:
@ -3069,8 +3339,17 @@ static int32_t specificNodeToJson(const void* pObj, SJson* pJson) {
return selectStmtToJson(pObj, pJson);
case QUERY_NODE_VNODE_MODIF_STMT:
case QUERY_NODE_CREATE_DATABASE_STMT:
break;
case QUERY_NODE_ALTER_DATABASE_STMT:
return alterDatabaseStmtToJson(pObj, pJson);
case QUERY_NODE_CREATE_TABLE_STMT:
break;
case QUERY_NODE_ALTER_TABLE_STMT:
return alterTableStmtToJson(pObj, pJson);
case QUERY_NODE_USE_DATABASE_STMT:
break;
case QUERY_NODE_ALTER_DNODE_STMT:
return alterDnodeStmtToJson(pObj, pJson);
case QUERY_NODE_SHOW_DATABASES_STMT:
case QUERY_NODE_SHOW_TABLES_STMT:
break;
@ -3177,12 +3456,20 @@ static int32_t jsonToSpecificNode(const SJson* pJson, void* pObj) {
return jsonToSlotDescNode(pJson, pObj);
case QUERY_NODE_DOWNSTREAM_SOURCE:
return jsonToDownstreamSourceNode(pJson, pObj);
case QUERY_NODE_DATABASE_OPTIONS:
return jsonToDatabaseOptions(pJson, pObj);
case QUERY_NODE_LEFT_VALUE:
return TSDB_CODE_SUCCESS; // SLeftValueNode has no fields to deserialize.
case QUERY_NODE_SET_OPERATOR:
return jsonToSetOperator(pJson, pObj);
case QUERY_NODE_SELECT_STMT:
return jsonToSelectStmt(pJson, pObj);
case QUERY_NODE_ALTER_DATABASE_STMT:
return jsonToAlterDatabaseStmt(pJson, pObj);
case QUERY_NODE_ALTER_TABLE_STMT:
return jsonToAlterTableStmt(pJson, pObj);
case QUERY_NODE_ALTER_DNODE_STMT:
return jsonToAlterDnodeStmt(pJson, pObj);
case QUERY_NODE_CREATE_TOPIC_STMT:
return jsonToCreateTopicStmt(pJson, pObj);
case QUERY_NODE_LOGIC_PLAN_SCAN:

17
source/libs/parser/inc/parAst.h
View File

@ -27,14 +27,13 @@ extern "C" {
#include "querynodes.h"
typedef struct SAstCreateContext {
SParseContext* pQueryCxt;
SMsgBuf msgBuf;
bool notSupport;
SNode* pRootNode;
int16_t placeholderNo;
SArray* pPlaceholderValues;
int32_t errCode;
SParseMetaCache* pMetaCache;
SParseContext* pQueryCxt;
SMsgBuf msgBuf;
bool notSupport;
SNode* pRootNode;
int16_t placeholderNo;
SArray* pPlaceholderValues;
int32_t errCode;
} SAstCreateContext;
typedef enum EDatabaseOptionType {
@ -75,7 +74,7 @@ typedef struct SAlterOption {
extern SToken nil_token;
int32_t initAstCreateContext(SParseContext* pParseCxt, SAstCreateContext* pCxt);
void initAstCreateContext(SParseContext* pParseCxt, SAstCreateContext* pCxt);
SNode* createRawExprNode(SAstCreateContext* pCxt, const SToken* pToken, SNode* pNode);
SNode* createRawExprNodeExt(SAstCreateContext* pCxt, const SToken* pStart, const SToken* pEnd, SNode* pNode);

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

@ -26,6 +26,7 @@ extern "C" {
int32_t parseInsertSql(SParseContext* pContext, SQuery** pQuery);
int32_t parse(SParseContext* pParseCxt, SQuery** pQuery);
int32_t collectMetaKey(SParseContext* pParseCxt, SQuery* pQuery);
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);

23
source/libs/parser/inc/parUtil.h
View File

@ -24,12 +24,12 @@ extern "C" {
#include "os.h"
#include "query.h"
#define parserFatal(param, ...) qFatal("PARSER: " param, __VA_ARGS__)
#define parserError(param, ...) qError("PARSER: " param, __VA_ARGS__)
#define parserWarn(param, ...) qWarn("PARSER: " param, __VA_ARGS__)
#define parserInfo(param, ...) qInfo("PARSER: " param, __VA_ARGS__)
#define parserDebug(param, ...) qDebug("PARSER: " param, __VA_ARGS__)
#define parserTrace(param, ...) qTrace("PARSER: " param, __VA_ARGS__)
#define parserFatal(param, ...) qFatal("PARSER: " param, ##__VA_ARGS__)
#define parserError(param, ...) qError("PARSER: " param, ##__VA_ARGS__)
#define parserWarn(param, ...) qWarn("PARSER: " param, ##__VA_ARGS__)
#define parserInfo(param, ...) qInfo("PARSER: " param, ##__VA_ARGS__)
#define parserDebug(param, ...) qDebug("PARSER: " param, ##__VA_ARGS__)
#define parserTrace(param, ...) qTrace("PARSER: " param, ##__VA_ARGS__)
#define PK_TS_COL_INTERNAL_NAME "_rowts"
@ -62,12 +62,15 @@ int32_t trimString(const char* src, int32_t len, char* dst, int32_t dlen);
int32_t buildCatalogReq(const SParseMetaCache* pMetaCache, SCatalogReq* pCatalogReq);
int32_t putMetaDataToCache(const SCatalogReq* pCatalogReq, const SMetaData* pMetaData, SParseMetaCache* pMetaCache);
int32_t reserveTableMetaInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache);
int32_t reserveDbVgInfoInCache(int32_t acctId, const char* pDb, SParseMetaCache* pMetaCache);
int32_t reserveTableVgroupInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache);
int32_t reserveDbCfgInCache(int32_t acctId, const char* pDb, SParseMetaCache* pMetaCache);
int32_t getTableMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta);
int32_t getDBVgInfoFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SArray** pVgInfo);
int32_t getTableHashVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName, SVgroupInfo* pVgroup);
int32_t getDBVgVersionFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, int32_t* pVersion, int64_t* pDbId,
int32_t getDbVgInfoFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SArray** pVgInfo);
int32_t getTableVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName, SVgroupInfo* pVgroup);
int32_t getDbVgVersionFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, int32_t* pVersion, int64_t* pDbId,
int32_t* pTableNum);
int32_t getDBCfgFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SDbCfgInfo* pInfo);
int32_t getDbCfgFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SDbCfgInfo* pInfo);
#ifdef __cplusplus
}

16
source/libs/parser/src/parAstCreater.c
View File

@ -38,7 +38,7 @@
SToken nil_token = {.type = TK_NK_NIL, .n = 0, .z = NULL};
int32_t initAstCreateContext(SParseContext* pParseCxt, SAstCreateContext* pCxt) {
void initAstCreateContext(SParseContext* pParseCxt, SAstCreateContext* pCxt) {
memset(pCxt, 0, sizeof(SAstCreateContext));
pCxt->pQueryCxt = pParseCxt;
pCxt->msgBuf.buf = pParseCxt->pMsg;
@ -48,13 +48,6 @@ int32_t initAstCreateContext(SParseContext* pParseCxt, SAstCreateContext* pCxt)
pCxt->placeholderNo = 0;
pCxt->pPlaceholderValues = NULL;
pCxt->errCode = TSDB_CODE_SUCCESS;
if (pParseCxt->async) {
pCxt->pMetaCache = taosMemoryCalloc(1, sizeof(SParseMetaCache));
if (NULL == pCxt->pMetaCache) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
return TSDB_CODE_SUCCESS;
}
static void copyStringFormStringToken(SToken* pToken, char* pBuf, int32_t len) {
@ -472,13 +465,6 @@ SNode* createRealTableNode(SAstCreateContext* pCxt, SToken* pDbName, SToken* pTa
strncpy(realTable->table.tableAlias, pTableName->z, pTableName->n);
}
strncpy(realTable->table.tableName, pTableName->z, pTableName->n);
if (NULL != pCxt->pMetaCache) {
if (TSDB_CODE_SUCCESS != reserveTableMetaInCache(pCxt->pQueryCxt->acctId, realTable->table.dbName,
realTable->table.tableName, pCxt->pMetaCache)) {
nodesDestroyNode(realTable);
CHECK_OUT_OF_MEM(NULL);
}
}
return (SNode*)realTable;
}

367
source/libs/parser/src/parAstParser.c
View File

@ -14,10 +14,10 @@
*/
#include "os.h"
#include "parInt.h"
#include "parAst.h"
#include "parInt.h"
#include "parToken.h"
#include "systable.h"
typedef void* (*FMalloc)(size_t);
typedef void (*FFree)(void*);
@ -82,8 +82,369 @@ abort_parse:
(*pQuery)->pRoot = cxt.pRootNode;
(*pQuery)->placeholderNum = cxt.placeholderNo;
TSWAP((*pQuery)->pPlaceholderValues, cxt.pPlaceholderValues);
TSWAP((*pQuery)->pMetaCache, cxt.pMetaCache);
}
taosArrayDestroy(cxt.pPlaceholderValues);
return cxt.errCode;
}
typedef struct SCollectMetaKeyCxt {
SParseContext* pParseCxt;
SParseMetaCache* pMetaCache;
} SCollectMetaKeyCxt;
static void destroyCollectMetaKeyCxt(SCollectMetaKeyCxt* pCxt) {
if (NULL != pCxt->pMetaCache) {
// TODO
}
}
typedef struct SCollectMetaKeyFromExprCxt {
SCollectMetaKeyCxt* pComCxt;
int32_t errCode;
} SCollectMetaKeyFromExprCxt;
static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt);
static EDealRes collectMetaKeyFromRealTable(SCollectMetaKeyFromExprCxt* pCxt, SRealTableNode* pRealTable) {
pCxt->errCode = reserveTableMetaInCache(pCxt->pComCxt->pParseCxt->acctId, pRealTable->table.dbName,
pRealTable->table.tableName, pCxt->pComCxt->pMetaCache);
if (TSDB_CODE_SUCCESS == pCxt->errCode) {
pCxt->errCode = reserveTableVgroupInCache(pCxt->pComCxt->pParseCxt->acctId, pRealTable->table.dbName,
pRealTable->table.tableName, pCxt->pComCxt->pMetaCache);
}
return TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_CONTINUE : DEAL_RES_ERROR;
}
static EDealRes collectMetaKeyFromTempTable(SCollectMetaKeyFromExprCxt* pCxt, STempTableNode* pTempTable) {
pCxt->errCode = collectMetaKeyFromQuery(pCxt->pComCxt, pTempTable->pSubquery);
return TSDB_CODE_SUCCESS == pCxt->errCode ? DEAL_RES_CONTINUE : DEAL_RES_ERROR;
}
static EDealRes collectMetaKeyFromExprImpl(SNode* pNode, void* pContext) {
SCollectMetaKeyFromExprCxt* pCxt = pContext;
switch (nodeType(pNode)) {
case QUERY_NODE_FUNCTION:
break;
case QUERY_NODE_REAL_TABLE:
return collectMetaKeyFromRealTable(pCxt, (SRealTableNode*)pNode);
case QUERY_NODE_TEMP_TABLE:
return collectMetaKeyFromTempTable(pCxt, (STempTableNode*)pNode);
default:
break;
}
return DEAL_RES_CONTINUE;
}
static int32_t collectMetaKeyFromExprs(SCollectMetaKeyCxt* pCxt, SNodeList* pList) {
SCollectMetaKeyFromExprCxt cxt = {.pComCxt = pCxt, .errCode = TSDB_CODE_SUCCESS};
nodesWalkExprs(pList, collectMetaKeyFromExprImpl, &cxt);
return cxt.errCode;
}
static int32_t collectMetaKeyFromSetOperator(SCollectMetaKeyCxt* pCxt, SSetOperator* pStmt) {
int32_t code = collectMetaKeyFromQuery(pCxt, pStmt->pLeft);
if (TSDB_CODE_SUCCESS == code) {
code = collectMetaKeyFromQuery(pCxt, pStmt->pRight);
}
if (TSDB_CODE_SUCCESS == code) {
code = collectMetaKeyFromExprs(pCxt, pStmt->pOrderByList);
}
return code;
}
static int32_t collectMetaKeyFromSelect(SCollectMetaKeyCxt* pCxt, SSelectStmt* pStmt) {
SCollectMetaKeyFromExprCxt cxt = {.pComCxt = pCxt, .errCode = TSDB_CODE_SUCCESS};
nodesWalkSelectStmt(pStmt, SQL_CLAUSE_FROM, collectMetaKeyFromExprImpl, &cxt);
return cxt.errCode;
}
static int32_t collectMetaKeyFromCreateTable(SCollectMetaKeyCxt* pCxt, SCreateTableStmt* pStmt) {
if (NULL == pStmt->pTags) {
return reserveTableVgroupInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, pCxt->pMetaCache);
} else {
return reserveDbCfgInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pCxt->pMetaCache);
}
}
static int32_t collectMetaKeyFromCreateMultiTable(SCollectMetaKeyCxt* pCxt, SCreateMultiTableStmt* pStmt) {
int32_t code = TSDB_CODE_SUCCESS;
SNode* pNode = NULL;
FOREACH(pNode, pStmt->pSubTables) {
SCreateSubTableClause* pClause = (SCreateSubTableClause*)pNode;
code =
reserveTableMetaInCache(pCxt->pParseCxt->acctId, pClause->useDbName, pClause->useTableName, pCxt->pMetaCache);
if (TSDB_CODE_SUCCESS == code) {
code = reserveTableVgroupInCache(pCxt->pParseCxt->acctId, pClause->dbName, pClause->tableName, pCxt->pMetaCache);
}
if (TSDB_CODE_SUCCESS != code) {
break;
}
}
return code;
}
static int32_t collectMetaKeyFromAlterTable(SCollectMetaKeyCxt* pCxt, SAlterTableStmt* pStmt) {
int32_t code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, pCxt->pMetaCache);
if (TSDB_CODE_SUCCESS == code) {
code = reserveTableVgroupInCache(pCxt->pParseCxt->acctId, pStmt->dbName, pStmt->tableName, pCxt->pMetaCache);
}
return code;
}
static int32_t collectMetaKeyFromCreateIndex(SCollectMetaKeyCxt* pCxt, SCreateIndexStmt* pStmt) {
int32_t code = TSDB_CODE_SUCCESS;
if (INDEX_TYPE_SMA == pStmt->indexType) {
code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, pCxt->pParseCxt->db, pStmt->tableName, pCxt->pMetaCache);
if (TSDB_CODE_SUCCESS == code) {
code =
reserveTableVgroupInCache(pCxt->pParseCxt->acctId, pCxt->pParseCxt->db, pStmt->tableName, pCxt->pMetaCache);
}
}
return code;
}
static int32_t collectMetaKeyFromCreateTopic(SCollectMetaKeyCxt* pCxt, SCreateTopicStmt* pStmt) {
if (NULL != pStmt->pQuery) {
return collectMetaKeyFromQuery(pCxt, pStmt->pQuery);
}
return TSDB_CODE_SUCCESS;
}
static int32_t collectMetaKeyFromExplain(SCollectMetaKeyCxt* pCxt, SExplainStmt* pStmt) {
return collectMetaKeyFromQuery(pCxt, pStmt->pQuery);
}
static int32_t collectMetaKeyFromCreateStream(SCollectMetaKeyCxt* pCxt, SCreateStreamStmt* pStmt) {
return collectMetaKeyFromQuery(pCxt, pStmt->pQuery);
}
static int32_t collectMetaKeyFromShowDnodes(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_DNODES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowMnodes(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_MNODES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowModules(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_MODULES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowQnodes(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_QNODES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowSnodes(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_SNODES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowBnodes(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_BNODES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowDatabases(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_DATABASES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowFunctions(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_FUNCTIONS,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowIndexes(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_INDEXES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowStables(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_STABLES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowStreams(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_STREAMS,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowTables(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
int32_t code = reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB,
TSDB_INS_TABLE_USER_TABLES, pCxt->pMetaCache);
if (TSDB_CODE_SUCCESS == code) {
if (NULL != pStmt->pDbName) {
code = reserveDbVgInfoInCache(pCxt->pParseCxt->acctId, ((SValueNode*)pStmt->pDbName)->literal, pCxt->pMetaCache);
} else {
code = reserveDbVgInfoInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, pCxt->pMetaCache);
}
}
return code;
}
static int32_t collectMetaKeyFromShowUsers(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_USER_USERS,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowLicence(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_LICENCES,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowVgroups(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_INFORMATION_SCHEMA_DB, TSDB_INS_TABLE_VGROUPS,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowTopics(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_TOPICS,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromShowTransactions(SCollectMetaKeyCxt* pCxt, SShowStmt* pStmt) {
return reserveTableMetaInCache(pCxt->pParseCxt->acctId, TSDB_PERFORMANCE_SCHEMA_DB, TSDB_PERFS_TABLE_TRANS,
pCxt->pMetaCache);
}
static int32_t collectMetaKeyFromQuery(SCollectMetaKeyCxt* pCxt, SNode* pStmt) {
switch (nodeType(pStmt)) {
case QUERY_NODE_SET_OPERATOR:
return collectMetaKeyFromSetOperator(pCxt, (SSetOperator*)pStmt);
case QUERY_NODE_SELECT_STMT:
return collectMetaKeyFromSelect(pCxt, (SSelectStmt*)pStmt);
case QUERY_NODE_VNODE_MODIF_STMT:
case QUERY_NODE_CREATE_DATABASE_STMT:
case QUERY_NODE_DROP_DATABASE_STMT:
case QUERY_NODE_ALTER_DATABASE_STMT:
break;
case QUERY_NODE_CREATE_TABLE_STMT:
return collectMetaKeyFromCreateTable(pCxt, (SCreateTableStmt*)pStmt);
case QUERY_NODE_CREATE_SUBTABLE_CLAUSE:
break;
case QUERY_NODE_CREATE_MULTI_TABLE_STMT:
return collectMetaKeyFromCreateMultiTable(pCxt, (SCreateMultiTableStmt*)pStmt);
case QUERY_NODE_DROP_TABLE_CLAUSE:
case QUERY_NODE_DROP_TABLE_STMT:
case QUERY_NODE_DROP_SUPER_TABLE_STMT:
break;
case QUERY_NODE_ALTER_TABLE_STMT:
return collectMetaKeyFromAlterTable(pCxt, (SAlterTableStmt*)pStmt);
case QUERY_NODE_CREATE_USER_STMT:
case QUERY_NODE_ALTER_USER_STMT:
case QUERY_NODE_DROP_USER_STMT:
case QUERY_NODE_USE_DATABASE_STMT:
case QUERY_NODE_CREATE_DNODE_STMT:
case QUERY_NODE_DROP_DNODE_STMT:
case QUERY_NODE_ALTER_DNODE_STMT:
break;
case QUERY_NODE_CREATE_INDEX_STMT:
return collectMetaKeyFromCreateIndex(pCxt, (SCreateIndexStmt*)pStmt);
case QUERY_NODE_DROP_INDEX_STMT:
case QUERY_NODE_CREATE_QNODE_STMT:
case QUERY_NODE_DROP_QNODE_STMT:
case QUERY_NODE_CREATE_BNODE_STMT:
case QUERY_NODE_DROP_BNODE_STMT:
case QUERY_NODE_CREATE_SNODE_STMT:
case QUERY_NODE_DROP_SNODE_STMT:
case QUERY_NODE_CREATE_MNODE_STMT:
case QUERY_NODE_DROP_MNODE_STMT:
break;
case QUERY_NODE_CREATE_TOPIC_STMT:
return collectMetaKeyFromCreateTopic(pCxt, (SCreateTopicStmt*)pStmt);
case QUERY_NODE_DROP_TOPIC_STMT:
case QUERY_NODE_DROP_CGROUP_STMT:
case QUERY_NODE_ALTER_LOCAL_STMT:
break;
case QUERY_NODE_EXPLAIN_STMT:
return collectMetaKeyFromExplain(pCxt, (SExplainStmt*)pStmt);
case QUERY_NODE_DESCRIBE_STMT:
case QUERY_NODE_RESET_QUERY_CACHE_STMT:
case QUERY_NODE_COMPACT_STMT:
case QUERY_NODE_CREATE_FUNCTION_STMT:
case QUERY_NODE_DROP_FUNCTION_STMT:
break;
case QUERY_NODE_CREATE_STREAM_STMT:
return collectMetaKeyFromCreateStream(pCxt, (SCreateStreamStmt*)pStmt);
case QUERY_NODE_DROP_STREAM_STMT:
case QUERY_NODE_MERGE_VGROUP_STMT:
case QUERY_NODE_REDISTRIBUTE_VGROUP_STMT:
case QUERY_NODE_SPLIT_VGROUP_STMT:
case QUERY_NODE_SYNCDB_STMT:
case QUERY_NODE_GRANT_STMT:
case QUERY_NODE_REVOKE_STMT:
case QUERY_NODE_SHOW_DNODES_STMT:
return collectMetaKeyFromShowDnodes(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_MNODES_STMT:
return collectMetaKeyFromShowMnodes(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_MODULES_STMT:
return collectMetaKeyFromShowModules(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_QNODES_STMT:
return collectMetaKeyFromShowQnodes(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_SNODES_STMT:
return collectMetaKeyFromShowSnodes(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_BNODES_STMT:
return collectMetaKeyFromShowBnodes(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_CLUSTER_STMT:
break;
case QUERY_NODE_SHOW_DATABASES_STMT:
return collectMetaKeyFromShowDatabases(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_FUNCTIONS_STMT:
return collectMetaKeyFromShowFunctions(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_INDEXES_STMT:
return collectMetaKeyFromShowIndexes(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_STABLES_STMT:
return collectMetaKeyFromShowStables(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_STREAMS_STMT:
return collectMetaKeyFromShowStreams(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_TABLES_STMT:
return collectMetaKeyFromShowTables(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_USERS_STMT:
return collectMetaKeyFromShowUsers(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_LICENCE_STMT:
return collectMetaKeyFromShowLicence(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_VGROUPS_STMT:
return collectMetaKeyFromShowVgroups(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_TOPICS_STMT:
return collectMetaKeyFromShowTopics(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_SHOW_CONSUMERS_STMT:
case QUERY_NODE_SHOW_SUBSCRIBES_STMT:
case QUERY_NODE_SHOW_SMAS_STMT:
case QUERY_NODE_SHOW_CONFIGS_STMT:
case QUERY_NODE_SHOW_CONNECTIONS_STMT:
case QUERY_NODE_SHOW_QUERIES_STMT:
case QUERY_NODE_SHOW_VNODES_STMT:
case QUERY_NODE_SHOW_APPS_STMT:
case QUERY_NODE_SHOW_SCORES_STMT:
case QUERY_NODE_SHOW_VARIABLE_STMT:
case QUERY_NODE_SHOW_CREATE_DATABASE_STMT:
case QUERY_NODE_SHOW_CREATE_TABLE_STMT:
case QUERY_NODE_SHOW_CREATE_STABLE_STMT:
break;
case QUERY_NODE_SHOW_TRANSACTIONS_STMT:
return collectMetaKeyFromShowTransactions(pCxt, (SShowStmt*)pStmt);
case QUERY_NODE_KILL_CONNECTION_STMT:
case QUERY_NODE_KILL_QUERY_STMT:
case QUERY_NODE_KILL_TRANSACTION_STMT:
default:
break;
}
return TSDB_CODE_SUCCESS;
}
int32_t collectMetaKey(SParseContext* pParseCxt, SQuery* pQuery) {
SCollectMetaKeyCxt cxt = {.pParseCxt = pParseCxt, .pMetaCache = taosMemoryCalloc(1, sizeof(SParseMetaCache))};
if (NULL == cxt.pMetaCache) {
return TSDB_CODE_OUT_OF_MEMORY;
}
int32_t code = collectMetaKeyFromQuery(&cxt, pQuery->pRoot);
if (TSDB_CODE_SUCCESS == code) {
TSWAP(pQuery->pMetaCache, cxt.pMetaCache);
}
destroyCollectMetaKeyCxt(&cxt);
return code;
}

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

@ -152,7 +152,7 @@ static int32_t getDBVgInfoImpl(STranslateContext* pCxt, const SName* pName, SArr
tNameGetFullDbName(pName, fullDbName);
int32_t code = TSDB_CODE_SUCCESS;
if (pParCxt->async) {
code = getDBVgInfoFromCache(pCxt->pMetaCache, fullDbName, pVgInfo);
code = getDbVgInfoFromCache(pCxt->pMetaCache, fullDbName, pVgInfo);
} else {
code = collectUseDatabaseImpl(fullDbName, pCxt->pDbs);
if (TSDB_CODE_SUCCESS == code) {
@ -177,7 +177,7 @@ static int32_t getTableHashVgroupImpl(STranslateContext* pCxt, const SName* pNam
SParseContext* pParCxt = pCxt->pParseCxt;
int32_t code = TSDB_CODE_SUCCESS;
if (pParCxt->async) {
code = getTableHashVgroupFromCache(pCxt->pMetaCache, pName, pInfo);
code = getTableVgroupFromCache(pCxt->pMetaCache, pName, pInfo);
} else {
code = collectUseDatabase(pName, pCxt->pDbs);
if (TSDB_CODE_SUCCESS == code) {
@ -205,7 +205,7 @@ static int32_t getDBVgVersion(STranslateContext* pCxt, const char* pDbFName, int
SParseContext* pParCxt = pCxt->pParseCxt;
int32_t code = TSDB_CODE_SUCCESS;
if (pParCxt->async) {
code = getDBVgVersionFromCache(pCxt->pMetaCache, pDbFName, pVersion, pDbId, pTableNum);
code = getDbVgVersionFromCache(pCxt->pMetaCache, pDbFName, pVersion, pDbId, pTableNum);
} else {
code = collectUseDatabaseImpl(pDbFName, pCxt->pDbs);
if (TSDB_CODE_SUCCESS == code) {
@ -226,7 +226,7 @@ static int32_t getDBCfg(STranslateContext* pCxt, const char* pDbName, SDbCfgInfo
tNameGetFullDbName(&name, dbFname);
int32_t code = TSDB_CODE_SUCCESS;
if (pParCxt->async) {
code = getDBCfgFromCache(pCxt->pMetaCache, dbFname, pInfo);
code = getDbCfgFromCache(pCxt->pMetaCache, dbFname, pInfo);
} else {
code = collectUseDatabaseImpl(dbFname, pCxt->pDbs);
if (TSDB_CODE_SUCCESS == code) {
@ -1212,7 +1212,6 @@ static int32_t setSysTableVgroupList(STranslateContext* pCxt, SName* pName, SRea
int32_t code = TSDB_CODE_SUCCESS;
SArray* vgroupList = NULL;
if ('\0' != pRealTable->qualDbName[0]) {
// todo release after mnode can be processed
if (0 != strcmp(pRealTable->qualDbName, TSDB_INFORMATION_SCHEMA_DB)) {
code = getDBVgInfo(pCxt, pRealTable->qualDbName, &vgroupList);
}
@ -1220,7 +1219,6 @@ static int32_t setSysTableVgroupList(STranslateContext* pCxt, SName* pName, SRea
code = getDBVgInfoImpl(pCxt, pName, &vgroupList);
}
// todo release after mnode can be processed
if (TSDB_CODE_SUCCESS == code) {
code = addMnodeToVgroupList(&pCxt->pParseCxt->mgmtEpSet, &vgroupList);
}

109
source/libs/parser/src/parUtil.c
View File

@ -15,6 +15,7 @@
#include "parUtil.h"
#include "cJSON.h"
#include "querynodes.h"
static char* getSyntaxErrFormat(int32_t errCode) {
switch (errCode) {
@ -255,17 +256,8 @@ STableComInfo getTableInfo(const STableMeta* pTableMeta) {
return pTableMeta->tableInfo;
}
static uint32_t getTableMetaSize(const STableMeta* pTableMeta) {
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) {
size_t size = getTableMetaSize(pTableMeta);
size_t size = TABLE_META_SIZE(pTableMeta);
STableMeta* p = taosMemoryMalloc(size);
memcpy(p, pTableMeta, size);
@ -512,7 +504,7 @@ int32_t buildCatalogReq(const SParseMetaCache* pMetaCache, SCatalogReq* pCatalog
code = buildTableVgroupReq(pMetaCache->pTableVgroup, &pCatalogReq->pTableHash);
}
if (TSDB_CODE_SUCCESS == code) {
code = buildDbCfgReq(pMetaCache->pDbVgroup, &pCatalogReq->pDbCfg);
code = buildDbCfgReq(pMetaCache->pDbCfg, &pCatalogReq->pDbCfg);
}
return code;
}
@ -582,51 +574,96 @@ int32_t putMetaDataToCache(const SCatalogReq* pCatalogReq, const SMetaData* pMet
return code;
}
int32_t reserveTableMetaInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache) {
if (NULL == pMetaCache->pTableMeta) {
pMetaCache->pTableMeta = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
if (NULL == pMetaCache->pTableMeta) {
static int32_t reserveTableReqInCache(int32_t acctId, const char* pDb, const char* pTable, SHashObj** pTables) {
if (NULL == *pTables) {
*pTables = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
if (NULL == *pTables) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
char fullName[TSDB_TABLE_FNAME_LEN];
int32_t len = snprintf(fullName, sizeof(fullName), "%d.%s.%s", acctId, pDb, pTable);
return taosHashPut(pMetaCache->pTableMeta, fullName, len, &len, POINTER_BYTES);
return taosHashPut(*pTables, fullName, len, &len, POINTER_BYTES);
}
int32_t reserveTableMetaInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache) {
return reserveTableReqInCache(acctId, pDb, pTable, &pMetaCache->pTableMeta);
}
int32_t getTableMetaFromCache(SParseMetaCache* pMetaCache, const SName* pName, STableMeta** pMeta) {
char fullName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pName, fullName);
*pMeta = taosHashGet(pMetaCache->pTableMeta, fullName, strlen(fullName));
return NULL == *pMeta ? TSDB_CODE_PAR_INTERNAL_ERROR : TSDB_CODE_SUCCESS;
}
int32_t getDBVgInfoFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SArray** pVgInfo) {
*pVgInfo = taosHashGet(pMetaCache->pDbVgroup, pDbFName, strlen(pDbFName));
return NULL == *pVgInfo ? TSDB_CODE_PAR_INTERNAL_ERROR : TSDB_CODE_SUCCESS;
}
int32_t getTableHashVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName, SVgroupInfo* pVgroup) {
char fullName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pName, fullName);
SVgroupInfo* pInfo = taosHashGet(pMetaCache->pTableVgroup, fullName, strlen(fullName));
if (NULL == pInfo) {
STableMeta** pRes = taosHashGet(pMetaCache->pTableMeta, fullName, strlen(fullName));
if (NULL == pRes || NULL == *pRes) {
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
memcpy(pVgroup, pInfo, sizeof(SVgroupInfo));
*pMeta = tableMetaDup(*pRes);
if (NULL == *pMeta) {
return TSDB_CODE_OUT_OF_MEMORY;
}
return TSDB_CODE_SUCCESS;
}
int32_t getDBVgVersionFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, int32_t* pVersion, int64_t* pDbId,
static int32_t reserveDbReqInCache(int32_t acctId, const char* pDb, SHashObj** pDbs) {
if (NULL == *pDbs) {
*pDbs = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BINARY), true, HASH_NO_LOCK);
if (NULL == *pDbs) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
char fullName[TSDB_TABLE_FNAME_LEN];
int32_t len = snprintf(fullName, sizeof(fullName), "%d.%s", acctId, pDb);
return taosHashPut(*pDbs, fullName, len, &len, POINTER_BYTES);
}
int32_t reserveDbVgInfoInCache(int32_t acctId, const char* pDb, SParseMetaCache* pMetaCache) {
return reserveDbReqInCache(acctId, pDb, &pMetaCache->pDbVgroup);
}
int32_t getDbVgInfoFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SArray** pVgInfo) {
SArray** pRes = taosHashGet(pMetaCache->pDbVgroup, pDbFName, strlen(pDbFName));
if (NULL == pRes) {
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
// *pRes is null, which is a legal value, indicating that the user DB has not been created
if (NULL != *pRes) {
*pVgInfo = taosArrayDup(*pRes);
if (NULL == *pVgInfo) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
return TSDB_CODE_SUCCESS;
}
int32_t reserveTableVgroupInCache(int32_t acctId, const char* pDb, const char* pTable, SParseMetaCache* pMetaCache) {
return reserveTableReqInCache(acctId, pDb, pTable, &pMetaCache->pTableVgroup);
}
int32_t getTableVgroupFromCache(SParseMetaCache* pMetaCache, const SName* pName, SVgroupInfo* pVgroup) {
char fullName[TSDB_TABLE_FNAME_LEN];
tNameExtractFullName(pName, fullName);
SVgroupInfo** pRes = taosHashGet(pMetaCache->pTableVgroup, fullName, strlen(fullName));
if (NULL == pRes || NULL == *pRes) {
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
memcpy(pVgroup, *pRes, sizeof(SVgroupInfo));
return TSDB_CODE_SUCCESS;
}
int32_t getDbVgVersionFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, int32_t* pVersion, int64_t* pDbId,
int32_t* pTableNum) {
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
int32_t getDBCfgFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SDbCfgInfo* pInfo) {
SDbCfgInfo* pDbCfg = taosHashGet(pMetaCache->pDbCfg, pDbFName, strlen(pDbFName));
if (NULL == pDbCfg) {
int32_t reserveDbCfgInCache(int32_t acctId, const char* pDb, SParseMetaCache* pMetaCache) {
return reserveDbReqInCache(acctId, pDb, &pMetaCache->pDbCfg);
}
int32_t getDbCfgFromCache(SParseMetaCache* pMetaCache, const char* pDbFName, SDbCfgInfo* pInfo) {
SDbCfgInfo** pRes = taosHashGet(pMetaCache->pDbCfg, pDbFName, strlen(pDbFName));
if (NULL == pRes || NULL == *pRes) {
return TSDB_CODE_PAR_INTERNAL_ERROR;
}
memcpy(pInfo, pDbCfg, sizeof(SDbCfgInfo));
memcpy(pInfo, *pRes, sizeof(SDbCfgInfo));
return TSDB_CODE_SUCCESS;
}

10
source/libs/parser/src/parser.c
View File

@ -59,6 +59,14 @@ static int32_t parseSqlIntoAst(SParseContext* pCxt, SQuery** pQuery) {
return code;
}
static int32_t syntaxParseSql(SParseContext* pCxt, SQuery** pQuery) {
int32_t code = parse(pCxt, pQuery);
if (TSDB_CODE_SUCCESS == code) {
code = collectMetaKey(pCxt, *pQuery);
}
return code;
}
static int32_t setValueByBindParam(SValueNode* pVal, TAOS_MULTI_BIND* pParam) {
if (pParam->is_null && 1 == *(pParam->is_null)) {
pVal->node.resType.type = TSDB_DATA_TYPE_NULL;
@ -188,7 +196,7 @@ int32_t qSyntaxParseSql(SParseContext* pCxt, SQuery** pQuery, struct SCatalogReq
if (qIsInsertSql(pCxt->pSql, pCxt->sqlLen)) {
// todo insert sql
} else {
code = parse(pCxt, pQuery);
code = syntaxParseSql(pCxt, pQuery);
}
if (TSDB_CODE_SUCCESS == code) {
code = buildCatalogReq((*pQuery)->pMetaCache, pCatalogReq);

40
source/libs/parser/test/mockCatalogService.cpp
View File

@ -125,6 +125,12 @@ class MockCatalogServiceImpl {
if (TSDB_CODE_SUCCESS == code) {
code = getAllTableVgroup(pCatalogReq->pTableHash, &pMetaData->pTableHash);
}
if (TSDB_CODE_SUCCESS == code) {
code = getAllDbVgroup(pCatalogReq->pDbVgroup, &pMetaData->pDbVgroup);
}
if (TSDB_CODE_SUCCESS == code) {
code = getAllDbCfg(pCatalogReq->pDbCfg, &pMetaData->pDbCfg);
}
return code;
}
@ -330,12 +336,12 @@ class MockCatalogServiceImpl {
int32_t code = TSDB_CODE_SUCCESS;
if (NULL != pTableVgroupReq) {
int32_t ntables = taosArrayGetSize(pTableVgroupReq);
*pTableVgroupData = taosArrayInit(ntables, POINTER_BYTES);
*pTableVgroupData = taosArrayInit(ntables, sizeof(SVgroupInfo));
for (int32_t i = 0; i < ntables; ++i) {
SVgroupInfo* pVgInfo = (SVgroupInfo*)taosMemoryCalloc(1, sizeof(SVgroupInfo));
code = catalogGetTableHashVgroup((const SName*)taosArrayGet(pTableVgroupReq, i), pVgInfo);
SVgroupInfo vgInfo = {0};
code = catalogGetTableHashVgroup((const SName*)taosArrayGet(pTableVgroupReq, i), &vgInfo);
if (TSDB_CODE_SUCCESS == code) {
taosArrayPush(*pTableVgroupData, &pVgInfo);
taosArrayPush(*pTableVgroupData, &vgInfo);
} else {
break;
}
@ -344,6 +350,32 @@ class MockCatalogServiceImpl {
return code;
}
int32_t getAllDbVgroup(SArray* pDbVgroupReq, SArray** pDbVgroupData) const {
int32_t code = TSDB_CODE_SUCCESS;
if (NULL != pDbVgroupReq) {
int32_t ndbs = taosArrayGetSize(pDbVgroupReq);
*pDbVgroupData = taosArrayInit(ndbs, POINTER_BYTES);
for (int32_t i = 0; i < ndbs; ++i) {
int64_t zeroVg = 0;
taosArrayPush(*pDbVgroupData, &zeroVg);
}
}
return code;
}
int32_t getAllDbCfg(SArray* pDbCfgReq, SArray** pDbCfgData) const {
int32_t code = TSDB_CODE_SUCCESS;
if (NULL != pDbCfgReq) {
int32_t ndbs = taosArrayGetSize(pDbCfgReq);
*pDbCfgData = taosArrayInit(ndbs, sizeof(SDbCfgInfo));
for (int32_t i = 0; i < ndbs; ++i) {
SDbCfgInfo dbCfg = {0};
taosArrayPush(*pDbCfgData, &dbCfg);
}
}
return code;
}
uint64_t id_;
std::unique_ptr<TableBuilder> builder_;
DbMetaCache meta_;

44
source/libs/parser/test/parTestMain.cpp
View File

@ -37,6 +37,7 @@ class ParserEnv : public testing::Environment {
virtual void SetUp() {
initMetaDataEnv();
generateMetaData();
initLog(TD_TMP_DIR_PATH "td");
}
virtual void TearDown() {
@ -47,20 +48,55 @@ class ParserEnv : public testing::Environment {
ParserEnv() {}
virtual ~ParserEnv() {}
private:
void initLog(const char* path) {
int32_t logLevel = getLogLevel();
dDebugFlag = logLevel;
vDebugFlag = logLevel;
mDebugFlag = logLevel;
cDebugFlag = logLevel;
jniDebugFlag = logLevel;
tmrDebugFlag = logLevel;
uDebugFlag = logLevel;
rpcDebugFlag = logLevel;
qDebugFlag = logLevel;
wDebugFlag = logLevel;
sDebugFlag = logLevel;
tsdbDebugFlag = logLevel;
tsLogEmbedded = 1;
tsAsyncLog = 0;
taosRemoveDir(path);
taosMkDir(path);
tstrncpy(tsLogDir, path, PATH_MAX);
if (taosInitLog("taoslog", 1) != 0) {
std::cout << "failed to init log file" << std::endl;
}
}
};
static void parseArg(int argc, char* argv[]) {
int opt = 0;
const char* optstring = "";
int opt = 0;
const char* optstring = "";
// clang-format off
static struct option long_options[] = {
{"dump", no_argument, NULL, 'd'}, {"async", no_argument, NULL, 'a'}, {0, 0, 0, 0}};
{"dump", no_argument, NULL, 'd'},
{"async", required_argument, NULL, 'a'},
{"skipSql", required_argument, NULL, 's'},
{0, 0, 0, 0}
};
// clang-format on
while ((opt = getopt_long(argc, argv, optstring, long_options, NULL)) != -1) {
switch (opt) {
case 'd':
g_dump = true;
break;
case 'a':
g_testAsyncApis = true;
setAsyncFlag(optarg);
break;
case 's':
setSkipSqlNum(optarg);
break;
default:
break;

133
source/libs/parser/test/parTestUtil.cpp
View File

@ -44,13 +44,22 @@ namespace ParserTest {
} \
} while (0);
bool g_dump = false;
bool g_testAsyncApis = false;
bool g_dump = false;
bool g_testAsyncApis = false;
int32_t g_logLevel = 131;
int32_t g_skipSql = 0;
void setAsyncFlag(const char* pFlag) { g_testAsyncApis = stoi(pFlag) > 0 ? true : false; }
void setSkipSqlNum(const char* pNum) { g_skipSql = stoi(optarg); }
struct TerminateFlag : public exception {
const char* what() const throw() { return "success and terminate"; }
};
void setLogLevel(const char* pLogLevel) { g_logLevel = stoi(pLogLevel); }
int32_t getLogLevel() { return g_logLevel; }
class ParserTestBaseImpl {
public:
ParserTestBaseImpl(ParserTestBase* pBase) : pBase_(pBase) {}
@ -58,9 +67,15 @@ class ParserTestBaseImpl {
void useDb(const string& acctId, const string& db) {
caseEnv_.acctId_ = acctId;
caseEnv_.db_ = db;
caseEnv_.nsql_ = g_skipSql;
}
void run(const string& sql, int32_t expect, ParserStage checkStage) {
if (caseEnv_.nsql_ > 0) {
--(caseEnv_.nsql_);
return;
}
reset(expect, checkStage);
try {
SParseContext cxt = {0};
@ -89,59 +104,11 @@ class ParserTestBaseImpl {
}
}
void runAsync(const string& sql, int32_t expect, ParserStage checkStage) {
reset(expect, checkStage);
try {
SParseContext cxt = {0};
setParseContext(sql, &cxt, true);
SQuery* pQuery = nullptr;
doParse(&cxt, &pQuery);
SCatalogReq catalogReq = {0};
doBuildCatalogReq(pQuery->pMetaCache, &catalogReq);
string err;
thread t1([&]() {
try {
SMetaData metaData = {0};
doGetAllMeta(&catalogReq, &metaData);
doPutMetaDataToCache(&catalogReq, &metaData, pQuery->pMetaCache);
doTranslate(&cxt, pQuery);
doCalculateConstant(&cxt, pQuery);
} catch (const TerminateFlag& e) {
// success and terminate
} catch (const runtime_error& e) {
err = e.what();
} catch (...) {
err = "unknown error";
}
});
t1.join();
if (!err.empty()) {
throw runtime_error(err);
}
if (g_dump) {
dump();
}
} catch (const TerminateFlag& e) {
// success and terminate
return;
} catch (...) {
dump();
throw;
}
}
private:
struct caseEnv {
string acctId_;
string db_;
string acctId_;
string db_;
int32_t nsql_;
};
struct stmtEnv {
@ -220,6 +187,11 @@ class ParserTestBaseImpl {
res_.parsedAst_ = toString((*pQuery)->pRoot);
}
void doCollectMetaKey(SParseContext* pCxt, SQuery* pQuery) {
DO_WITH_THROW(collectMetaKey, pCxt, pQuery);
ASSERT_NE(pQuery->pMetaCache, nullptr);
}
void doBuildCatalogReq(const SParseMetaCache* pMetaCache, SCatalogReq* pCatalogReq) {
DO_WITH_THROW(buildCatalogReq, pMetaCache, pCatalogReq);
}
@ -254,6 +226,57 @@ class ParserTestBaseImpl {
void checkQuery(const SQuery* pQuery, ParserStage stage) { pBase_->checkDdl(pQuery, stage); }
void runAsync(const string& sql, int32_t expect, ParserStage checkStage) {
reset(expect, checkStage);
try {
SParseContext cxt = {0};
setParseContext(sql, &cxt, true);
SQuery* pQuery = nullptr;
doParse(&cxt, &pQuery);
doCollectMetaKey(&cxt, pQuery);
SCatalogReq catalogReq = {0};
doBuildCatalogReq(pQuery->pMetaCache, &catalogReq);
string err;
thread t1([&]() {
try {
SMetaData metaData = {0};
doGetAllMeta(&catalogReq, &metaData);
doPutMetaDataToCache(&catalogReq, &metaData, pQuery->pMetaCache);
doTranslate(&cxt, pQuery);
doCalculateConstant(&cxt, pQuery);
} catch (const TerminateFlag& e) {
// success and terminate
} catch (const runtime_error& e) {
err = e.what();
} catch (...) {
err = "unknown error";
}
});
t1.join();
if (!err.empty()) {
throw runtime_error(err);
}
if (g_dump) {
dump();
}
} catch (const TerminateFlag& e) {
// success and terminate
return;
} catch (...) {
dump();
throw;
}
}
caseEnv caseEnv_;
stmtEnv stmtEnv_;
stmtRes res_;
@ -270,10 +293,6 @@ void ParserTestBase::run(const std::string& sql, int32_t expect, ParserStage che
return impl_->run(sql, expect, checkStage);
}
void ParserTestBase::runAsync(const std::string& sql, int32_t expect, ParserStage checkStage) {
return impl_->runAsync(sql, expect, checkStage);
}
void ParserTestBase::checkDdl(const SQuery* pQuery, ParserStage stage) { return; }
} // namespace ParserTest

7
source/libs/parser/test/parTestUtil.h
View File

@ -36,7 +36,6 @@ class ParserTestBase : public testing::Test {
void useDb(const std::string& acctId, const std::string& db);
void run(const std::string& sql, int32_t expect = TSDB_CODE_SUCCESS, ParserStage checkStage = PARSER_STAGE_ALL);
void runAsync(const std::string& sql, int32_t expect = TSDB_CODE_SUCCESS, ParserStage checkStage = PARSER_STAGE_ALL);
virtual void checkDdl(const SQuery* pQuery, ParserStage stage);
@ -65,7 +64,11 @@ class ParserDdlTest : public ParserTestBase {
};
extern bool g_dump;
extern bool g_testAsyncApis;
extern void setAsyncFlag(const char* pFlag);
extern void setLogLevel(const char* pLogLevel);
extern int32_t getLogLevel();
extern void setSkipSqlNum(const char* pNum);
} // namespace ParserTest