Skip to content

T
TDengine

Mr、小李 / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine

体验新版 GitCode，发现更多精彩内容 >>

提交 f2016993 编写于 作者: H hjxilinx
浏览文件
操作

add the union support in sql parser: fix some bugs. #1032. [TBASE-1140]

上级 c9827141
展开全部 显示空白变更内容
内联 并排
Showing with 589 addition and 531 deletion
src/client/inc/tscUtil.h
浏览文件 @ f2016993
......@@ -110,7 +110,7 @@ void tscAddSpecialColumnForSelect(SQueryInfo* pQueryInfo, int32_t outputColIndex
void addRequiredTagColumn(SQueryInfo* pQueryInfo, int32_t tagColIndex, int32_t tableIndex);
int32_t setMeterID(SSqlObj* pSql, int32_t subClauseIndex, SSQLToken* pzTableName, int32_t tableIndex);
int32_t setMeterID(SMeterMetaInfo* pMeterMetaInfo, SSQLToken* pzTableName, SSqlObj* pSql);
void tscClearInterpInfo(SQueryInfo* pQueryInfo);
bool tscIsInsertOrImportData(char* sqlstr);
......@@ -198,8 +198,8 @@ void tscClearSubqueryInfo(SSqlCmd* pCmd);
void tscGetMetricMetaCacheKey(SSqlCmd* pCmd, int32_t subClauseIndex, char* keyStr, uint64_t uid);
int tscGetMetricMeta(SSqlObj* pSql);
int tscGetMeterMeta(SSqlObj* pSql, char* meterId, int32_t tableIndex);
int tscGetMeterMetaEx(SSqlObj* pSql, char* meterId, bool createIfNotExists);
int tscGetMeterMeta(SSqlObj* pSql, SMeterMetaInfo* pMeterMetaInfo);
int tscGetMeterMetaEx(SSqlObj* pSql, SMeterMetaInfo* pMeterMetaInfo, bool createIfNotExists);
void tscResetForNextRetrieve(SSqlRes* pRes);
......
src/client/inc/tsclient.h
浏览文件 @ f2016993
......@@ -185,7 +185,7 @@ typedef struct STableDataBlocks {
* the metermeta for current table, the metermeta will be used during submit stage, keep a ref
* to avoid it to be removed from cache
*/
SMeterMeta* pMeterMeta;
SMeterMeta *pMeterMeta;
union {
char *filename;
......@@ -208,7 +208,7 @@ typedef struct SDataBlockList {
} SDataBlockList;
typedef struct SQueryInfo {
uint16_t type; // query type
uint16_t type; // query/insert/import type
char intervalTimeUnit;
int64_t etime, stime;
......@@ -227,60 +227,42 @@ typedef struct SQueryInfo {
int16_t numOfTables;
SMeterMetaInfo **pMeterInfo;
struct STSBuf * tsBuf;
// todo use dynamic allocated memory for defaultVal
int64_t defaultVal[TSDB_MAX_COLUMNS]; // default value for interpolation
char* msg; // pointer to the pCmd->payload to keep error message temporarily
int64_t * defaultVal; // default value for interpolation
char * msg; // pointer to the pCmd->payload to keep error message temporarily
} SQueryInfo;
// data source from sql string or from file
enum {
DATA_FROM_SQL_STRING = 1,
DATA_FROM_DATA_FILE = 2,
};
typedef struct {
// SOrderVal order;
int command;
int count; // TODO refactor
uint8_t msgType;
union {
bool existsCheck; // check if the table exists
bool import; // import/insert type
bool existsCheck; // check if the table exists or not
bool inStream; // denote if current sql is executed in stream or not
bool createOnDemand; // if the table is missing, on-the-fly create it. during getmeterMeta
int8_t dataSourceType; // load data from file or not
};
int8_t isInsertFromFile; // load data from file or not
uint8_t msgType;
union {
int32_t count;
int32_t numOfTablesInSubmit;
};
/*
* use to keep short request msg and error msg, in such case, SSqlCmd->payload == SSqlCmd->ext;
* create table/query/insert operations will exceed the TSDB_SQLCMD_SIZE.
*
* In such cases, allocate the memory dynamically, and need to free the memory
*/
short numOfCols;
uint32_t allocSize;
char * payload;
int payloadLen;
short numOfCols;
int64_t globalLimit;
SQueryInfo **pQueryInfo;
int32_t numOfClause;
// char intervalTimeUnit;
// int64_t etime, stime;
// int64_t nAggTimeInterval; // aggregation time interval
// int64_t nSlidingTime; // sliding window in mseconds
// SSqlGroupbyExpr groupbyExpr; // group by tags info
//
// SColumnBaseInfo colList;
// SFieldInfo fieldsInfo;
// SSqlExprInfo exprsInfo;
// SLimitVal limit;
// SLimitVal slimit;
// STagCond tagCond;
// int16_t interpoType; // interpolate type
// int16_t numOfTables;
// SMeterMetaInfo **pMeterInfo;
// struct STSBuf * tsBuf;
// // todo use dynamic allocated memory for defaultVal
// int64_t defaultVal[TSDB_MAX_COLUMNS]; // default value for interpolation
// submit data blocks branched according to vnode
SDataBlockList * pDataBlocks;
SDataBlockList *pDataBlocks;
// for parameter ('?') binding and batch processing
int32_t batchSize;
......@@ -359,8 +341,8 @@ typedef struct _sql_obj {
SSqlCmd cmd;
SSqlRes res;
uint8_t numOfSubs;
char* asyncTblPos;
void* pTableHashList;
char * asyncTblPos;
void * pTableHashList;
struct _sql_obj **pSubs;
struct _sql_obj * prev, *next;
} SSqlObj;
......@@ -423,12 +405,12 @@ int taos_retrieve(TAOS_RES *res);
* transfer function for metric query in stream computing, the function need to be change
* before send query message to vnode
*/
int32_t tscTansformSQLFunctionForMetricQuery(SQueryInfo* pQueryInfo);
void tscRestoreSQLFunctionForMetricQuery(SQueryInfo* pQueryInfo);
int32_t tscTansformSQLFunctionForSTableQuery(SQueryInfo *pQueryInfo);
void tscRestoreSQLFunctionForMetricQuery(SQueryInfo *pQueryInfo);
void tscClearSqlMetaInfoForce(SSqlCmd *pCmd);
int32_t tscCreateResPointerInfo(SQueryInfo* pQueryInfo, SSqlRes *pRes);
int32_t tscCreateResPointerInfo(SQueryInfo *pQueryInfo, SSqlRes *pRes);
void tscDestroyResPointerInfo(SSqlRes *pRes);
void tscFreeSqlCmdData(SSqlCmd *pCmd);
......@@ -450,13 +432,13 @@ void tscFreeSqlObj(SSqlObj *pObj);
void tscCloseTscObj(STscObj *pObj);
void tscProcessMultiVnodesInsert(SSqlObj *pSql);
void tscProcessMultiVnodesInsertForFile(SSqlObj *pSql);
void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql);
void tscKillMetricQuery(SSqlObj *pSql);
void tscInitResObjForLocalQuery(SSqlObj *pObj, int32_t numOfRes, int32_t rowLen);
bool tscIsUpdateQuery(STscObj *pObj);
bool tscHasReachLimitation(SSqlObj* pSql);
bool tscHasReachLimitation(SSqlObj *pSql);
char* tscGetErrorMsgPayload(SSqlCmd* pCmd);
char *tscGetErrorMsgPayload(SSqlCmd *pCmd);
int32_t tscInvalidSQLErrMsg(char *msg, const char *additionalInfo, const char *sql);
......
src/client/src/tscAsync.c
浏览文件 @ f2016993
......@@ -413,13 +413,13 @@ void tscAsyncInsertMultiVnodesProxy(void *param, TAOS_RES *tres, int numOfRows)
SSqlCmd *pCmd = &pSql->cmd;
int32_t code = TSDB_CODE_SUCCESS;
assert(!pCmd->isInsertFromFile && pSql->signature == pSql);
assert(pCmd->dataSourceType != 0 && pSql->signature == pSql);
int32_t index = 0;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, index);
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, index, 0);
assert(pQueryInfo->numOfTables == 1);
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0);
assert(pQueryInfo->numOfTables == 1 || pQueryInfo->numOfTables == 2);
SDataBlockList *pDataBlocks = pCmd->pDataBlocks;
if (pDataBlocks == NULL || pMeterMetaInfo->vnodeIndex >= pDataBlocks->nSize) {
......@@ -456,7 +456,6 @@ void tscMeterMetaCallBack(void *param, TAOS_RES *res, int code) {
SSqlObj *pSql = (SSqlObj *)param;
if (pSql == NULL || pSql->signature != pSql) return;
STscObj *pObj = pSql->pTscObj;
SSqlCmd *pCmd = &pSql->cmd;
SSqlRes *pRes = &pSql->res;
......@@ -480,7 +479,7 @@ void tscMeterMetaCallBack(void *param, TAOS_RES *res, int code) {
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfo(&pSql->cmd, 0, 0);
assert(pMeterMetaInfo->pMeterMeta == NULL);
tscGetMeterMeta(pSql, pMeterMetaInfo->name, 0);
tscGetMeterMeta(pSql, pMeterMetaInfo);
code = tscSendMsgToServer(pSql);
if (code != 0) {
pRes->code = code;
......@@ -513,7 +512,7 @@ void tscMeterMetaCallBack(void *param, TAOS_RES *res, int code) {
tscTrace("%p get metricMeta during metric query successfully", pSql);
code = tscGetMeterMeta(pSql, pMeterMetaInfo->name, 0);
code = tscGetMeterMeta(pSql, pMeterMetaInfo);
pRes->code = code;
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
......@@ -529,12 +528,11 @@ void tscMeterMetaCallBack(void *param, TAOS_RES *res, int code) {
} else { // stream computing
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, 0, 0);
code = tscGetMeterMeta(pSql, pMeterMetaInfo->name, 0);
code = tscGetMeterMeta(pSql, pMeterMetaInfo);
pRes->code = code;
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, 0, 0);
if (code == TSDB_CODE_SUCCESS && UTIL_METER_IS_SUPERTABLE(pMeterMetaInfo)) {
code = tscGetMetricMeta(pSql);
pRes->code = code;
......@@ -557,7 +555,7 @@ void tscMeterMetaCallBack(void *param, TAOS_RES *res, int code) {
*/
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
tscTansformSQLFunctionForMetricQuery(pQueryInfo);
tscTansformSQLFunctionForSTableQuery(pQueryInfo);
tscIncStreamExecutionCount(pSql->pStream);
} else {
tscTrace("%p get meterMeta/metricMeta successfully", pSql);
......
src/client/src/tscParseInsert.c
浏览文件 @ f2016993
......@@ -18,9 +18,8 @@
#define _XOPEN_SOURCE
#include <hash.h>
#include "os.h"
#include "ihash.h"
#include "os.h"
#include "tscSecondaryMerge.h"
#include "tscUtil.h"
#include "tschemautil.h"
......@@ -72,8 +71,6 @@ static int32_t tscToDouble(SSQLToken *pToken, double *value, char **endPtr) {
}
int tsParseTime(SSQLToken *pToken, int64_t *time, char **next, char *error, int16_t timePrec) {
//char * token; //fang not used
//int tokenlen; //fang not used
int32_t index = 0;
SSQLToken sToken;
int64_t interval;
......@@ -118,7 +115,6 @@ int tsParseTime(SSQLToken *pToken, int64_t *time, char **next, char *error, int1
pTokenEnd += index;
if (sToken.type == TK_MINUS || sToken.type == TK_PLUS) {
index = 0;
valueToken = tStrGetToken(pTokenEnd, &index, false, 0, NULL);
pTokenEnd += index;
......@@ -194,7 +190,7 @@ int32_t tsParseOneColumnData(SSchema *pSchema, SSQLToken *pToken, char *payload,
return tscInvalidSQLErrMsg(msg, "tinyint data overflow", pToken->z);
}
*((int8_t *)payload) = (int8_t) iv;
*((int8_t *)payload) = (int8_t)iv;
}
break;
......@@ -390,9 +386,9 @@ static int32_t tsCheckTimestamp(STableDataBlocks *pDataBlocks, const char *start
}
int tsParseOneRowData(char **str, STableDataBlocks *pDataBlocks, SSchema schema[], SParsedDataColInfo *spd, char *error,
int16_t timePrec, int32_t *code, char* tmpTokenBuf) {
int16_t timePrec, int32_t *code, char *tmpTokenBuf) {
int32_t index = 0;
//bool isPrevOptr; //fang, never used
// bool isPrevOptr; //fang, never used
SSQLToken sToken = {0};
char * payload = pDataBlocks->pData + pDataBlocks->size;
......@@ -497,7 +493,7 @@ static int32_t rowDataCompar(const void *lhs, const void *rhs) {
}
int tsParseValues(char **str, STableDataBlocks *pDataBlock, SMeterMeta *pMeterMeta, int maxRows,
SParsedDataColInfo *spd, char *error, int32_t *code, char* tmpTokenBuf) {
SParsedDataColInfo *spd, char *error, int32_t *code, char *tmpTokenBuf) {
int32_t index = 0;
SSQLToken sToken;
......@@ -520,7 +516,7 @@ int tsParseValues(char **str, STableDataBlocks *pDataBlock, SMeterMeta *pMeterMe
*str += index;
if (numOfRows >= maxRows || pDataBlock->size + pMeterMeta->rowSize >= pDataBlock->nAllocSize) {
int32_t tSize = tscAllocateMemIfNeed(pDataBlock, pMeterMeta->rowSize);
if (0 == tSize) { //TODO pass the correct error code to client
if (0 == tSize) { // TODO pass the correct error code to client
strcpy(error, "client out of memory");
*code = TSDB_CODE_CLI_OUT_OF_MEMORY;
return -1;
......@@ -588,7 +584,7 @@ int32_t tscAllocateMemIfNeed(STableDataBlocks *pDataBlock, int32_t rowSize) {
pDataBlock->pData = tmp;
memset(pDataBlock->pData + pDataBlock->size, 0, pDataBlock->nAllocSize - pDataBlock->size);
} else {
//assert(false);
// assert(false);
// do nothing
pDataBlock->nAllocSize = nAllocSizeOld;
return 0;
......@@ -667,7 +663,7 @@ static int32_t doParseInsertStatement(SSqlObj *pSql, void *pTableHashList, char
}
int32_t code = TSDB_CODE_INVALID_SQL;
char* tmpTokenBuf = calloc(1, 4096); // used for deleting Escape character: \\, \', \"
char * tmpTokenBuf = calloc(1, 4096); // used for deleting Escape character: \\, \', \"
if (NULL == tmpTokenBuf) {
return TSDB_CODE_CLI_OUT_OF_MEMORY;
}
......@@ -679,7 +675,7 @@ static int32_t doParseInsertStatement(SSqlObj *pSql, void *pTableHashList, char
}
for (uint32_t i = 0; i < dataBuf->numOfParams; ++i) {
SParamInfo* param = dataBuf->params + i;
SParamInfo *param = dataBuf->params + i;
if (param->idx == -1) {
param->idx = pCmd->numOfParams++;
param->offset -= sizeof(SShellSubmitBlock);
......@@ -700,16 +696,20 @@ static int32_t doParseInsertStatement(SSqlObj *pSql, void *pTableHashList, char
return TSDB_CODE_SUCCESS;
}
static int32_t tscParseSqlForCreateTableOnDemand(char **sqlstr, SSqlObj *pSql) {
static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
int32_t index = 0;
SSQLToken sToken;
SSQLToken tableToken;
SSQLToken sToken = {0};
SSQLToken tableToken = {0};
int32_t code = TSDB_CODE_SUCCESS;
const int32_t TABLE_INDEX = 0;
const int32_t STABLE_INDEX = 1;
SSqlCmd * pCmd = &pSql->cmd;
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, 0, 0);
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
char *sql = *sqlstr;
// get the token of specified table
index = 0;
tableToken = tStrGetToken(sql, &index, false, 0, NULL);
......@@ -747,40 +747,53 @@ static int32_t tscParseSqlForCreateTableOnDemand(char **sqlstr, SSqlObj *pSql) {
return TSDB_CODE_INVALID_SQL;
}
if (sToken.type == TK_USING) { // create table if not exists
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, TABLE_INDEX);
if (sToken.type == TK_USING) { // create table if not exists according to the super table
index = 0;
sToken = tStrGetToken(sql, &index, false, 0, NULL);
sql += index;
STagData *pTag = (STagData *)pCmd->payload;
memset(pTag, 0, sizeof(STagData));
setMeterID(pSql, 0, &sToken, 0);
strncpy(pTag->name, pMeterMetaInfo->name, TSDB_METER_ID_LEN);
code = tscGetMeterMeta(pSql, pTag->name, 0);
/*
* the source super table is moved to the secondary position of the pMeterMetaInfo list
*/
if (pQueryInfo->numOfTables < 2) {
tscAddEmptyMeterMetaInfo(pQueryInfo);
}
SMeterMetaInfo *pSTableMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, STABLE_INDEX);
setMeterID(pSTableMeterMetaInfo, &sToken, pSql);
strncpy(pTag->name, pSTableMeterMetaInfo->name, TSDB_METER_ID_LEN);
code = tscGetMeterMeta(pSql, pSTableMeterMetaInfo);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
if (!UTIL_METER_IS_SUPERTABLE(pMeterMetaInfo)) {
if (!UTIL_METER_IS_SUPERTABLE(pSTableMeterMetaInfo)) {
return tscInvalidSQLErrMsg(pCmd->payload, "create table only from super table is allowed", sToken.z);
}
SSchema *pTagSchema = tsGetTagSchema(pMeterMetaInfo->pMeterMeta);
SSchema *pTagSchema = tsGetTagSchema(pSTableMeterMetaInfo->pMeterMeta);
index = 0;
sToken = tStrGetToken(sql, &index, false, 0, NULL);
sql += index;
SParsedDataColInfo spd = {0};
uint8_t numOfTags = pMeterMetaInfo->pMeterMeta->numOfTags;
uint8_t numOfTags = pSTableMeterMetaInfo->pMeterMeta->numOfTags;
spd.numOfCols = numOfTags;
// if specify some tags column
if (sToken.type != TK_LP) {
tscSetAssignedColumnInfo(&spd, pTagSchema, numOfTags);
} else {
/* insert into tablename (col1, col2,..., coln) using superTableName (tagName1, tagName2, ..., tagNamen) tags(tagVal1, tagVal2, ..., tagValn) values(v1, v2,... vn); */
/* insert into tablename (col1, col2,..., coln) using superTableName (tagName1, tagName2, ..., tagNamen)
* tags(tagVal1, tagVal2, ..., tagValn) values(v1, v2,... vn); */
int16_t offset[TSDB_MAX_COLUMNS] = {0};
for (int32_t t = 1; t < numOfTags; ++t) {
offset[t] = offset[t - 1] + pTagSchema[t - 1].bytes;
......@@ -841,7 +854,7 @@ static int32_t tscParseSqlForCreateTableOnDemand(char **sqlstr, SSqlObj *pSql) {
uint32_t ignoreTokenTypes = TK_LP;
uint32_t numOfIgnoreToken = 1;
for (int i = 0; i < spd.numOfAssignedCols; ++i) {
char* tagVal = pTag->data + spd.elems[i].offset;
char * tagVal = pTag->data + spd.elems[i].offset;
int16_t colIndex = spd.elems[i].colIndex;
index = 0;
......@@ -859,13 +872,14 @@ static int32_t tscParseSqlForCreateTableOnDemand(char **sqlstr, SSqlObj *pSql) {
sToken.n -= 2;
}
code = tsParseOneColumnData(&pTagSchema[colIndex], &sToken, tagVal, pCmd->payload, &sql, false, pMeterMetaInfo->pMeterMeta->precision);
code = tsParseOneColumnData(&pTagSchema[colIndex], &sToken, tagVal, pCmd->payload, &sql, false,
pSTableMeterMetaInfo->pMeterMeta->precision);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
if ((pTagSchema[colIndex].type == TSDB_DATA_TYPE_BINARY ||
pTagSchema[colIndex].type == TSDB_DATA_TYPE_NCHAR) && sToken.n > pTagSchema[colIndex].bytes) {
if ((pTagSchema[colIndex].type == TSDB_DATA_TYPE_BINARY || pTagSchema[colIndex].type == TSDB_DATA_TYPE_NCHAR) &&
sToken.n > pTagSchema[colIndex].bytes) {
return tscInvalidSQLErrMsg(pCmd->payload, "string too long", sToken.z);
}
}
......@@ -894,20 +908,20 @@ static int32_t tscParseSqlForCreateTableOnDemand(char **sqlstr, SSqlObj *pSql) {
return tscInvalidSQLErrMsg(pCmd->payload, "invalid table name", *sqlstr);
}
int32_t ret = setMeterID(pSql, 0, &tableToken, 0);
int32_t ret = setMeterID(pMeterMetaInfo, &tableToken, pSql);
if (ret != TSDB_CODE_SUCCESS) {
return ret;
}
createTable = true;
code = tscGetMeterMetaEx(pSql, pMeterMetaInfo->name, true);
code = tscGetMeterMetaEx(pSql, pMeterMetaInfo, true);
} else {
if (cstart != NULL) {
sql = cstart;
} else {
sql = sToken.z;
}
code = tscGetMeterMeta(pSql, pMeterMetaInfo->name, 0);
code = tscGetMeterMeta(pSql, pMeterMetaInfo);
}
int32_t len = cend - cstart + 1;
......@@ -932,6 +946,15 @@ int validateTableName(char *tblName, int len) {
return tscValidateName(&token);
}
static int32_t validateDataSource(SSqlCmd *pCmd, int8_t type, const char *sql) {
if (pCmd->dataSourceType != 0 && pCmd->dataSourceType != type) {
return tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES and FILE are not allowed to mix up", sql);
}
pCmd->dataSourceType = type;
return TSDB_CODE_SUCCESS;
}
/**
* usage: insert into table1 values() () table2 values()()
*
......@@ -945,16 +968,18 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
SSqlCmd *pCmd = &pSql->cmd;
int32_t totalNum = 0;
SQueryInfo* pQueryInfo = NULL;
SMeterMetaInfo* pMeterMetaInfo = NULL;
int32_t code = TSDB_CODE_SUCCESS;
int32_t code = tscGetQueryInfoDetailSafely(pCmd, 0, &pQueryInfo);
SMeterMetaInfo *pMeterMetaInfo = NULL;
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
assert(pQueryInfo != NULL);
if (pQueryInfo->numOfTables == 0) {
pMeterMetaInfo = tscAddEmptyMeterMetaInfo(pQueryInfo);
} else {
pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0);
assert(pQueryInfo->numOfTables == 1);
// assert(pQueryInfo->numOfTables == 1);
}
if ((code = tscAllocPayload(pCmd, TSDB_PAYLOAD_SIZE)) != TSDB_CODE_SUCCESS) {
......@@ -978,16 +1003,26 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
while (1) {
int32_t index = 0;
SSQLToken sToken = tStrGetToken(str, &index, false, 0, NULL);
if (sToken.n == 0) { // parse file, do not release the STableDataBlock
if (pCmd->isInsertFromFile == 1) {
// no data in the sql string anymore.
if (sToken.n == 0) {
/*
* if the data is from the data file, no data has been generated yet. So, there no data to
* merge or submit, save the file path and parse the file in other routines.
*/
if (pCmd->dataSourceType == DATA_FROM_DATA_FILE) {
goto _clean;
}
if (totalNum > 0) {
break;
} else { // no data in current sql string, error
/*
* if no data has been generated during parsing the sql string, error msg will return
* Otherwise, create the first submit block and submit to virtual node.
*/
if (totalNum == 0) {
code = TSDB_CODE_INVALID_SQL;
goto _error_clean;
} else {
break;
}
}
......@@ -999,22 +1034,21 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
goto _error_clean;
}
//TODO refactor
if ((code = setMeterID(pSql, 0, &sToken, 0)) != TSDB_CODE_SUCCESS) {
if ((code = setMeterID(pMeterMetaInfo, &sToken, pSql)) != TSDB_CODE_SUCCESS) {
goto _error_clean;
}
void *fp = pSql->fp;
if ((code = tscParseSqlForCreateTableOnDemand(&str, pSql)) != TSDB_CODE_SUCCESS) {
if ((code = tscCheckIfCreateTable(&str, pSql)) != TSDB_CODE_SUCCESS) {
/*
* For async insert, after get the metermeta from server, the sql string will not be
* parsed using the new metermeta to avoid the overhead cause by get metermeta data information.
* And during the getMeterMetaCallback function, the sql string will be parsed from the
* interrupted position.
*/
if (fp != NULL) {
//goto _clean;
return code;
} else {
/*
* for async insert, the free data block operations, which is tscDestroyBlockArrayList,
* must be executed before launch another threads to get metermeta, since the
* later ops may manipulate SSqlObj through another thread in getMeterMetaCallback function.
*/
goto _error_clean;
}
}
......@@ -1027,8 +1061,9 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
index = 0;
sToken = tStrGetToken(str, &index, false, 0, NULL);
str += index;
if (sToken.n == 0) {
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES or FILE are required", sToken.z);
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES or FILE required", sToken.z);
goto _error_clean;
}
......@@ -1038,14 +1073,9 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
tscSetAssignedColumnInfo(&spd, pSchema, pMeterMetaInfo->pMeterMeta->numOfColumns);
if (pCmd->isInsertFromFile == -1) {
pCmd->isInsertFromFile = 0;
} else {
if (pCmd->isInsertFromFile == 1) {
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES and FILE are not allowed to mix up", sToken.z);
if (validateDataSource(pCmd, DATA_FROM_SQL_STRING, sToken.z) != TSDB_CODE_SUCCESS) {
goto _error_clean;
}
}
/*
* app here insert data in different vnodes, so we need to set the following
......@@ -1056,14 +1086,9 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
goto _error_clean;
}
} else if (sToken.type == TK_FILE) {
if (pCmd->isInsertFromFile == -1) {
pCmd->isInsertFromFile = 1;
} else {
if (pCmd->isInsertFromFile == 0) {
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES and FILE are not allowed to mix up", sToken.z);
if (validateDataSource(pCmd, DATA_FROM_DATA_FILE, sToken.z) != TSDB_CODE_SUCCESS) {
goto _error_clean;
}
}
index = 0;
sToken = tStrGetToken(str, &index, false, 0, NULL);
......@@ -1099,10 +1124,7 @@ int doParseInsertSql(SSqlObj *pSql, char *str) {
SMeterMeta *pMeterMeta = tscGetMeterMetaInfo(pCmd, 0, 0)->pMeterMeta;
SSchema * pSchema = tsGetSchema(pMeterMeta);
if (pCmd->isInsertFromFile == -1) {
pCmd->isInsertFromFile = 0;
} else if (pCmd->isInsertFromFile == 1) {
code = tscInvalidSQLErrMsg(pCmd->payload, "keyword VALUES and FILE are not allowed to mix up", sToken.z);
if (validateDataSource(pCmd, DATA_FROM_SQL_STRING, sToken.z) != TSDB_CODE_SUCCESS) {
goto _error_clean;
}
......@@ -1223,24 +1245,29 @@ int tsParseInsertSql(SSqlObj *pSql) {
int32_t index = 0;
SSqlCmd *pCmd = &pSql->cmd;
char* sql = pSql->sqlstr;
SSQLToken sToken = tStrGetToken(sql, &index, false, 0, NULL);
SSQLToken sToken = tStrGetToken(pSql->sqlstr, &index, false, 0, NULL);
assert(sToken.type == TK_INSERT || sToken.type == TK_IMPORT);
pCmd->import = (sToken.type == TK_IMPORT);
sToken = tStrGetToken(sql, &index, false, 0, NULL);
pCmd->count = 0;
pCmd->command = TSDB_SQL_INSERT;
SQueryInfo *pQueryInfo = NULL;
tscGetQueryInfoDetailSafely(pCmd, 0, &pQueryInfo);
if (sToken.type == TK_INSERT) {
TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_INSERT);
} else {
TSDB_QUERY_SET_TYPE(pQueryInfo->type, TSDB_QUERY_TYPE_IMPORT);
}
sToken = tStrGetToken(pSql->sqlstr, &index, false, 0, NULL);
if (sToken.type != TK_INTO) {
return tscInvalidSQLErrMsg(pCmd->payload, "keyword INTO is expected", sToken.z);
}
pCmd->count = 0;
pCmd->command = TSDB_SQL_INSERT;
pCmd->isInsertFromFile = -1;
pSql->res.numOfRows = 0;
return doParseInsertSql(pSql, sql + index);
return doParseInsertSql(pSql, pSql->sqlstr + index);
}
int tsParseSql(SSqlObj *pSql, bool multiVnodeInsertion) {
......@@ -1351,7 +1378,7 @@ static int tscInsertDataFromFile(SSqlObj *pSql, FILE *fp, char *tmpTokenBuf) {
while ((readLen = getline(&line, &n, fp)) != -1) {
// line[--readLen] = '\0';
if (('\r' == line[readLen - 1]) || ('\n' == line[readLen - 1])) line[--readLen] = 0;
if (readLen == 0) continue; //fang, <= to ==
if (readLen == 0) continue; // fang, <= to ==
char *lineptr = line;
strtolower(line, line);
......@@ -1362,7 +1389,8 @@ static int tscInsertDataFromFile(SSqlObj *pSql, FILE *fp, char *tmpTokenBuf) {
maxRows += tSize;
}
len = tsParseOneRowData(&lineptr, pTableDataBlock, pSchema, &spd, pCmd->payload, pMeterMeta->precision, &code, tmpTokenBuf);
len = tsParseOneRowData(&lineptr, pTableDataBlock, pSchema, &spd, pCmd->payload, pMeterMeta->precision, &code,
tmpTokenBuf);
if (len <= 0 || pTableDataBlock->numOfParams > 0) {
pSql->res.code = code;
return (-code);
......@@ -1425,7 +1453,7 @@ void tscProcessMultiVnodesInsert(SSqlObj *pSql) {
int32_t code = TSDB_CODE_SUCCESS;
/* the first block has been sent to server in processSQL function */
assert(pCmd->isInsertFromFile != -1 && pMeterMetaInfo->vnodeIndex >= 1 && pCmd->pDataBlocks != NULL);
assert(pMeterMetaInfo->vnodeIndex >= 1 && pCmd->pDataBlocks != NULL);
if (pMeterMetaInfo->vnodeIndex < pCmd->pDataBlocks->nSize) {
SDataBlockList *pDataBlocks = pCmd->pDataBlocks;
......@@ -1437,7 +1465,8 @@ void tscProcessMultiVnodesInsert(SSqlObj *pSql) {
}
if ((code = tscCopyDataBlockToPayload(pSql, pDataBlock)) != TSDB_CODE_SUCCESS) {
tscTrace("%p build submit data block failed, vnodeIdx:%d, total:%d", pSql, pMeterMetaInfo->vnodeIndex, pDataBlocks->nSize);
tscTrace("%p build submit data block failed, vnodeIdx:%d, total:%d", pSql, pMeterMetaInfo->vnodeIndex,
pDataBlocks->nSize);
continue;
}
......@@ -1450,17 +1479,19 @@ void tscProcessMultiVnodesInsert(SSqlObj *pSql) {
}
// multi-vnodes insertion in sync query model
void tscProcessMultiVnodesInsertForFile(SSqlObj *pSql) {
void tscProcessMultiVnodesInsertFromFile(SSqlObj *pSql) {
SSqlCmd *pCmd = &pSql->cmd;
if (pCmd->command != TSDB_SQL_INSERT) {
return;
}
SMeterMetaInfo * pInfo = tscGetMeterMetaInfo(pCmd, 0, 0);
SQueryInfo * pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0);
STableDataBlocks *pDataBlock = NULL;
int32_t affected_rows = 0;
assert(pCmd->isInsertFromFile == 1 && pCmd->pDataBlocks != NULL);
assert(pCmd->dataSourceType == DATA_FROM_DATA_FILE && pCmd->pDataBlocks != NULL);
SDataBlockList *pDataBlockList = pCmd->pDataBlocks;
pCmd->pDataBlocks = NULL;
......@@ -1486,16 +1517,16 @@ void tscProcessMultiVnodesInsertForFile(SSqlObj *pSql) {
continue;
}
strncpy(pInfo->name, pDataBlock->meterId, TSDB_METER_ID_LEN);
strncpy(pMeterMetaInfo->name, pDataBlock->meterId, TSDB_METER_ID_LEN);
memset(pDataBlock->pData, 0, pDataBlock->nAllocSize);
int32_t ret = tscGetMeterMeta(pSql, pInfo->name, 0);
int32_t ret = tscGetMeterMeta(pSql, pMeterMetaInfo);
if (ret != TSDB_CODE_SUCCESS) {
tscError("%p get meter meta failed, abort", pSql);
continue;
}
char* tmpTokenBuf = calloc(1, 4096); // used for deleting Escape character: \\, \', \"
char *tmpTokenBuf = calloc(1, 4096); // used for deleting Escape character: \\, \', \"
if (NULL == tmpTokenBuf) {
tscError("%p calloc failed", pSql);
continue;
......
src/client/src/tscSQLParser.c
浏览文件 @ f2016993
此差异已折叠。
src/client/src/tscSQLParserImpl.c
浏览文件 @ f2016993
......@@ -589,6 +589,8 @@ void destroyAllSelectClause(SSubclauseInfo *pClause) {
SQuerySQL *pQuerySql = pClause->pClause[i];
doDestroyQuerySql(pQuerySql);
}
tfree(pClause->pClause);
}
SCreateTableSQL *tSetCreateSQLElems(tFieldList *pCols, tFieldList *pTags, SSQLToken *pStableName,
......
src/client/src/tscServer.c
浏览文件 @ f2016993
此差异已折叠。
src/client/src/tscSql.c
浏览文件 @ f2016993
......@@ -298,8 +298,11 @@ int taos_num_fields(TAOS_RES *res) {
if (pSql == NULL || pSql->signature != pSql) return 0;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SFieldInfo *pFieldsInfo = &pQueryInfo->fieldsInfo;
if (pQueryInfo == NULL) {
return 0;
}
SFieldInfo *pFieldsInfo = &pQueryInfo->fieldsInfo;
return (pFieldsInfo->numOfOutputCols - pFieldsInfo->numOfHiddenCols);
}
......@@ -993,11 +996,8 @@ static int tscParseTblNameList(SSqlObj *pSql, const char *tblNameList, int32_t t
int code = TSDB_CODE_INVALID_METER_ID;
char *str = (char *)tblNameList;
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
if (pQueryInfo == NULL) {
tscAddSubqueryInfo(pCmd);
pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
}
SQueryInfo* pQueryInfo = NULL;
tscGetQueryInfoDetailSafely(pCmd, 0, &pQueryInfo);
SMeterMetaInfo *pMeterMetaInfo = tscAddEmptyMeterMetaInfo(pQueryInfo);
......@@ -1034,7 +1034,7 @@ static int tscParseTblNameList(SSqlObj *pSql, const char *tblNameList, int32_t t
return code;
}
if ((code = setMeterID(pSql, 0, &sToken, 0)) != TSDB_CODE_SUCCESS) {
if ((code = setMeterID(pMeterMetaInfo, &sToken, pSql)) != TSDB_CODE_SUCCESS) {
return code;
}
......
src/client/src/tscStream.c
浏览文件 @ f2016993
......@@ -70,7 +70,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
SQueryInfo *pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SMeterMetaInfo *pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0);
int code = tscGetMeterMeta(pSql, pMeterMetaInfo->name, 0);
int code = tscGetMeterMeta(pSql, pMeterMetaInfo);
pSql->res.code = code;
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
......@@ -82,7 +82,7 @@ static void tscProcessStreamLaunchQuery(SSchedMsg *pMsg) {
if (code == TSDB_CODE_ACTION_IN_PROGRESS) return;
}
tscTansformSQLFunctionForMetricQuery(pQueryInfo);
tscTansformSQLFunctionForSTableQuery(pQueryInfo);
// failed to get meter/metric meta, retry in 10sec.
if (code != TSDB_CODE_SUCCESS) {
......@@ -391,6 +391,7 @@ static void tscSetSlidingWindowInfo(SSqlObj *pSql, SSqlStream *pStream) {
}
pStream->slidingTime = pQueryInfo->nSlidingTime;
pQueryInfo->nAggTimeInterval = 0; // clear the interval value to avoid the force time window split by query processor
}
static int64_t tscGetStreamStartTimestamp(SSqlObj *pSql, SSqlStream *pStream, int64_t stime) {
......@@ -507,8 +508,7 @@ TAOS_STREAM *taos_open_stream(TAOS *taos, const char *sqlstr, void (*fp)(void *p
return NULL;
}
// TODO later refactor use enum
pSql->cmd.count = 1; // 1 means sql in stream, allowed the sliding clause.
pSql->cmd.inStream = 1; // 1 means sql in stream, allowed the sliding clause.
pRes->code = tscToSQLCmd(pSql, &SQLInfo);
SQLInfoDestroy(&SQLInfo);
......
src/client/src/tscUtil.c
浏览文件 @ f2016993
......@@ -319,7 +319,7 @@ void tscClearInterpInfo(SQueryInfo* pQueryInfo) {
}
pQueryInfo->interpoType = TSDB_INTERPO_NONE;
memset(pQueryInfo->defaultVal, 0, sizeof(pQueryInfo->defaultVal));
tfree(pQueryInfo->defaultVal);
}
void tscClearSqlMetaInfoForce(SSqlCmd* pCmd) {
......@@ -398,20 +398,20 @@ void tscFreeSqlObjPartial(SSqlObj* pSql) {
tfree(pSql->sqlstr);
pthread_mutex_unlock(&pObj->mutex);
tfree(pSql->res.pRsp);
pSql->res.row = 0;
pSql->res.numOfRows = 0;
pSql->res.numOfTotal = 0;
tfree(pRes->pRsp);
pRes->row = 0;
pRes->numOfRows = 0;
pRes->numOfTotal = 0;
pSql->res.numOfGroups = 0;
tfree(pSql->res.pGroupRec);
pRes->numOfGroups = 0;
tfree(pRes->pGroupRec);
tscDestroyLocalReducer(pSql);
tfree(pSql->pSubs);
pSql->numOfSubs = 0;
tscDestroyResPointerInfo(pRes);
tfree(pSql->res.pColumnIndex);
tfree(pRes->pColumnIndex);
tscFreeSqlCmdData(pCmd);
}
......@@ -535,7 +535,7 @@ int32_t tscCopyDataBlockToPayload(SSqlObj* pSql, STableDataBlocks* pDataBlock) {
SSqlCmd* pCmd = &pSql->cmd;
assert(pDataBlock->pMeterMeta != NULL);
pCmd->count = pDataBlock->numOfMeters;
pCmd->numOfTablesInSubmit = pDataBlock->numOfMeters;
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfo(pCmd, 0, 0);
// set the correct metermeta object, the metermeta has been locked in pDataBlocks, so it must be in the cache
......@@ -1548,7 +1548,7 @@ bool tscShouldFreeAsyncSqlObj(SSqlObj* pSql) {
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(&pSql->cmd, 0);
SMeterMetaInfo* pMeterMetaInfo = tscGetMeterMetaInfoFromQueryInfo(pQueryInfo, 0);
assert(pQueryInfo->numOfTables == 1);
assert(pQueryInfo->numOfTables == 1 || pQueryInfo->numOfTables == 2);
if (pDataBlocks == NULL || pMeterMetaInfo->vnodeIndex >= pDataBlocks->nSize) {
tscTrace("%p object should be release since all data blocks have been submit", pSql);
......@@ -1664,6 +1664,8 @@ static void doClearSubqueryInfo(SQueryInfo* pQueryInfo) {
memset(&pQueryInfo->colList, 0, sizeof(pQueryInfo->colList));
pQueryInfo->tsBuf = tsBufDestory(pQueryInfo->tsBuf);
tfree(pQueryInfo->defaultVal);
}
void tscClearSubqueryInfo(SSqlCmd* pCmd) {
......@@ -1814,15 +1816,18 @@ SSqlObj* createSubqueryObj(SSqlObj* pSql, int16_t tableIndex, void (*fp)(), void
SQueryInfo* pQueryInfo = tscGetQueryInfoDetail(pCmd, 0);
memcpy(pNewQueryInfo, pQueryInfo, sizeof(SQueryInfo));
pNewQueryInfo->pMeterInfo = NULL;
memset(&pNewQueryInfo->colList, 0, sizeof(pNewQueryInfo->colList));
memset(&pNewQueryInfo->fieldsInfo, 0, sizeof(SFieldInfo));
pNewQueryInfo->pMeterInfo = NULL;
pNewQueryInfo->defaultVal = NULL;
pNewQueryInfo->numOfTables = 0;
pNewQueryInfo->tsBuf = NULL;
tscTagCondCopy(&pNewQueryInfo->tagCond, &pQueryInfo->tagCond);
pNewQueryInfo->defaultVal = malloc(pQueryInfo->fieldsInfo.numOfOutputCols * sizeof(int64_t));
memcpy(pNewQueryInfo->defaultVal, pQueryInfo->defaultVal, pQueryInfo->fieldsInfo.numOfOutputCols * sizeof(int64_t));
if (tscAllocPayload(&pNew->cmd, TSDB_DEFAULT_PAYLOAD_SIZE) != TSDB_CODE_SUCCESS) {
tscError("%p new subquery failed, tableIndex:%d, vnodeIndex:%d", pSql, tableIndex, pMeterMetaInfo->vnodeIndex);
......@@ -1910,8 +1915,8 @@ void tscDoQuery(SSqlObj* pSql) {
tscAddIntoSqlList(pSql);
}
if (pCmd->isInsertFromFile == 1) {
tscProcessMultiVnodesInsertForFile(pSql);
if (pCmd->dataSourceType == DATA_FROM_DATA_FILE) {
tscProcessMultiVnodesInsertFromFile(pSql);
} else {
// pSql may be released in this function if it is a async insertion.
tscProcessSql(pSql);
......
src/inc/tsdb.h
浏览文件 @ f2016993
......@@ -208,7 +208,7 @@ extern "C" {
#define TSDB_MAX_RPC_THREADS 5
#define TSDB_QUERY_TYPE_QUERY 0 // normal query
#define TSDB_QUERY_TYPE_NON_TYPE 0x00U // none type
#define TSDB_QUERY_TYPE_FREE_RESOURCE 0x01U // free qhandle at vnode
/*
......@@ -224,6 +224,13 @@ extern "C" {
#define TSDB_QUERY_TYPE_PROJECTION_QUERY 0x40U // select *,columns... query
#define TSDB_QUERY_TYPE_JOIN_SEC_STAGE 0x80U // join sub query at the second stage
#define TSDB_QUERY_TYPE_INSERT 0x100U // insert type
#define TSDB_QUERY_TYPE_IMPORT 0x200U // import data
#define TSDB_QUERY_HAS_TYPE(x, _type) (((x) & (_type)) != 0)
#define TSDB_QUERY_SET_TYPE(x, _type) ((x) |= (_type))
#define TSDB_QUERY_RESET_TYPE(x) ((x) = TSDB_QUERY_TYPE_NON_TYPE)
#define TSQL_SO_ASC 1
#define TSQL_SO_DESC 0
......
src/system/detail/src/vnodeShell.c
浏览文件 @ f2016993
......@@ -585,6 +585,7 @@ int vnodeProcessShellSubmitRequest(char *pMsg, int msgLen, SShellObj *pObj) {
SShellSubmitMsg *pSubmit = &shellSubmit;
SShellSubmitBlock *pBlocks = NULL;
pSubmit->import = htons(pSubmit->import);
pSubmit->vnode = htons(pSubmit->vnode);
pSubmit->numOfSid = htonl(pSubmit->numOfSid);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或