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

Merge pull request #4578 from taosdata/feature/query 

Feature/query
This commit is contained in:
haojun Liao 2020-12-16 10:02:13 +08:00 committed by GitHub
parent d63dc4c60d 37c4530974
commit ecf37fa634
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 1421 additions and 1202 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/client/inc/tscUtil.h
View File

@ -234,7 +234,7 @@ void tscVgroupTableCopy(SVgroupTableInfo* info, SVgroupTableInfo* pInfo);
int tscGetSTableVgroupInfo(SSqlObj* pSql, int32_t clauseIndex); int tscGetSTableVgroupInfo(SSqlObj* pSql, int32_t clauseIndex);
int tscGetTableMeta(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo); int tscGetTableMeta(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo);
int tscGetMeterMetaEx(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, bool createIfNotExists); int tscGetTableMetaEx(SSqlObj* pSql, STableMetaInfo* pTableMetaInfo, bool createIfNotExists);
void tscResetForNextRetrieve(SSqlRes* pRes); void tscResetForNextRetrieve(SSqlRes* pRes);
void tscDoQuery(SSqlObj* pSql); void tscDoQuery(SSqlObj* pSql);
@ -287,6 +287,9 @@ bool tscSetSqlOwner(SSqlObj* pSql);
void tscClearSqlOwner(SSqlObj* pSql); void tscClearSqlOwner(SSqlObj* pSql);
int32_t doArithmeticCalculate(SQueryInfo* pQueryInfo, tFilePage* pOutput, int32_t rowSize, int32_t finalRowSize); int32_t doArithmeticCalculate(SQueryInfo* pQueryInfo, tFilePage* pOutput, int32_t rowSize, int32_t finalRowSize);
char* serializeTagData(STagData* pTagData, char* pMsg);
int32_t copyTagData(STagData* dst, const STagData* src);
void* malloc_throw(size_t size); void* malloc_throw(size_t size);
void* calloc_throw(size_t nmemb, size_t size); void* calloc_throw(size_t nmemb, size_t size);
char* strdup_throw(const char* str); char* strdup_throw(const char* str);

2
src/client/inc/tsclient.h
View File

@ -246,7 +246,7 @@ typedef struct {
int8_t dataSourceType; // load data from file or not int8_t dataSourceType; // load data from file or not
int8_t submitSchema; // submit block is built with table schema int8_t submitSchema; // submit block is built with table schema
STagData *pTagData; // NOTE: pTagData->data is used as a variant length array STagData tagData; // NOTE: pTagData->data is used as a variant length array
STableMeta **pTableMetaList; // all involved tableMeta list of current insert sql statement. STableMeta **pTableMetaList; // all involved tableMeta list of current insert sql statement.
int32_t numOfTables; int32_t numOfTables;

24
src/client/src/tscParseInsert.c
View File

@ -796,8 +796,6 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
sToken = tStrGetToken(sql, &index, false, 0, NULL); sToken = tStrGetToken(sql, &index, false, 0, NULL);
sql += index; sql += index;
tscAllocPayload(pCmd, sizeof(STagData));
//the source super table is moved to the secondary position of the pTableMetaInfo list //the source super table is moved to the secondary position of the pTableMetaInfo list
if (pQueryInfo->numOfTables < 2) { if (pQueryInfo->numOfTables < 2) {
tscAddEmptyMetaInfo(pQueryInfo); tscAddEmptyMetaInfo(pQueryInfo);
@ -809,13 +807,8 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
return code; return code;
} }
STagData *pTag = realloc(pCmd->pTagData, offsetof(STagData, data)); tstrncpy(pCmd->tagData.name, pSTableMeterMetaInfo->name, sizeof(pCmd->tagData.name));
if (pTag == NULL) { pCmd->tagData.dataLen = 0;
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
memset(pTag, 0, offsetof(STagData, data));
tstrncpy(pTag->name, pSTableMeterMetaInfo->name, sizeof(pTag->name));
pCmd->pTagData = pTag;
code = tscGetTableMeta(pSql, pSTableMeterMetaInfo); code = tscGetTableMeta(pSql, pSTableMeterMetaInfo);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
@ -946,14 +939,15 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
} }
tdSortKVRowByColIdx(row); tdSortKVRowByColIdx(row);
pTag = (STagData*)realloc(pCmd->pTagData, offsetof(STagData, data) + kvRowLen(row)); pCmd->tagData.dataLen = kvRowLen(row);
char* pTag = realloc(pCmd->tagData.data, pCmd->tagData.dataLen);
if (pTag == NULL) { if (pTag == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY; return TSDB_CODE_TSC_OUT_OF_MEMORY;
} }
pCmd->pTagData = pTag;
pTag->dataLen = htonl(kvRowLen(row)); kvRowCpy(pTag, row);
kvRowCpy(pTag->data, row);
free(row); free(row);
pCmd->tagData.data = pTag;
index = 0; index = 0;
sToken = tStrGetToken(sql, &index, false, 0, NULL); sToken = tStrGetToken(sql, &index, false, 0, NULL);
@ -972,7 +966,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
} }
createTable = true; createTable = true;
code = tscGetMeterMetaEx(pSql, pTableMetaInfo, true); code = tscGetTableMetaEx(pSql, pTableMetaInfo, true);
if (TSDB_CODE_TSC_ACTION_IN_PROGRESS == code) { if (TSDB_CODE_TSC_ACTION_IN_PROGRESS == code) {
return code; return code;
} }
@ -983,7 +977,7 @@ static int32_t tscCheckIfCreateTable(char **sqlstr, SSqlObj *pSql) {
} else { } else {
sql = sToken.z; sql = sToken.z;
} }
code = tscGetMeterMetaEx(pSql, pTableMetaInfo, false); code = tscGetTableMetaEx(pSql, pTableMetaInfo, false);
if (pCmd->curSql == NULL) { if (pCmd->curSql == NULL) {
assert(code == TSDB_CODE_TSC_ACTION_IN_PROGRESS); assert(code == TSDB_CODE_TSC_ACTION_IN_PROGRESS);

184
src/client/src/tscSQLParser.c
View File

@ -1775,12 +1775,15 @@ void setResultColName(char* name, tSQLExprItem* pItem, int32_t functionId, SStrT
int32_t len = MIN(pToken->n + 1, TSDB_COL_NAME_LEN); int32_t len = MIN(pToken->n + 1, TSDB_COL_NAME_LEN);
tstrncpy(uname, pToken->z, len); tstrncpy(uname, pToken->z, len);
int32_t size = TSDB_COL_NAME_LEN + tListLen(aAggs[functionId].aName) + 2 + 1; if (tsKeepOriginalColumnName) { // keep the original column name
char tmp[TSDB_COL_NAME_LEN + tListLen(aAggs[functionId].aName) + 2 + 1] = {0}; tstrncpy(name, uname, TSDB_COL_NAME_LEN);
} else {
int32_t size = TSDB_COL_NAME_LEN + tListLen(aAggs[functionId].aName) + 2 + 1;
char tmp[TSDB_COL_NAME_LEN + tListLen(aAggs[functionId].aName) + 2 + 1] = {0};
snprintf(tmp, size, "%s(%s)", aAggs[functionId].aName, uname);
snprintf(tmp, size, "%s(%s)", aAggs[functionId].aName, uname); tstrncpy(name, tmp, TSDB_COL_NAME_LEN);
}
tstrncpy(name, tmp, TSDB_COL_NAME_LEN);
} else { // use the user-input result column name } else { // use the user-input result column name
int32_t len = MIN(pItem->pNode->token.n + 1, TSDB_COL_NAME_LEN); int32_t len = MIN(pItem->pNode->token.n + 1, TSDB_COL_NAME_LEN);
tstrncpy(name, pItem->pNode->token.z, len); tstrncpy(name, pItem->pNode->token.z, len);
@ -4910,6 +4913,8 @@ int32_t setAlterTableInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
tVariantListItem* pItem = taosArrayGet(pVarList, 1); tVariantListItem* pItem = taosArrayGet(pVarList, 1);
SSchema* pTagsSchema = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, columnIndex.columnIndex); SSchema* pTagsSchema = tscGetTableColumnSchema(pTableMetaInfo->pTableMeta, columnIndex.columnIndex);
pAlterSQL->tagData.data = calloc(1, pTagsSchema->bytes * TSDB_NCHAR_SIZE + VARSTR_HEADER_SIZE);
if (tVariantDump(&pItem->pVar, pAlterSQL->tagData.data, pTagsSchema->type, true) != TSDB_CODE_SUCCESS) { if (tVariantDump(&pItem->pVar, pAlterSQL->tagData.data, pTagsSchema->type, true) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg13); return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg13);
} }
@ -6149,96 +6154,105 @@ int32_t doCheckForCreateFromStable(SSqlObj* pSql, SSqlInfo* pInfo) {
const int32_t TABLE_INDEX = 0; const int32_t TABLE_INDEX = 0;
const int32_t STABLE_INDEX = 1; const int32_t STABLE_INDEX = 1;
STableMetaInfo* pStableMeterMetaInfo = tscGetMetaInfo(pQueryInfo, STABLE_INDEX); STableMetaInfo* pStableMetaInfo = tscGetMetaInfo(pQueryInfo, STABLE_INDEX);
// super table name, create table by using dst // super table name, create table by using dst
SStrToken* pToken = &(pCreateTable->usingInfo.stableName); int32_t numOfTables = (int32_t) taosArrayGetSize(pCreateTable->childTableInfo);
for(int32_t j = 0; j < numOfTables; ++j) {
SCreatedTableInfo* pCreateTableInfo = taosArrayGet(pCreateTable->childTableInfo, j);
if (tscValidateName(pToken) != TSDB_CODE_SUCCESS) { SStrToken* pToken = &pCreateTableInfo->stableName;
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1); if (tscValidateName(pToken) != TSDB_CODE_SUCCESS) {
} return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
int32_t code = tscSetTableFullName(pStableMeterMetaInfo, pToken, pSql);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
// get meter meta from mnode
tstrncpy(pCreateTable->usingInfo.tagdata.name, pStableMeterMetaInfo->name, sizeof(pCreateTable->usingInfo.tagdata.name));
SArray* pList = pInfo->pCreateTableInfo->usingInfo.pTagVals;
code = tscGetTableMeta(pSql, pStableMeterMetaInfo);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
size_t size = taosArrayGetSize(pList);
if (tscGetNumOfTags(pStableMeterMetaInfo->pTableMeta) != size) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg5);
}
// too long tag values will return invalid sql, not be truncated automatically
SSchema* pTagSchema = tscGetTableTagSchema(pStableMeterMetaInfo->pTableMeta);
STagData* pTag = &pCreateTable->usingInfo.tagdata;
SKVRowBuilder kvRowBuilder = {0};
if (tdInitKVRowBuilder(&kvRowBuilder) < 0) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
int32_t ret = TSDB_CODE_SUCCESS;
for (int32_t i = 0; i < size; ++i) {
SSchema* pSchema = &pTagSchema[i];
tVariantListItem* pItem = taosArrayGet(pList, i);
char tagVal[TSDB_MAX_TAGS_LEN];
if (pSchema->type == TSDB_DATA_TYPE_BINARY || pSchema->type == TSDB_DATA_TYPE_NCHAR) {
if (pItem->pVar.nLen > pSchema->bytes) {
tdDestroyKVRowBuilder(&kvRowBuilder);
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
}
} }
ret = tVariantDump(&(pItem->pVar), tagVal, pSchema->type, true); int32_t code = tscSetTableFullName(pStableMetaInfo, pToken, pSql);
if (code != TSDB_CODE_SUCCESS) {
// check again after the convert since it may be converted from binary to nchar. return code;
if (pSchema->type == TSDB_DATA_TYPE_BINARY || pSchema->type == TSDB_DATA_TYPE_NCHAR) {
int16_t len = varDataTLen(tagVal);
if (len > pSchema->bytes) {
tdDestroyKVRowBuilder(&kvRowBuilder);
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
}
} }
// get table meta from mnode
tstrncpy(pCreateTableInfo->tagdata.name, pStableMetaInfo->name, tListLen(pCreateTableInfo->tagdata.name));
SArray* pList = pCreateTableInfo->pTagVals;
code = tscGetTableMeta(pSql, pStableMetaInfo);
if (code != TSDB_CODE_SUCCESS) {
return code;
}
size_t size = taosArrayGetSize(pList);
if (tscGetNumOfTags(pStableMetaInfo->pTableMeta) != size) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg5);
}
// too long tag values will return invalid sql, not be truncated automatically
SSchema *pTagSchema = tscGetTableTagSchema(pStableMetaInfo->pTableMeta);
STagData *pTag = &pCreateTableInfo->tagdata;
SKVRowBuilder kvRowBuilder = {0};
if (tdInitKVRowBuilder(&kvRowBuilder) < 0) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
int32_t ret = TSDB_CODE_SUCCESS;
for (int32_t i = 0; i < size; ++i) {
SSchema* pSchema = &pTagSchema[i];
tVariantListItem* pItem = taosArrayGet(pList, i);
char tagVal[TSDB_MAX_TAGS_LEN];
if (pSchema->type == TSDB_DATA_TYPE_BINARY || pSchema->type == TSDB_DATA_TYPE_NCHAR) {
if (pItem->pVar.nLen > pSchema->bytes) {
tdDestroyKVRowBuilder(&kvRowBuilder);
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
}
}
ret = tVariantDump(&(pItem->pVar), tagVal, pSchema->type, true);
// check again after the convert since it may be converted from binary to nchar.
if (pSchema->type == TSDB_DATA_TYPE_BINARY || pSchema->type == TSDB_DATA_TYPE_NCHAR) {
int16_t len = varDataTLen(tagVal);
if (len > pSchema->bytes) {
tdDestroyKVRowBuilder(&kvRowBuilder);
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
}
}
if (ret != TSDB_CODE_SUCCESS) {
tdDestroyKVRowBuilder(&kvRowBuilder);
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg4);
}
tdAddColToKVRow(&kvRowBuilder, pSchema->colId, pSchema->type, tagVal);
}
SKVRow row = tdGetKVRowFromBuilder(&kvRowBuilder);
tdDestroyKVRowBuilder(&kvRowBuilder);
if (row == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
tdSortKVRowByColIdx(row);
pTag->dataLen = kvRowLen(row);
if (pTag->data == NULL) {
pTag->data = malloc(pTag->dataLen);
}
kvRowCpy(pTag->data, row);
free(row);
// table name
if (tscValidateName(&(pCreateTableInfo->name)) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
}
STableMetaInfo* pTableMetaInfo = tscGetMetaInfo(pQueryInfo, TABLE_INDEX);
ret = tscSetTableFullName(pTableMetaInfo, &pCreateTableInfo->name, pSql);
if (ret != TSDB_CODE_SUCCESS) { if (ret != TSDB_CODE_SUCCESS) {
tdDestroyKVRowBuilder(&kvRowBuilder); return ret;
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg4);
} }
pCreateTableInfo->fullname = strndup(pTableMetaInfo->name, TSDB_TABLE_FNAME_LEN);
tdAddColToKVRow(&kvRowBuilder, pSchema->colId, pSchema->type, tagVal);
}
SKVRow row = tdGetKVRowFromBuilder(&kvRowBuilder);
tdDestroyKVRowBuilder(&kvRowBuilder);
if (row == NULL) {
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
tdSortKVRowByColIdx(row);
pTag->dataLen = kvRowLen(row);
kvRowCpy(pTag->data, row);
free(row);
// table name
if (tscValidateName(&pInfo->pCreateTableInfo->name) != TSDB_CODE_SUCCESS) {
return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
}
STableMetaInfo* pTableMeterMetaInfo = tscGetMetaInfo(pQueryInfo, TABLE_INDEX);
ret = tscSetTableFullName(pTableMeterMetaInfo, &pInfo->pCreateTableInfo->name, pSql);
if (ret != TSDB_CODE_SUCCESS) {
return ret;
} }
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;

104
src/client/src/tscServer.c
View File

@ -185,7 +185,7 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
pSql->res.code = TSDB_CODE_RPC_NETWORK_UNAVAIL; pSql->res.code = TSDB_CODE_RPC_NETWORK_UNAVAIL;
} }
if (pRes->buffer == NULL) { if (pRes->length == NULL) {
pRes->length = calloc(2, sizeof(int32_t)); pRes->length = calloc(2, sizeof(int32_t));
} }
@ -193,7 +193,7 @@ void tscProcessHeartBeatRsp(void *param, TAOS_RES *tres, int code) {
pRes->length[1] = online; pRes->length[1] = online;
} else { } else {
tscDebug("%" PRId64 " heartbeat failed, code:%s", pObj->hbrid, tstrerror(code)); tscDebug("%" PRId64 " heartbeat failed, code:%s", pObj->hbrid, tstrerror(code));
if (pRes->buffer == NULL) { if (pRes->length == NULL) {
pRes->length = calloc(2, sizeof(int32_t)); pRes->length = calloc(2, sizeof(int32_t));
} }
@ -1268,10 +1268,10 @@ int32_t tscBuildKillMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
// TODO update it
int tscEstimateCreateTableMsgLength(SSqlObj *pSql, SSqlInfo *pInfo) { int tscEstimateCreateTableMsgLength(SSqlObj *pSql, SSqlInfo *pInfo) {
SSqlCmd *pCmd = &(pSql->cmd); SSqlCmd *pCmd = &(pSql->cmd);
int32_t size = minMsgSize() + sizeof(SCMCreateTableMsg) + sizeof(SCreatedTableInfo);
int32_t size = minMsgSize() + sizeof(SCMCreateTableMsg);
SCreateTableSQL *pCreateTableInfo = pInfo->pCreateTableInfo; SCreateTableSQL *pCreateTableInfo = pInfo->pCreateTableInfo;
if (pCreateTableInfo->type == TSQL_CREATE_TABLE_FROM_STABLE) { if (pCreateTableInfo->type == TSQL_CREATE_TABLE_FROM_STABLE) {
@ -1303,33 +1303,55 @@ int tscBuildCreateTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
return TSDB_CODE_TSC_OUT_OF_MEMORY; return TSDB_CODE_TSC_OUT_OF_MEMORY;
} }
SCMCreateTableMsg *pCreateTableMsg = (SCMCreateTableMsg *)pCmd->payload; SCMCreateTableMsg *pCreateTableMsg = (SCMCreateTableMsg *)pCmd->payload;
strcpy(pCreateTableMsg->tableId, pTableMetaInfo->name);
// use dbinfo from table id without modifying current db info SCreateTableMsg* pCreateMsg = (SCreateTableMsg*)((char*) pCreateTableMsg + sizeof(SCMCreateTableMsg));
tscGetDBInfoFromTableFullName(pTableMetaInfo->name, pCreateTableMsg->db); char* pMsg = NULL;
SCreateTableSQL *pCreateTable = pInfo->pCreateTableInfo;
pCreateTableMsg->igExists = pCreateTable->existCheck ? 1 : 0;
pCreateTableMsg->numOfColumns = htons(pCmd->numOfCols);
pCreateTableMsg->numOfTags = htons(pCmd->count);
pCreateTableMsg->sqlLen = 0;
char *pMsg = (char *)pCreateTableMsg->schema;
int8_t type = pInfo->pCreateTableInfo->type; int8_t type = pInfo->pCreateTableInfo->type;
if (type == TSQL_CREATE_TABLE_FROM_STABLE) { // create by using super table, tags value if (type == TSQL_CREATE_TABLE_FROM_STABLE) { // create by using super table, tags value
STagData* pTag = &pInfo->pCreateTableInfo->usingInfo.tagdata; SArray* list = pInfo->pCreateTableInfo->childTableInfo;
*(int32_t*)pMsg = htonl(pTag->dataLen);
pMsg += sizeof(int32_t); int32_t numOfTables = (int32_t) taosArrayGetSize(list);
memcpy(pMsg, pTag->name, sizeof(pTag->name)); pCreateTableMsg->numOfTables = htonl(numOfTables);
pMsg += sizeof(pTag->name);
memcpy(pMsg, pTag->data, pTag->dataLen); pMsg = (char*) pCreateMsg;
pMsg += pTag->dataLen; for(int32_t i = 0; i < numOfTables; ++i) {
SCreateTableMsg* pCreate = (SCreateTableMsg*) pMsg;
pCreate->numOfColumns = htons(pCmd->numOfCols);
pCreate->numOfTags = htons(pCmd->count);
pMsg += sizeof(SCreateTableMsg);
SCreatedTableInfo* p = taosArrayGet(list, i);
strcpy(pCreate->tableId, p->fullname);
pCreate->igExists = (p->igExist)? 1 : 0;
// use dbinfo from table id without modifying current db info
tscGetDBInfoFromTableFullName(p->fullname, pCreate->db);
pMsg = serializeTagData(&p->tagdata, pMsg);
int32_t len = (int32_t)(pMsg - (char*) pCreate);
pCreate->len = htonl(len);
}
} else { // create (super) table } else { // create (super) table
pSchema = (SSchema *)pCreateTableMsg->schema; pCreateTableMsg->numOfTables = htonl(1); // only one table will be created
strcpy(pCreateMsg->tableId, pTableMetaInfo->name);
// use dbinfo from table id without modifying current db info
tscGetDBInfoFromTableFullName(pTableMetaInfo->name, pCreateMsg->db);
SCreateTableSQL *pCreateTable = pInfo->pCreateTableInfo;
pCreateMsg->igExists = pCreateTable->existCheck ? 1 : 0;
pCreateMsg->numOfColumns = htons(pCmd->numOfCols);
pCreateMsg->numOfTags = htons(pCmd->count);
pCreateMsg->sqlLen = 0;
pMsg = (char *)pCreateMsg->schema;
pSchema = (SSchema *)pCreateMsg->schema;
for (int i = 0; i < pCmd->numOfCols + pCmd->count; ++i) { for (int i = 0; i < pCmd->numOfCols + pCmd->count; ++i) {
TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i); TAOS_FIELD *pField = tscFieldInfoGetField(&pQueryInfo->fieldsInfo, i);
@ -1346,7 +1368,7 @@ int tscBuildCreateTableMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
SQuerySQL *pQuerySql = pInfo->pCreateTableInfo->pSelect; SQuerySQL *pQuerySql = pInfo->pCreateTableInfo->pSelect;
strncpy(pMsg, pQuerySql->selectToken.z, pQuerySql->selectToken.n + 1); strncpy(pMsg, pQuerySql->selectToken.z, pQuerySql->selectToken.n + 1);
pCreateTableMsg->sqlLen = htons(pQuerySql->selectToken.n + 1); pCreateMsg->sqlLen = htons(pQuerySql->selectToken.n + 1);
pMsg += pQuerySql->selectToken.n + 1; pMsg += pQuerySql->selectToken.n + 1;
} }
} }
@ -1623,13 +1645,8 @@ int tscBuildTableMetaMsg(SSqlObj *pSql, SSqlInfo *pInfo) {
char *pMsg = (char *)pInfoMsg + sizeof(STableInfoMsg); char *pMsg = (char *)pInfoMsg + sizeof(STableInfoMsg);
if (pCmd->autoCreated && pCmd->pTagData != NULL) { if (pCmd->autoCreated && pCmd->tagData.dataLen != 0) {
int len = htonl(pCmd->pTagData->dataLen); pMsg = serializeTagData(&pCmd->tagData, pMsg);
if (len > 0) {
len += sizeof(pCmd->pTagData->name) + sizeof(pCmd->pTagData->dataLen);
memcpy(pInfoMsg->tags, pCmd->pTagData, len);
pMsg += len;
}
} }
pCmd->payloadLen = (int32_t)(pMsg - (char*)pInfoMsg); pCmd->payloadLen = (int32_t)(pMsg - (char*)pInfoMsg);
@ -2175,10 +2192,7 @@ int tscProcessDropTableRsp(SSqlObj *pSql) {
*/ */
tscDebug("%p force release table meta after drop table:%s", pSql, pTableMetaInfo->name); tscDebug("%p force release table meta after drop table:%s", pSql, pTableMetaInfo->name);
taosCacheRelease(tscMetaCache, (void **)&pTableMeta, true); taosCacheRelease(tscMetaCache, (void **)&pTableMeta, true);
assert(pTableMetaInfo->pTableMeta == NULL);
if (pTableMetaInfo->pTableMeta) {
taosCacheRelease(tscMetaCache, (void **)&(pTableMetaInfo->pTableMeta), true);
}
return 0; return 0;
} }
@ -2286,7 +2300,7 @@ int tscProcessRetrieveRspFromNode(SSqlObj *pSql) {
void tscTableMetaCallBack(void *param, TAOS_RES *res, int code); void tscTableMetaCallBack(void *param, TAOS_RES *res, int code);
static int32_t getTableMetaFromMgmt(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo) { static int32_t getTableMetaFromMnode(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo) {
SSqlObj *pNew = calloc(1, sizeof(SSqlObj)); SSqlObj *pNew = calloc(1, sizeof(SSqlObj));
if (NULL == pNew) { if (NULL == pNew) {
tscError("%p malloc failed for new sqlobj to get table meta", pSql); tscError("%p malloc failed for new sqlobj to get table meta", pSql);
@ -2313,15 +2327,13 @@ static int32_t getTableMetaFromMgmt(SSqlObj *pSql, STableMetaInfo *pTableMetaInf
tstrncpy(pNewMeterMetaInfo->name, pTableMetaInfo->name, sizeof(pNewMeterMetaInfo->name)); tstrncpy(pNewMeterMetaInfo->name, pTableMetaInfo->name, sizeof(pNewMeterMetaInfo->name));
if (pSql->cmd.pTagData != NULL) { if (pSql->cmd.autoCreated) {
int size = offsetof(STagData, data) + htonl(pSql->cmd.pTagData->dataLen); int32_t code = copyTagData(&pNew->cmd.tagData, &pSql->cmd.tagData);
pNew->cmd.pTagData = calloc(1, size); if (code != TSDB_CODE_SUCCESS) {
if (pNew->cmd.pTagData == NULL) {
tscError("%p malloc failed for new tag data to get table meta", pSql); tscError("%p malloc failed for new tag data to get table meta", pSql);
tscFreeSqlObj(pNew); tscFreeSqlObj(pNew);
return TSDB_CODE_TSC_OUT_OF_MEMORY; return TSDB_CODE_TSC_OUT_OF_MEMORY;
} }
memcpy(pNew->cmd.pTagData, pSql->cmd.pTagData, size);
} }
tscDebug("%p new pSqlObj:%p to get tableMeta, auto create:%d", pSql, pNew, pNew->cmd.autoCreated); tscDebug("%p new pSqlObj:%p to get tableMeta, auto create:%d", pSql, pNew, pNew->cmd.autoCreated);
@ -2356,10 +2368,10 @@ int32_t tscGetTableMeta(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo) {
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} }
return getTableMetaFromMgmt(pSql, pTableMetaInfo); return getTableMetaFromMnode(pSql, pTableMetaInfo);
} }
int tscGetMeterMetaEx(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo, bool createIfNotExists) { int tscGetTableMetaEx(SSqlObj *pSql, STableMetaInfo *pTableMetaInfo, bool createIfNotExists) {
pSql->cmd.autoCreated = createIfNotExists; pSql->cmd.autoCreated = createIfNotExists;
return tscGetTableMeta(pSql, pTableMetaInfo); return tscGetTableMeta(pSql, pTableMetaInfo);
} }
@ -2383,7 +2395,7 @@ int tscRenewTableMeta(SSqlObj *pSql, int32_t tableIndex) {
} }
taosCacheRelease(tscMetaCache, (void **)&(pTableMetaInfo->pTableMeta), true); taosCacheRelease(tscMetaCache, (void **)&(pTableMetaInfo->pTableMeta), true);
return getTableMetaFromMgmt(pSql, pTableMetaInfo); return getTableMetaFromMnode(pSql, pTableMetaInfo);
} }
static bool allVgroupInfoRetrieved(SSqlCmd* pCmd, int32_t clauseIndex) { static bool allVgroupInfoRetrieved(SSqlCmd* pCmd, int32_t clauseIndex) {

8
src/client/src/tscSubquery.c
View File

@ -2209,10 +2209,10 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
// restore user defined fp // restore user defined fp
pParentObj->fp = pParentObj->fetchFp; pParentObj->fp = pParentObj->fetchFp;
int32_t numOfSub = pParentObj->subState.numOfSub; int32_t numOfSub = pParentObj->subState.numOfSub;
doFreeInsertSupporter(pParentObj);
if (pParentObj->res.code == TSDB_CODE_SUCCESS) { if (pParentObj->res.code == TSDB_CODE_SUCCESS) {
tscDebug("%p Async insertion completed, total inserted:%d", pParentObj, pParentObj->res.numOfRows); tscDebug("%p Async insertion completed, total inserted:%d", pParentObj, pParentObj->res.numOfRows);
doFreeInsertSupporter(pParentObj);
// todo remove this parameter in async callback function definition. // todo remove this parameter in async callback function definition.
// all data has been sent to vnode, call user function // all data has been sent to vnode, call user function
@ -2220,7 +2220,6 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
(*pParentObj->fp)(pParentObj->param, pParentObj, v); (*pParentObj->fp)(pParentObj->param, pParentObj, v);
} else { } else {
if (!needRetryInsert(pParentObj, numOfSub)) { if (!needRetryInsert(pParentObj, numOfSub)) {
doFreeInsertSupporter(pParentObj);
tscQueueAsyncRes(pParentObj); tscQueueAsyncRes(pParentObj);
return; return;
} }
@ -2263,7 +2262,6 @@ static void multiVnodeInsertFinalize(void* param, TAOS_RES* tres, int numOfRows)
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
pParentObj->res.code = code; pParentObj->res.code = code;
doFreeInsertSupporter(pParentObj);
tscQueueAsyncRes(pParentObj); tscQueueAsyncRes(pParentObj);
return; return;
} }
@ -2303,7 +2301,11 @@ int32_t tscHandleMultivnodeInsert(SSqlObj *pSql) {
if (pSql->pSubs != NULL) { if (pSql->pSubs != NULL) {
for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) { for(int32_t i = 0; i < pSql->subState.numOfSub; ++i) {
SSqlObj* pSub = pSql->pSubs[i]; SSqlObj* pSub = pSql->pSubs[i];
SInsertSupporter* pSup = calloc(1, sizeof(SInsertSupporter));
pSup->index = i;
pSup->pSql = pSql;
pSub->param = pSup;
tscDebug("%p sub:%p launch sub insert, orderOfSub:%d", pSql, pSub, i); tscDebug("%p sub:%p launch sub insert, orderOfSub:%d", pSql, pSub, i);
if (pSub->res.code != TSDB_CODE_SUCCESS) { if (pSub->res.code != TSDB_CODE_SUCCESS) {
tscHandleInsertRetry(pSql, pSub); tscHandleInsertRetry(pSql, pSub);

52
src/client/src/tscUtil.c
View File

@ -518,7 +518,8 @@ void tscFreeSqlObj(SSqlObj* pSql) {
tscFreeSqlResult(pSql); tscFreeSqlResult(pSql);
tscResetSqlCmdObj(pCmd, false); tscResetSqlCmdObj(pCmd, false);
tfree(pCmd->pTagData); tfree(pCmd->tagData.data);
pCmd->tagData.dataLen = 0;
memset(pCmd->payload, 0, (size_t)pCmd->allocSize); memset(pCmd->payload, 0, (size_t)pCmd->allocSize);
tfree(pCmd->payload); tfree(pCmd->payload);
@ -1937,15 +1938,11 @@ SSqlObj* createSimpleSubObj(SSqlObj* pSql, void (*fp)(), void* param, int32_t cm
pCmd->parseFinished = 1; pCmd->parseFinished = 1;
pCmd->autoCreated = pSql->cmd.autoCreated; pCmd->autoCreated = pSql->cmd.autoCreated;
if (pSql->cmd.pTagData != NULL) { int32_t code = copyTagData(&pNew->cmd.tagData, &pSql->cmd.tagData);
int size = offsetof(STagData, data) + htonl(pSql->cmd.pTagData->dataLen); if (code != TSDB_CODE_SUCCESS) {
pNew->cmd.pTagData = calloc(1, size); tscError("%p new subquery failed, unable to malloc tag data, tableIndex:%d", pSql, 0);
if (pNew->cmd.pTagData == NULL) { free(pNew);
tscError("%p new subquery failed, unable to malloc tag data, tableIndex:%d", pSql, 0); return NULL;
free(pNew);
return NULL;
}
memcpy(pNew->cmd.pTagData, pSql->cmd.pTagData, size);
} }
if (tscAddSubqueryInfo(pCmd) != TSDB_CODE_SUCCESS) { if (tscAddSubqueryInfo(pCmd) != TSDB_CODE_SUCCESS) {
@ -2592,6 +2589,41 @@ void tscSVgroupInfoCopy(SVgroupInfo* dst, const SVgroupInfo* src) {
for(int32_t i = 0; i < dst->numOfEps; ++i) { for(int32_t i = 0; i < dst->numOfEps; ++i) {
tfree(dst->epAddr[i].fqdn); tfree(dst->epAddr[i].fqdn);
dst->epAddr[i].port = src->epAddr[i].port; dst->epAddr[i].port = src->epAddr[i].port;
assert(dst->epAddr[i].fqdn == NULL);
dst->epAddr[i].fqdn = strdup(src->epAddr[i].fqdn); dst->epAddr[i].fqdn = strdup(src->epAddr[i].fqdn);
} }
} }
char* serializeTagData(STagData* pTagData, char* pMsg) {
int32_t n = (int32_t) strlen(pTagData->name);
*(int32_t*) pMsg = htonl(n);
pMsg += sizeof(n);
memcpy(pMsg, pTagData->name, n);
pMsg += n;
*(int32_t*)pMsg = htonl(pTagData->dataLen);
pMsg += sizeof(int32_t);
memcpy(pMsg, pTagData->data, pTagData->dataLen);
pMsg += pTagData->dataLen;
return pMsg;
}
int32_t copyTagData(STagData* dst, const STagData* src) {
dst->dataLen = src->dataLen;
tstrncpy(dst->name, src->name, tListLen(dst->name));
if (dst->dataLen > 0) {
dst->data = malloc(dst->dataLen);
if (dst->data == NULL) {
return -1;
}
memcpy(dst->data, src->data, dst->dataLen);
}
return 0;
}

4
src/common/inc/tglobal.h
View File

@ -57,7 +57,9 @@ extern char tsTempDir[];
//query buffer management //query buffer management
extern int32_t tsQueryBufferSize; // maximum allowed usage buffer for each data node during query processing extern int32_t tsQueryBufferSize; // maximum allowed usage buffer for each data node during query processing
extern int32_t tsRetrieveBlockingModel; // only 50% will be used in query processing extern int32_t tsRetrieveBlockingModel;// retrieve threads will be blocked
extern int32_t tsKeepOriginalColumnName;
// client // client
extern int32_t tsTableMetaKeepTimer; extern int32_t tsTableMetaKeepTimer;

13
src/common/src/tglobal.c
View File

@ -110,6 +110,9 @@ int32_t tsQueryBufferSize = -1;
// in retrieve blocking model, the retrieve threads will wait for the completion of the query processing. // in retrieve blocking model, the retrieve threads will wait for the completion of the query processing.
int32_t tsRetrieveBlockingModel = 0; int32_t tsRetrieveBlockingModel = 0;
// last_row(*), first(*), last_row(ts, col1, col2) query, the result fields will be the original column name
int32_t tsKeepOriginalColumnName = 0;
// db parameters // db parameters
int32_t tsCacheBlockSize = TSDB_DEFAULT_CACHE_BLOCK_SIZE; int32_t tsCacheBlockSize = TSDB_DEFAULT_CACHE_BLOCK_SIZE;
int32_t tsBlocksPerVnode = TSDB_DEFAULT_TOTAL_BLOCKS; int32_t tsBlocksPerVnode = TSDB_DEFAULT_TOTAL_BLOCKS;
@ -897,6 +900,16 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE; cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg); taosInitConfigOption(cfg);
cfg.option = "keepColumnName";
cfg.ptr = &tsKeepOriginalColumnName;
cfg.valType = TAOS_CFG_VTYPE_INT32;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLIENT;
cfg.minValue = 0;
cfg.maxValue = 1;
cfg.ptrLength = 1;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
// locale & charset // locale & charset
cfg.option = "timezone"; cfg.option = "timezone";
cfg.ptr = tsTimezone; cfg.ptr = tsTimezone;

11
src/inc/taosmsg.h
View File

@ -271,6 +271,7 @@ typedef struct {
} SMDCreateTableMsg; } SMDCreateTableMsg;
typedef struct { typedef struct {
int32_t len; // one create table message
char tableId[TSDB_TABLE_FNAME_LEN]; char tableId[TSDB_TABLE_FNAME_LEN];
char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN]; char db[TSDB_ACCT_LEN + TSDB_DB_NAME_LEN];
int8_t igExists; int8_t igExists;
@ -278,9 +279,13 @@ typedef struct {
int16_t numOfTags; int16_t numOfTags;
int16_t numOfColumns; int16_t numOfColumns;
int16_t sqlLen; // the length of SQL, it starts after schema , sql is a null-terminated string int16_t sqlLen; // the length of SQL, it starts after schema , sql is a null-terminated string
int32_t contLen;
int8_t reserved[16]; int8_t reserved[16];
char schema[]; char schema[];
} SCreateTableMsg;
typedef struct {
int32_t numOfTables;
int32_t contLen;
} SCMCreateTableMsg; } SCMCreateTableMsg;
typedef struct { typedef struct {
@ -735,8 +740,8 @@ typedef struct SMultiTableMeta {
typedef struct { typedef struct {
int32_t dataLen; int32_t dataLen;
char name[TSDB_TABLE_FNAME_LEN]; char name[TSDB_TABLE_FNAME_LEN];
char data[TSDB_MAX_TAGS_LEN + TD_KV_ROW_HEAD_SIZE + sizeof(SColIdx) * TSDB_MAX_TAGS]; char *data;
} STagData; } STagData;
/* /*

131
src/mnode/src/mnodeTable.c
View File

@ -722,10 +722,17 @@ static void mnodeExtractTableName(char* tableId, char* name) {
static int32_t mnodeProcessCreateTableMsg(SMnodeMsg *pMsg) { static int32_t mnodeProcessCreateTableMsg(SMnodeMsg *pMsg) {
SCMCreateTableMsg *pCreate = pMsg->rpcMsg.pCont; SCMCreateTableMsg *pCreate = pMsg->rpcMsg.pCont;
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(pCreate->db); int32_t numOfTables = htonl(pCreate->numOfTables);
int32_t contentLen = htonl(pCreate->contLen);
if (numOfTables == 0 || contentLen == 0) {
// todo return error
}
SCreateTableMsg *p = (SCreateTableMsg*)((char*) pCreate + sizeof(SCMCreateTableMsg));
if (pMsg->pDb == NULL) pMsg->pDb = mnodeGetDb(p->db);
if (pMsg->pDb == NULL) { if (pMsg->pDb == NULL) {
mError("msg:%p, app:%p table:%s, failed to create, db not selected", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableId); mError("msg:%p, app:%p table:%s, failed to create, db not selected", pMsg, pMsg->rpcMsg.ahandle, p->tableId);
return TSDB_CODE_MND_DB_NOT_SELECTED; return TSDB_CODE_MND_DB_NOT_SELECTED;
} }
@ -734,28 +741,28 @@ static int32_t mnodeProcessCreateTableMsg(SMnodeMsg *pMsg) {
return TSDB_CODE_MND_DB_IN_DROPPING; return TSDB_CODE_MND_DB_IN_DROPPING;
} }
if (pMsg->pTable == NULL) pMsg->pTable = mnodeGetTable(pCreate->tableId); if (pMsg->pTable == NULL) pMsg->pTable = mnodeGetTable(p->tableId);
if (pMsg->pTable != NULL && pMsg->retry == 0) { if (pMsg->pTable != NULL && pMsg->retry == 0) {
if (pCreate->getMeta) { if (p->getMeta) {
mDebug("msg:%p, app:%p table:%s, continue to get meta", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableId); mDebug("msg:%p, app:%p table:%s, continue to get meta", pMsg, pMsg->rpcMsg.ahandle, p->tableId);
return mnodeGetChildTableMeta(pMsg); return mnodeGetChildTableMeta(pMsg);
} else if (pCreate->igExists) { } else if (p->igExists) {
mDebug("msg:%p, app:%p table:%s, is already exist", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableId); mDebug("msg:%p, app:%p table:%s, is already exist", pMsg, pMsg->rpcMsg.ahandle, p->tableId);
return TSDB_CODE_SUCCESS; return TSDB_CODE_SUCCESS;
} else { } else {
mError("msg:%p, app:%p table:%s, failed to create, table already exist", pMsg, pMsg->rpcMsg.ahandle, mError("msg:%p, app:%p table:%s, failed to create, table already exist", pMsg, pMsg->rpcMsg.ahandle,
pCreate->tableId); p->tableId);
return TSDB_CODE_MND_TABLE_ALREADY_EXIST; return TSDB_CODE_MND_TABLE_ALREADY_EXIST;
} }
} }
if (pCreate->numOfTags != 0) { if (p->numOfTags != 0) {
mDebug("msg:%p, app:%p table:%s, create stable msg is received from thandle:%p", pMsg, pMsg->rpcMsg.ahandle, mDebug("msg:%p, app:%p table:%s, create stable msg is received from thandle:%p", pMsg, pMsg->rpcMsg.ahandle,
pCreate->tableId, pMsg->rpcMsg.handle); p->tableId, pMsg->rpcMsg.handle);
return mnodeProcessCreateSuperTableMsg(pMsg); return mnodeProcessCreateSuperTableMsg(pMsg);
} else { } else {
mDebug("msg:%p, app:%p table:%s, create ctable msg is received from thandle:%p", pMsg, pMsg->rpcMsg.ahandle, mDebug("msg:%p, app:%p table:%s, create ctable msg is received from thandle:%p", pMsg, pMsg->rpcMsg.ahandle,
pCreate->tableId, pMsg->rpcMsg.handle); p->tableId, pMsg->rpcMsg.handle);
return mnodeProcessCreateChildTableMsg(pMsg); return mnodeProcessCreateChildTableMsg(pMsg);
} }
} }
@ -861,7 +868,13 @@ static int32_t mnodeCreateSuperTableCb(SMnodeMsg *pMsg, int32_t code) {
static int32_t mnodeProcessCreateSuperTableMsg(SMnodeMsg *pMsg) { static int32_t mnodeProcessCreateSuperTableMsg(SMnodeMsg *pMsg) {
if (pMsg == NULL) return TSDB_CODE_MND_APP_ERROR; if (pMsg == NULL) return TSDB_CODE_MND_APP_ERROR;
SCMCreateTableMsg *pCreate = pMsg->rpcMsg.pCont; SCMCreateTableMsg *pCreate1 = pMsg->rpcMsg.pCont;
if (pCreate1->numOfTables == 0) {
// todo return to error message
}
SCreateTableMsg* pCreate = (SCreateTableMsg*)((char*)pCreate1 + sizeof(SCMCreateTableMsg));
SSTableObj * pStable = calloc(1, sizeof(SSTableObj)); SSTableObj * pStable = calloc(1, sizeof(SSTableObj));
if (pStable == NULL) { if (pStable == NULL) {
mError("msg:%p, app:%p table:%s, failed to create, no enough memory", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableId); mError("msg:%p, app:%p table:%s, failed to create, no enough memory", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableId);
@ -1601,8 +1614,11 @@ static void mnodeProcessDropSuperTableRsp(SRpcMsg *rpcMsg) {
mInfo("drop stable rsp received, result:%s", tstrerror(rpcMsg->code)); mInfo("drop stable rsp received, result:%s", tstrerror(rpcMsg->code));
} }
static void *mnodeBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SCTableObj *pTable) { static void *mnodeBuildCreateChildTableMsg(SCMCreateTableMsg *pCreateMsg, SCTableObj *pTable) {
STagData * pTagData = NULL; SCreateTableMsg* pMsg = (SCreateTableMsg*) ((char*)pCreateMsg + sizeof(SCMCreateTableMsg));
char* tagData = NULL;
int32_t tagDataLen = 0; int32_t tagDataLen = 0;
int32_t totalCols = 0; int32_t totalCols = 0;
int32_t contLen = 0; int32_t contLen = 0;
@ -1610,9 +1626,13 @@ static void *mnodeBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SCTableObj *
totalCols = pTable->superTable->numOfColumns + pTable->superTable->numOfTags; totalCols = pTable->superTable->numOfColumns + pTable->superTable->numOfTags;
contLen = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + pTable->sqlLen; contLen = sizeof(SMDCreateTableMsg) + totalCols * sizeof(SSchema) + pTable->sqlLen;
if (pMsg != NULL) { if (pMsg != NULL) {
pTagData = (STagData *)pMsg->schema; int32_t nameLen = htonl(*(int32_t*)pMsg->schema);
tagDataLen = htonl(pTagData->dataLen); char* p = pMsg->schema + nameLen + sizeof(int32_t);
tagDataLen = htonl(*(int32_t*) p);
contLen += tagDataLen; contLen += tagDataLen;
tagData = p + sizeof(int32_t);
} }
} else { } else {
totalCols = pTable->numOfColumns; totalCols = pTable->numOfColumns;
@ -1664,7 +1684,7 @@ static void *mnodeBuildCreateChildTableMsg(SCMCreateTableMsg *pMsg, SCTableObj *
} }
if (pTable->info.type == TSDB_CHILD_TABLE && pMsg != NULL) { if (pTable->info.type == TSDB_CHILD_TABLE && pMsg != NULL) {
memcpy(pCreate->data + totalCols * sizeof(SSchema), pTagData->data, tagDataLen); memcpy(pCreate->data + totalCols * sizeof(SSchema), tagData, tagDataLen);
} }
if (pTable->info.type == TSDB_STREAM_TABLE) { if (pTable->info.type == TSDB_STREAM_TABLE) {
@ -1702,7 +1722,8 @@ static int32_t mnodeDoCreateChildTableFp(SMnodeMsg *pMsg) {
static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) { static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) {
SCTableObj *pTable = (SCTableObj *)pMsg->pTable; SCTableObj *pTable = (SCTableObj *)pMsg->pTable;
SCMCreateTableMsg *pCreate = pMsg->rpcMsg.pCont;
SCreateTableMsg *pCreate = (SCreateTableMsg*) ((char*)pMsg->rpcMsg.pCont + sizeof(SCMCreateTableMsg));
assert(pTable); assert(pTable);
if (code == TSDB_CODE_SUCCESS) { if (code == TSDB_CODE_SUCCESS) {
@ -1730,40 +1751,42 @@ static int32_t mnodeDoCreateChildTableCb(SMnodeMsg *pMsg, int32_t code) {
static int32_t mnodeDoCreateChildTable(SMnodeMsg *pMsg, int32_t tid) { static int32_t mnodeDoCreateChildTable(SMnodeMsg *pMsg, int32_t tid) {
SVgObj *pVgroup = pMsg->pVgroup; SVgObj *pVgroup = pMsg->pVgroup;
SCMCreateTableMsg *pCreate = pMsg->rpcMsg.pCont;
SCMCreateTableMsg *p1 = pMsg->rpcMsg.pCont;
SCreateTableMsg *pCreate = (SCreateTableMsg*)((char*)p1 + sizeof(SCMCreateTableMsg));
SCTableObj *pTable = calloc(1, sizeof(SCTableObj)); SCTableObj *pTable = calloc(1, sizeof(SCTableObj));
if (pTable == NULL) { if (pTable == NULL) {
mError("msg:%p, app:%p table:%s, failed to alloc memory", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableId); mError("msg:%p, app:%p table:%s, failed to alloc memory", pMsg, pMsg->rpcMsg.ahandle, pCreate->tableId);
return TSDB_CODE_MND_OUT_OF_MEMORY; return TSDB_CODE_MND_OUT_OF_MEMORY;
} }
if (pCreate->numOfColumns == 0) { pTable->info.type = (pCreate->numOfColumns == 0)? TSDB_CHILD_TABLE:TSDB_NORMAL_TABLE;
pTable->info.type = TSDB_CHILD_TABLE; pTable->info.tableId = strdup(pCreate->tableId);
} else {
pTable->info.type = TSDB_NORMAL_TABLE;
}
pTable->info.tableId = strdup(pCreate->tableId);
pTable->createdTime = taosGetTimestampMs(); pTable->createdTime = taosGetTimestampMs();
pTable->tid = tid; pTable->tid = tid;
pTable->vgId = pVgroup->vgId; pTable->vgId = pVgroup->vgId;
if (pTable->info.type == TSDB_CHILD_TABLE) { if (pTable->info.type == TSDB_CHILD_TABLE) {
STagData *pTagData = (STagData *)pCreate->schema; // it is a tag key int32_t nameLen = htonl(*(int32_t*) pCreate->schema);
char* name = (char*)pCreate->schema + sizeof(int32_t);
char stableName[TSDB_TABLE_FNAME_LEN] = {0};
memcpy(stableName, name, nameLen);
char prefix[64] = {0}; char prefix[64] = {0};
size_t prefixLen = tableIdPrefix(pMsg->pDb->name, prefix, 64); size_t prefixLen = tableIdPrefix(pMsg->pDb->name, prefix, 64);
if (0 != strncasecmp(prefix, pTagData->name, prefixLen)) { if (0 != strncasecmp(prefix, stableName, prefixLen)) {
mError("msg:%p, app:%p table:%s, corresponding super table:%s not in this db", pMsg, pMsg->rpcMsg.ahandle, mError("msg:%p, app:%p table:%s, corresponding super table:%s not in this db", pMsg, pMsg->rpcMsg.ahandle,
pCreate->tableId, pTagData->name); pCreate->tableId, stableName);
mnodeDestroyChildTable(pTable); mnodeDestroyChildTable(pTable);
return TSDB_CODE_TDB_INVALID_CREATE_TB_MSG; return TSDB_CODE_TDB_INVALID_CREATE_TB_MSG;
} }
if (pMsg->pSTable == NULL) pMsg->pSTable = mnodeGetSuperTable(pTagData->name); if (pMsg->pSTable == NULL) pMsg->pSTable = mnodeGetSuperTable(stableName);
if (pMsg->pSTable == NULL) { if (pMsg->pSTable == NULL) {
mError("msg:%p, app:%p table:%s, corresponding super table:%s does not exist", pMsg, pMsg->rpcMsg.ahandle, mError("msg:%p, app:%p table:%s, corresponding super table:%s does not exist", pMsg, pMsg->rpcMsg.ahandle,
pCreate->tableId, pTagData->name); pCreate->tableId, stableName);
mnodeDestroyChildTable(pTable); mnodeDestroyChildTable(pTable);
return TSDB_CODE_MND_INVALID_TABLE_NAME; return TSDB_CODE_MND_INVALID_TABLE_NAME;
} }
@ -1841,7 +1864,9 @@ static int32_t mnodeDoCreateChildTable(SMnodeMsg *pMsg, int32_t tid) {
} }
static int32_t mnodeProcessCreateChildTableMsg(SMnodeMsg *pMsg) { static int32_t mnodeProcessCreateChildTableMsg(SMnodeMsg *pMsg) {
SCMCreateTableMsg *pCreate = pMsg->rpcMsg.pCont; //SCMCreateTableMsg* p1 = pMsg->rpcMsg.pCont; // there are several tables here.
SCreateTableMsg* pCreate = (SCreateTableMsg*)(pMsg->rpcMsg.pCont + sizeof(SCMCreateTableMsg));
int32_t code = grantCheck(TSDB_GRANT_TIMESERIES); int32_t code = grantCheck(TSDB_GRANT_TIMESERIES);
if (code != TSDB_CODE_SUCCESS) { if (code != TSDB_CODE_SUCCESS) {
mError("msg:%p, app:%p table:%s, failed to create, grant timeseries failed", pMsg, pMsg->rpcMsg.ahandle, mError("msg:%p, app:%p table:%s, failed to create, grant timeseries failed", pMsg, pMsg->rpcMsg.ahandle,
@ -2189,15 +2214,23 @@ static int32_t mnodeDoGetChildTableMeta(SMnodeMsg *pMsg, STableMetaMsg *pMeta) {
static int32_t mnodeAutoCreateChildTable(SMnodeMsg *pMsg) { static int32_t mnodeAutoCreateChildTable(SMnodeMsg *pMsg) {
STableInfoMsg *pInfo = pMsg->rpcMsg.pCont; STableInfoMsg *pInfo = pMsg->rpcMsg.pCont;
STagData *pTags = (STagData *)pInfo->tags;
int32_t tagLen = htonl(pTags->dataLen); char* p = pInfo->tags;
if (pTags->name[0] == 0) { int32_t nameLen = htonl(*(int32_t*) p);
mError("msg:%p, app:%p table:%s, failed to create table on demand for stable is empty, tagLen:%d", pMsg, p += sizeof(int32_t);
p += nameLen;
int32_t tagLen = htonl(*(int32_t*) p);
p += sizeof(int32_t);
int32_t totalLen = nameLen + tagLen + sizeof(int32_t)*2;
if (tagLen == 0 || nameLen == 0) {
mError("msg:%p, app:%p table:%s, failed to create table on demand for super table is empty, tagLen:%d", pMsg,
pMsg->rpcMsg.ahandle, pInfo->tableId, tagLen); pMsg->rpcMsg.ahandle, pInfo->tableId, tagLen);
return TSDB_CODE_MND_INVALID_STABLE_NAME; return TSDB_CODE_MND_INVALID_STABLE_NAME;
} }
int32_t contLen = sizeof(SCMCreateTableMsg) + offsetof(STagData, data) + tagLen; int32_t contLen = sizeof(SCMCreateTableMsg) + sizeof(SCreateTableMsg) + totalLen;
SCMCreateTableMsg *pCreateMsg = calloc(1, contLen); SCMCreateTableMsg *pCreateMsg = calloc(1, contLen);
if (pCreateMsg == NULL) { if (pCreateMsg == NULL) {
mError("msg:%p, app:%p table:%s, failed to create table while get meta info, no enough memory", pMsg, mError("msg:%p, app:%p table:%s, failed to create table while get meta info, no enough memory", pMsg,
@ -2205,16 +2238,24 @@ static int32_t mnodeAutoCreateChildTable(SMnodeMsg *pMsg) {
return TSDB_CODE_MND_OUT_OF_MEMORY; return TSDB_CODE_MND_OUT_OF_MEMORY;
} }
size_t size = sizeof(pInfo->tableId); SCreateTableMsg* pCreate = (SCreateTableMsg*) ((char*) pCreateMsg + sizeof(SCMCreateTableMsg));
tstrncpy(pCreateMsg->tableId, pInfo->tableId, size);
tstrncpy(pCreateMsg->db, pMsg->pDb->name, sizeof(pCreateMsg->db)); size_t size = tListLen(pInfo->tableId);
pCreateMsg->igExists = 1; tstrncpy(pCreate->tableId, pInfo->tableId, size);
pCreateMsg->getMeta = 1; tstrncpy(pCreate->db, pMsg->pDb->name, sizeof(pCreate->db));
pCreate->igExists = 1;
pCreate->getMeta = 1;
pCreateMsg->numOfTables = htonl(1);
pCreateMsg->contLen = htonl(contLen); pCreateMsg->contLen = htonl(contLen);
memcpy(pCreateMsg->schema, pTags, contLen - sizeof(SCMCreateTableMsg)); memcpy(pCreate->schema, pInfo->tags, totalLen);
char name[TSDB_TABLE_FNAME_LEN] = {0};
memcpy(name, pInfo->tags + sizeof(int32_t), nameLen);
mDebug("msg:%p, app:%p table:%s, start to create on demand, tagLen:%d stable:%s", pMsg, pMsg->rpcMsg.ahandle, mDebug("msg:%p, app:%p table:%s, start to create on demand, tagLen:%d stable:%s", pMsg, pMsg->rpcMsg.ahandle,
pInfo->tableId, tagLen, pTags->name); pInfo->tableId, tagLen, name);
if (pMsg->rpcMsg.pCont != pMsg->pCont) { if (pMsg->rpcMsg.pCont != pMsg->pCont) {
tfree(pMsg->rpcMsg.pCont); tfree(pMsg->rpcMsg.pCont);

2
src/os/inc/osString.h
View File

@ -39,7 +39,7 @@ extern "C" {
do { \ do { \
strncpy((dst), (src), (size)); \ strncpy((dst), (src), (size)); \
(dst)[(size)-1] = 0; \ (dst)[(size)-1] = 0; \
} while (0); } while (0)
#ifndef TAOS_OS_FUNC_STRING_STR2INT64 #ifndef TAOS_OS_FUNC_STRING_STR2INT64
int64_t tsosStr2int64(char *str); int64_t tsosStr2int64(char *str);

43
src/query/inc/qSqlparser.h
View File

@ -73,23 +73,27 @@ typedef struct SQuerySQL {
SStrToken selectToken; // sql string SStrToken selectToken; // sql string
} SQuerySQL; } SQuerySQL;
typedef struct SCreatedTableInfo {
SStrToken name; // table name token
SStrToken stableName; // super table name token , for using clause
SArray *pTagVals; // create by using super table, tag value
char *fullname; // table full name
STagData tagdata; // true tag data, super table full name is in STagData
int8_t igExist; // ignore if exists
} SCreatedTableInfo;
typedef struct SCreateTableSQL { typedef struct SCreateTableSQL {
struct SStrToken name; // meter name, create table [meterName] xxx SStrToken name; // table name, create table [name] xxx
bool existCheck; int8_t type; // create normal table/from super table/ stream
bool existCheck;
int8_t type; // create normal table/from super table/ stream
struct { struct {
SArray *pTagColumns; // SArray<TAOS_FIELD> SArray *pTagColumns; // SArray<TAOS_FIELD>
SArray *pColumns; // SArray<TAOS_FIELD> SArray *pColumns; // SArray<TAOS_FIELD>
} colInfo; } colInfo;
struct { SArray *childTableInfo; // SArray<SCreatedTableInfo>
SStrToken stableName; // super table name, for using clause SQuerySQL *pSelect;
SArray *pTagVals; // create by using metric, tag value
STagData tagdata;
} usingInfo;
SQuerySQL *pSelect;
} SCreateTableSQL; } SCreateTableSQL;
typedef struct SAlterTableSQL { typedef struct SAlterTableSQL {
@ -241,22 +245,23 @@ SQuerySQL *tSetQuerySQLElems(SStrToken *pSelectToken, tSQLExprList *pSelection,
SArray *pGroupby, SArray *pSortOrder, SIntervalVal *pInterval, SArray *pGroupby, SArray *pSortOrder, SIntervalVal *pInterval,
SStrToken *pSliding, SArray *pFill, SLimitVal *pLimit, SLimitVal *pGLimit); SStrToken *pSliding, SArray *pFill, SLimitVal *pLimit, SLimitVal *pGLimit);
SCreateTableSQL *tSetCreateSQLElems(SArray *pCols, SArray *pTags, SStrToken *pMetricName, SCreateTableSQL *tSetCreateSQLElems(SArray *pCols, SArray *pTags, SQuerySQL *pSelect, int32_t type);
SArray *pTagVals, SQuerySQL *pSelect, int32_t type);
void tSQLExprNodeDestroy(tSQLExpr *pExpr); void tSQLExprNodeDestroy(tSQLExpr *pExpr);
SAlterTableSQL *tAlterTableSQLElems(SStrToken *pMeterName, SArray *pCols, SArray *pVals, int32_t type); SAlterTableSQL *tAlterTableSQLElems(SStrToken *pTableName, SArray *pCols, SArray *pVals, int32_t type);
SCreatedTableInfo createNewChildTableInfo(SStrToken *pTableName, SArray *pTagVals, SStrToken *pToken, SStrToken* igExists);
void destroyAllSelectClause(SSubclauseInfo *pSql); void destroyAllSelectClause(SSubclauseInfo *pSql);
void doDestroyQuerySql(SQuerySQL *pSql); void doDestroyQuerySql(SQuerySQL *pSql);
void freeCreateTableInfo(void* p);
SSqlInfo * setSQLInfo(SSqlInfo *pInfo, void *pSqlExprInfo, SStrToken *pMeterName, int32_t type); SSqlInfo *setSQLInfo(SSqlInfo *pInfo, void *pSqlExprInfo, SStrToken *pTableName, int32_t type);
SSubclauseInfo *setSubclause(SSubclauseInfo *pClause, void *pSqlExprInfo); SSubclauseInfo *setSubclause(SSubclauseInfo *pClause, void *pSqlExprInfo);
SSubclauseInfo *appendSelectClause(SSubclauseInfo *pInfo, void *pSubclause); SSubclauseInfo *appendSelectClause(SSubclauseInfo *pInfo, void *pSubclause);
void setCreatedTableName(SSqlInfo *pInfo, SStrToken *pMeterName, SStrToken *pIfNotExists); void setCreatedTableName(SSqlInfo *pInfo, SStrToken *pTableNameToken, SStrToken *pIfNotExists);
void SQLInfoDestroy(SSqlInfo *pInfo); void SQLInfoDestroy(SSqlInfo *pInfo);

182
src/query/inc/sql.y
View File

@ -168,12 +168,12 @@ ids(A) ::= ID(X). {A = X; }
ids(A) ::= STRING(X). {A = X; } ids(A) ::= STRING(X). {A = X; }
%type ifexists {SStrToken} %type ifexists {SStrToken}
ifexists(X) ::= IF EXISTS. {X.n = 1;} ifexists(X) ::= IF EXISTS. { X.n = 1;}
ifexists(X) ::= . {X.n = 0;} ifexists(X) ::= . { X.n = 0;}
%type ifnotexists {SStrToken} %type ifnotexists {SStrToken}
ifnotexists(X) ::= IF NOT EXISTS. {X.n = 1;} ifnotexists(X) ::= IF NOT EXISTS. { X.n = 1;}
ifnotexists(X) ::= . {X.n = 0;} ifnotexists(X) ::= . { X.n = 0;}
/////////////////////////////////THE CREATE STATEMENT/////////////////////////////////////// /////////////////////////////////THE CREATE STATEMENT///////////////////////////////////////
//create option for dnode/db/user/account //create option for dnode/db/user/account
@ -183,32 +183,32 @@ cmd ::= CREATE ACCOUNT ids(X) PASS ids(Y) acct_optr(Z).
cmd ::= CREATE DATABASE ifnotexists(Z) ids(X) db_optr(Y). { setCreateDBSQL(pInfo, TSDB_SQL_CREATE_DB, &X, &Y, &Z);} cmd ::= CREATE DATABASE ifnotexists(Z) ids(X) db_optr(Y). { setCreateDBSQL(pInfo, TSDB_SQL_CREATE_DB, &X, &Y, &Z);}
cmd ::= CREATE USER ids(X) PASS ids(Y). { setCreateUserSQL(pInfo, &X, &Y);} cmd ::= CREATE USER ids(X) PASS ids(Y). { setCreateUserSQL(pInfo, &X, &Y);}
pps(Y) ::= . {Y.n = 0; } pps(Y) ::= . { Y.n = 0; }
pps(Y) ::= PPS INTEGER(X). {Y = X; } pps(Y) ::= PPS INTEGER(X). { Y = X; }
tseries(Y) ::= . {Y.n = 0; } tseries(Y) ::= . { Y.n = 0; }
tseries(Y) ::= TSERIES INTEGER(X). {Y = X; } tseries(Y) ::= TSERIES INTEGER(X). { Y = X; }
dbs(Y) ::= . {Y.n = 0; } dbs(Y) ::= . { Y.n = 0; }
dbs(Y) ::= DBS INTEGER(X). {Y = X; } dbs(Y) ::= DBS INTEGER(X). { Y = X; }
streams(Y) ::= . {Y.n = 0; } streams(Y) ::= . { Y.n = 0; }
streams(Y) ::= STREAMS INTEGER(X). {Y = X; } streams(Y) ::= STREAMS INTEGER(X). { Y = X; }
storage(Y) ::= . {Y.n = 0; } storage(Y) ::= . { Y.n = 0; }
storage(Y) ::= STORAGE INTEGER(X). {Y = X; } storage(Y) ::= STORAGE INTEGER(X). { Y = X; }
qtime(Y) ::= . {Y.n = 0; } qtime(Y) ::= . { Y.n = 0; }
qtime(Y) ::= QTIME INTEGER(X). {Y = X; } qtime(Y) ::= QTIME INTEGER(X). { Y = X; }
users(Y) ::= . {Y.n = 0; } users(Y) ::= . { Y.n = 0; }
users(Y) ::= USERS INTEGER(X). {Y = X; } users(Y) ::= USERS INTEGER(X). { Y = X; }
conns(Y) ::= . {Y.n = 0; } conns(Y) ::= . { Y.n = 0; }
conns(Y) ::= CONNS INTEGER(X). {Y = X; } conns(Y) ::= CONNS INTEGER(X). { Y = X; }
state(Y) ::= . {Y.n = 0; } state(Y) ::= . { Y.n = 0; }
state(Y) ::= STATE ids(X). {Y = X; } state(Y) ::= STATE ids(X). { Y = X; }
%type acct_optr {SCreateAcctSQL} %type acct_optr {SCreateAcctSQL}
acct_optr(Y) ::= pps(C) tseries(D) storage(P) streams(F) qtime(Q) dbs(E) users(K) conns(L) state(M). { acct_optr(Y) ::= pps(C) tseries(D) storage(P) streams(F) qtime(Q) dbs(E) users(K) conns(L) state(M). {
@ -269,7 +269,7 @@ alter_db_optr(Y) ::= alter_db_optr(Z) blocks(X). { Y = Z; Y.numOfBlocks = s
alter_db_optr(Y) ::= alter_db_optr(Z) comp(X). { Y = Z; Y.compressionLevel = strtol(X.z, NULL, 10); } alter_db_optr(Y) ::= alter_db_optr(Z) comp(X). { Y = Z; Y.compressionLevel = strtol(X.z, NULL, 10); }
alter_db_optr(Y) ::= alter_db_optr(Z) wal(X). { Y = Z; Y.walLevel = strtol(X.z, NULL, 10); } alter_db_optr(Y) ::= alter_db_optr(Z) wal(X). { Y = Z; Y.walLevel = strtol(X.z, NULL, 10); }
alter_db_optr(Y) ::= alter_db_optr(Z) fsync(X). { Y = Z; Y.fsyncPeriod = strtol(X.z, NULL, 10); } alter_db_optr(Y) ::= alter_db_optr(Z) fsync(X). { Y = Z; Y.fsyncPeriod = strtol(X.z, NULL, 10); }
alter_db_optr(Y) ::= alter_db_optr(Z) update(X). { Y = Z; Y.update = strtol(X.z, NULL, 10); } alter_db_optr(Y) ::= alter_db_optr(Z) update(X). { Y = Z; Y.update = strtol(X.z, NULL, 10); }
%type typename {TAOS_FIELD} %type typename {TAOS_FIELD}
typename(A) ::= ids(X). { typename(A) ::= ids(X). {
@ -279,13 +279,13 @@ typename(A) ::= ids(X). {
//define binary type, e.g., binary(10), nchar(10) //define binary type, e.g., binary(10), nchar(10)
typename(A) ::= ids(X) LP signed(Y) RP. { typename(A) ::= ids(X) LP signed(Y) RP. {
if (Y <= 0) { if (Y <= 0) {
X.type = 0; X.type = 0;
tSQLSetColumnType(&A, &X); tSQLSetColumnType(&A, &X);
} else { } else {
X.type = -Y; // negative value of name length X.type = -Y; // negative value of name length
tSQLSetColumnType(&A, &X); tSQLSetColumnType(&A, &X);
} }
} }
%type signed {int64_t} %type signed {int64_t}
@ -294,36 +294,60 @@ signed(A) ::= PLUS INTEGER(X). { A = strtol(X.z, NULL, 10); }
signed(A) ::= MINUS INTEGER(X). { A = -strtol(X.z, NULL, 10);} signed(A) ::= MINUS INTEGER(X). { A = -strtol(X.z, NULL, 10);}
////////////////////////////////// The CREATE TABLE statement /////////////////////////////// ////////////////////////////////// The CREATE TABLE statement ///////////////////////////////
cmd ::= CREATE TABLE ifnotexists(Y) ids(X) cpxName(Z) create_table_args. { cmd ::= CREATE TABLE create_table_args. {}
X.n += Z.n; cmd ::= CREATE TABLE create_table_list(Z). { pInfo->type = TSDB_SQL_CREATE_TABLE; pInfo->pCreateTableInfo = Z;}
setCreatedTableName(pInfo, &X, &Y);
%type create_table_list{SCreateTableSQL*}
%destructor create_table_list{destroyCreateTableSQL($$);}
create_table_list(A) ::= create_from_stable(Z). {
SCreateTableSQL* pCreateTable = calloc(1, sizeof(SCreateTableSQL));
pCreateTable->childTableInfo = taosArrayInit(4, sizeof(SCreatedTableInfo));
taosArrayPush(pCreateTable->childTableInfo, &Z);
pCreateTable->type = TSQL_CREATE_TABLE_FROM_STABLE;
A = pCreateTable;
}
create_table_list(A) ::= create_table_list(X) create_from_stable(Z). {
taosArrayPush(X->childTableInfo, &Z);
A = X;
} }
%type create_table_args{SCreateTableSQL*} %type create_table_args{SCreateTableSQL*}
create_table_args(A) ::= LP columnlist(X) RP. { create_table_args(A) ::= ifnotexists(U) ids(V) cpxName(Z) LP columnlist(X) RP. {
A = tSetCreateSQLElems(X, NULL, NULL, NULL, NULL, TSQL_CREATE_TABLE); A = tSetCreateSQLElems(X, NULL, NULL, TSQL_CREATE_TABLE);
setSQLInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE); setSQLInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE);
V.n += Z.n;
setCreatedTableName(pInfo, &V, &U);
} }
// create super table // create super table
create_table_args(A) ::= LP columnlist(X) RP TAGS LP columnlist(Y) RP. { create_table_args(A) ::= ifnotexists(U) ids(V) cpxName(Z) LP columnlist(X) RP TAGS LP columnlist(Y) RP. {
A = tSetCreateSQLElems(X, Y, NULL, NULL, NULL, TSQL_CREATE_STABLE); A = tSetCreateSQLElems(X, Y, NULL, TSQL_CREATE_STABLE);
setSQLInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE); setSQLInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE);
V.n += Z.n;
setCreatedTableName(pInfo, &V, &U);
} }
// create table by using super table // create table by using super table
// create table table_name using super_table_name tags(tag_values1, tag_values2) // create table table_name using super_table_name tags(tag_values1, tag_values2)
create_table_args(A) ::= USING ids(X) cpxName(F) TAGS LP tagitemlist(Y) RP. { %type create_from_stable{SCreatedTableInfo}
X.n += F.n; create_from_stable(A) ::= ifnotexists(U) ids(V) cpxName(Z) USING ids(X) cpxName(F) TAGS LP tagitemlist(Y) RP. {
A = tSetCreateSQLElems(NULL, NULL, &X, Y, NULL, TSQL_CREATE_TABLE_FROM_STABLE); X.n += F.n;
setSQLInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE); V.n += Z.n;
A = createNewChildTableInfo(&X, Y, &V, &U);
} }
// create stream // create stream
// create table table_name as select count(*) from super_table_name interval(time) // create table table_name as select count(*) from super_table_name interval(time)
create_table_args(A) ::= AS select(S). { create_table_args(A) ::= ifnotexists(U) ids(V) cpxName(Z) AS select(S). {
A = tSetCreateSQLElems(NULL, NULL, NULL, NULL, S, TSQL_CREATE_STREAM); A = tSetCreateSQLElems(NULL, NULL, S, TSQL_CREATE_STREAM);
setSQLInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE); setSQLInfo(pInfo, A, NULL, TSDB_SQL_CREATE_TABLE);
U.n += Z.n;
setCreatedTableName(pInfo, &U, &V);
} }
%type column{TAOS_FIELD} %type column{TAOS_FIELD}
@ -335,7 +359,7 @@ columnlist(A) ::= column(X). {A = taosArrayInit(4, sizeof(T
// The information used for a column is the name and type of column: // The information used for a column is the name and type of column:
// tinyint smallint int bigint float double bool timestamp binary(x) nchar(x) // tinyint smallint int bigint float double bool timestamp binary(x) nchar(x)
column(A) ::= ids(X) typename(Y). { column(A) ::= ids(X) typename(Y). {
tSQLSetColumnInfo(&A, &X, &Y); tSQLSetColumnInfo(&A, &X, &Y);
} }
%type tagitemlist {SArray*} %type tagitemlist {SArray*}
@ -345,10 +369,10 @@ column(A) ::= ids(X) typename(Y). {
tagitemlist(A) ::= tagitemlist(X) COMMA tagitem(Y). { A = tVariantListAppend(X, &Y, -1); } tagitemlist(A) ::= tagitemlist(X) COMMA tagitem(Y). { A = tVariantListAppend(X, &Y, -1); }
tagitemlist(A) ::= tagitem(X). { A = tVariantListAppend(NULL, &X, -1); } tagitemlist(A) ::= tagitem(X). { A = tVariantListAppend(NULL, &X, -1); }
tagitem(A) ::= INTEGER(X). {toTSDBType(X.type); tVariantCreate(&A, &X); } tagitem(A) ::= INTEGER(X). { toTSDBType(X.type); tVariantCreate(&A, &X); }
tagitem(A) ::= FLOAT(X). {toTSDBType(X.type); tVariantCreate(&A, &X); } tagitem(A) ::= FLOAT(X). { toTSDBType(X.type); tVariantCreate(&A, &X); }
tagitem(A) ::= STRING(X). {toTSDBType(X.type); tVariantCreate(&A, &X); } tagitem(A) ::= STRING(X). { toTSDBType(X.type); tVariantCreate(&A, &X); }
tagitem(A) ::= BOOL(X). {toTSDBType(X.type); tVariantCreate(&A, &X); } tagitem(A) ::= BOOL(X). { toTSDBType(X.type); tVariantCreate(&A, &X); }
tagitem(A) ::= NULL(X). { X.type = 0; tVariantCreate(&A, &X); } tagitem(A) ::= NULL(X). { X.type = 0; tVariantCreate(&A, &X); }
tagitem(A) ::= MINUS(X) INTEGER(Y).{ tagitem(A) ::= MINUS(X) INTEGER(Y).{
@ -445,11 +469,11 @@ tablelist(A) ::= ids(X) cpxName(Y). {
} }
tablelist(A) ::= ids(X) cpxName(Y) ids(Z). { tablelist(A) ::= ids(X) cpxName(Y) ids(Z). {
toTSDBType(X.type); toTSDBType(X.type);
toTSDBType(Z.type); toTSDBType(Z.type);
X.n += Y.n; X.n += Y.n;
A = tVariantListAppendToken(NULL, &X, -1); A = tVariantListAppendToken(NULL, &X, -1);
A = tVariantListAppendToken(A, &Z, -1); A = tVariantListAppendToken(A, &Z, -1);
} }
tablelist(A) ::= tablelist(Y) COMMA ids(X) cpxName(Z). { tablelist(A) ::= tablelist(Y) COMMA ids(X) cpxName(Z). {
@ -460,11 +484,11 @@ tablelist(A) ::= tablelist(Y) COMMA ids(X) cpxName(Z). {
} }
tablelist(A) ::= tablelist(Y) COMMA ids(X) cpxName(Z) ids(F). { tablelist(A) ::= tablelist(Y) COMMA ids(X) cpxName(Z) ids(F). {
toTSDBType(X.type); toTSDBType(X.type);
toTSDBType(F.type); toTSDBType(F.type);
X.n += Z.n; X.n += Z.n;
A = tVariantListAppendToken(Y, &X, -1); A = tVariantListAppendToken(Y, &X, -1);
A = tVariantListAppendToken(A, &F, -1); A = tVariantListAppendToken(A, &F, -1);
} }
// The value of interval should be the form of "number+[a,s,m,h,d,n,y]" or "now" // The value of interval should be the form of "number+[a,s,m,h,d,n,y]" or "now"
@ -526,9 +550,9 @@ item(A) ::= ids(X) cpxName(Y). {
} }
%type sortorder {int} %type sortorder {int}
sortorder(A) ::= ASC. {A = TSDB_ORDER_ASC; } sortorder(A) ::= ASC. { A = TSDB_ORDER_ASC; }
sortorder(A) ::= DESC. {A = TSDB_ORDER_DESC;} sortorder(A) ::= DESC. { A = TSDB_ORDER_DESC;}
sortorder(A) ::= . {A = TSDB_ORDER_ASC;} //default is descend order sortorder(A) ::= . { A = TSDB_ORDER_ASC; } // Ascending order by default
//group by clause //group by clause
%type groupby_opt {SArray*} %type groupby_opt {SArray*}
@ -536,8 +560,8 @@ sortorder(A) ::= . {A = TSDB_ORDER_ASC;} //default is descend orde
%type grouplist {SArray*} %type grouplist {SArray*}
%destructor grouplist {taosArrayDestroy($$);} %destructor grouplist {taosArrayDestroy($$);}
groupby_opt(A) ::= . {A = 0;} groupby_opt(A) ::= . { A = 0;}
groupby_opt(A) ::= GROUP BY grouplist(X). {A = X;} groupby_opt(A) ::= GROUP BY grouplist(X). { A = X;}
grouplist(A) ::= grouplist(X) COMMA item(Y). { grouplist(A) ::= grouplist(X) COMMA item(Y). {
A = tVariantListAppend(X, &Y, -1); A = tVariantListAppend(X, &Y, -1);
@ -583,20 +607,20 @@ where_opt(A) ::= WHERE expr(X). {A = X;}
expr(A) ::= LP(X) expr(Y) RP(Z). {A = Y; A->token.z = X.z; A->token.n = (Z.z - X.z + 1);} expr(A) ::= LP(X) expr(Y) RP(Z). {A = Y; A->token.z = X.z; A->token.n = (Z.z - X.z + 1);}
expr(A) ::= ID(X). {A = tSQLExprIdValueCreate(&X, TK_ID);} expr(A) ::= ID(X). { A = tSQLExprIdValueCreate(&X, TK_ID);}
expr(A) ::= ID(X) DOT ID(Y). {X.n += (1+Y.n); A = tSQLExprIdValueCreate(&X, TK_ID);} expr(A) ::= ID(X) DOT ID(Y). { X.n += (1+Y.n); A = tSQLExprIdValueCreate(&X, TK_ID);}
expr(A) ::= ID(X) DOT STAR(Y). {X.n += (1+Y.n); A = tSQLExprIdValueCreate(&X, TK_ALL);} expr(A) ::= ID(X) DOT STAR(Y). { X.n += (1+Y.n); A = tSQLExprIdValueCreate(&X, TK_ALL);}
expr(A) ::= INTEGER(X). {A = tSQLExprIdValueCreate(&X, TK_INTEGER);} expr(A) ::= INTEGER(X). { A = tSQLExprIdValueCreate(&X, TK_INTEGER);}
expr(A) ::= MINUS(X) INTEGER(Y). {X.n += Y.n; X.type = TK_INTEGER; A = tSQLExprIdValueCreate(&X, TK_INTEGER);} expr(A) ::= MINUS(X) INTEGER(Y). { X.n += Y.n; X.type = TK_INTEGER; A = tSQLExprIdValueCreate(&X, TK_INTEGER);}
expr(A) ::= PLUS(X) INTEGER(Y). {X.n += Y.n; X.type = TK_INTEGER; A = tSQLExprIdValueCreate(&X, TK_INTEGER);} expr(A) ::= PLUS(X) INTEGER(Y). { X.n += Y.n; X.type = TK_INTEGER; A = tSQLExprIdValueCreate(&X, TK_INTEGER);}
expr(A) ::= FLOAT(X). {A = tSQLExprIdValueCreate(&X, TK_FLOAT);} expr(A) ::= FLOAT(X). { A = tSQLExprIdValueCreate(&X, TK_FLOAT);}
expr(A) ::= MINUS(X) FLOAT(Y). {X.n += Y.n; X.type = TK_FLOAT; A = tSQLExprIdValueCreate(&X, TK_FLOAT);} expr(A) ::= MINUS(X) FLOAT(Y). { X.n += Y.n; X.type = TK_FLOAT; A = tSQLExprIdValueCreate(&X, TK_FLOAT);}
expr(A) ::= PLUS(X) FLOAT(Y). {X.n += Y.n; X.type = TK_FLOAT; A = tSQLExprIdValueCreate(&X, TK_FLOAT);} expr(A) ::= PLUS(X) FLOAT(Y). { X.n += Y.n; X.type = TK_FLOAT; A = tSQLExprIdValueCreate(&X, TK_FLOAT);}
expr(A) ::= STRING(X). {A = tSQLExprIdValueCreate(&X, TK_STRING);} expr(A) ::= STRING(X). { A = tSQLExprIdValueCreate(&X, TK_STRING);}
expr(A) ::= NOW(X). {A = tSQLExprIdValueCreate(&X, TK_NOW); } expr(A) ::= NOW(X). { A = tSQLExprIdValueCreate(&X, TK_NOW); }
expr(A) ::= VARIABLE(X). {A = tSQLExprIdValueCreate(&X, TK_VARIABLE);} expr(A) ::= VARIABLE(X). { A = tSQLExprIdValueCreate(&X, TK_VARIABLE);}
expr(A) ::= BOOL(X). {A = tSQLExprIdValueCreate(&X, TK_BOOL);} expr(A) ::= BOOL(X). { A = tSQLExprIdValueCreate(&X, TK_BOOL);}
// ordinary functions: min(x), max(x), top(k, 20) // ordinary functions: min(x), max(x), top(k, 20)
expr(A) ::= ID(X) LP exprlist(Y) RP(E). { A = tSQLExprCreateFunction(Y, &X, &E, X.type); } expr(A) ::= ID(X) LP exprlist(Y) RP(E). { A = tSQLExprCreateFunction(Y, &X, &E, X.type); }

75
src/query/src/qParserImpl.c
View File

@ -474,11 +474,18 @@ SQuerySQL *tSetQuerySQLElems(SStrToken *pSelectToken, tSQLExprList *pSelection,
return pQuery; return pQuery;
} }
void freeVariant(void *pItem) { static void freeVariant(void *pItem) {
tVariantListItem* p = (tVariantListItem*) pItem; tVariantListItem* p = (tVariantListItem*) pItem;
tVariantDestroy(&p->pVar); tVariantDestroy(&p->pVar);
} }
void freeCreateTableInfo(void* p) {
SCreatedTableInfo* pInfo = (SCreatedTableInfo*) p;
taosArrayDestroyEx(pInfo->pTagVals, freeVariant);
tfree(pInfo->fullname);
tfree(pInfo->tagdata.data);
}
void doDestroyQuerySql(SQuerySQL *pQuerySql) { void doDestroyQuerySql(SQuerySQL *pQuerySql) {
if (pQuerySql == NULL) { if (pQuerySql == NULL) {
return; return;
@ -519,31 +526,30 @@ void destroyAllSelectClause(SSubclauseInfo *pClause) {
tfree(pClause->pClause); tfree(pClause->pClause);
} }
SCreateTableSQL *tSetCreateSQLElems(SArray *pCols, SArray *pTags, SStrToken *pStableName, SCreateTableSQL *tSetCreateSQLElems(SArray *pCols, SArray *pTags, SQuerySQL *pSelect, int32_t type) {
SArray *pTagVals, SQuerySQL *pSelect, int32_t type) {
SCreateTableSQL *pCreate = calloc(1, sizeof(SCreateTableSQL)); SCreateTableSQL *pCreate = calloc(1, sizeof(SCreateTableSQL));
switch (type) { switch (type) {
case TSQL_CREATE_TABLE: { case TSQL_CREATE_TABLE: {
pCreate->colInfo.pColumns = pCols; pCreate->colInfo.pColumns = pCols;
assert(pTagVals == NULL && pTags == NULL); assert(pTags == NULL);
break; break;
} }
case TSQL_CREATE_STABLE: { case TSQL_CREATE_STABLE: {
pCreate->colInfo.pColumns = pCols; pCreate->colInfo.pColumns = pCols;
pCreate->colInfo.pTagColumns = pTags; pCreate->colInfo.pTagColumns = pTags;
assert(pTagVals == NULL && pTags != NULL && pCols != NULL); assert(pTags != NULL && pCols != NULL);
break;
}
case TSQL_CREATE_TABLE_FROM_STABLE: {
pCreate->usingInfo.pTagVals = pTagVals;
pCreate->usingInfo.stableName = *pStableName;
break; break;
} }
case TSQL_CREATE_STREAM: { case TSQL_CREATE_STREAM: {
pCreate->pSelect = pSelect; pCreate->pSelect = pSelect;
break; break;
} }
case TSQL_CREATE_TABLE_FROM_STABLE: {
assert(0);
}
default: default:
assert(false); assert(false);
} }
@ -552,10 +558,22 @@ SCreateTableSQL *tSetCreateSQLElems(SArray *pCols, SArray *pTags, SStrToken *pSt
return pCreate; return pCreate;
} }
SAlterTableSQL *tAlterTableSQLElems(SStrToken *pMeterName, SArray *pCols, SArray *pVals, int32_t type) { SCreatedTableInfo createNewChildTableInfo(SStrToken *pTableName, SArray *pTagVals, SStrToken *pToken, SStrToken* igExists) {
SCreatedTableInfo info;
memset(&info, 0, sizeof(SCreatedTableInfo));
info.name = *pToken;
info.pTagVals = pTagVals;
info.stableName = *pTableName;
info.igExist = (igExists->n > 0)? 1:0;
return info;
}
SAlterTableSQL *tAlterTableSQLElems(SStrToken *pTableName, SArray *pCols, SArray *pVals, int32_t type) {
SAlterTableSQL *pAlterTable = calloc(1, sizeof(SAlterTableSQL)); SAlterTableSQL *pAlterTable = calloc(1, sizeof(SAlterTableSQL));
pAlterTable->name = *pMeterName; pAlterTable->name = *pTableName;
pAlterTable->type = type; pAlterTable->type = type;
if (type == TSDB_ALTER_TABLE_ADD_COLUMN || type == TSDB_ALTER_TABLE_ADD_TAG_COLUMN) { if (type == TSDB_ALTER_TABLE_ADD_COLUMN || type == TSDB_ALTER_TABLE_ADD_TAG_COLUMN) {
@ -573,24 +591,29 @@ SAlterTableSQL *tAlterTableSQLElems(SStrToken *pMeterName, SArray *pCols, SArray
return pAlterTable; return pAlterTable;
} }
void* destroyCreateTableSQL(SCreateTableSQL* pCreate) {
doDestroyQuerySql(pCreate->pSelect);
taosArrayDestroy(pCreate->colInfo.pColumns);
taosArrayDestroy(pCreate->colInfo.pTagColumns);
taosArrayDestroyEx(pCreate->childTableInfo, freeCreateTableInfo);
tfree(pCreate);
return NULL;
}
void SQLInfoDestroy(SSqlInfo *pInfo) { void SQLInfoDestroy(SSqlInfo *pInfo) {
if (pInfo == NULL) return; if (pInfo == NULL) return;
if (pInfo->type == TSDB_SQL_SELECT) { if (pInfo->type == TSDB_SQL_SELECT) {
destroyAllSelectClause(&pInfo->subclauseInfo); destroyAllSelectClause(&pInfo->subclauseInfo);
} else if (pInfo->type == TSDB_SQL_CREATE_TABLE) { } else if (pInfo->type == TSDB_SQL_CREATE_TABLE) {
SCreateTableSQL *pCreateTableInfo = pInfo->pCreateTableInfo; pInfo->pCreateTableInfo = destroyCreateTableSQL(pInfo->pCreateTableInfo);
doDestroyQuerySql(pCreateTableInfo->pSelect);
taosArrayDestroy(pCreateTableInfo->colInfo.pColumns);
taosArrayDestroy(pCreateTableInfo->colInfo.pTagColumns);
taosArrayDestroyEx(pCreateTableInfo->usingInfo.pTagVals, freeVariant);
tfree(pInfo->pCreateTableInfo);
} else if (pInfo->type == TSDB_SQL_ALTER_TABLE) { } else if (pInfo->type == TSDB_SQL_ALTER_TABLE) {
taosArrayDestroyEx(pInfo->pAlterInfo->varList, freeVariant); taosArrayDestroyEx(pInfo->pAlterInfo->varList, freeVariant);
taosArrayDestroy(pInfo->pAlterInfo->pAddColumns); taosArrayDestroy(pInfo->pAlterInfo->pAddColumns);
tfree(pInfo->pAlterInfo->tagData.data);
tfree(pInfo->pAlterInfo); tfree(pInfo->pAlterInfo);
} else { } else {
if (pInfo->pDCLInfo != NULL && pInfo->pDCLInfo->nAlloc > 0) { if (pInfo->pDCLInfo != NULL && pInfo->pDCLInfo->nAlloc > 0) {
@ -624,7 +647,7 @@ SSubclauseInfo* setSubclause(SSubclauseInfo* pSubclause, void *pSqlExprInfo) {
return pSubclause; return pSubclause;
} }
SSqlInfo* setSQLInfo(SSqlInfo *pInfo, void *pSqlExprInfo, SStrToken *pMeterName, int32_t type) { SSqlInfo* setSQLInfo(SSqlInfo *pInfo, void *pSqlExprInfo, SStrToken *pTableName, int32_t type) {
pInfo->type = type; pInfo->type = type;
if (type == TSDB_SQL_SELECT) { if (type == TSDB_SQL_SELECT) {
@ -634,8 +657,8 @@ SSqlInfo* setSQLInfo(SSqlInfo *pInfo, void *pSqlExprInfo, SStrToken *pMeterName,
pInfo->pCreateTableInfo = pSqlExprInfo; pInfo->pCreateTableInfo = pSqlExprInfo;
} }
if (pMeterName != NULL) { if (pTableName != NULL) {
pInfo->pCreateTableInfo->name = *pMeterName; pInfo->pCreateTableInfo->name = *pTableName;
} }
return pInfo; return pInfo;
@ -653,8 +676,8 @@ SSubclauseInfo* appendSelectClause(SSubclauseInfo *pQueryInfo, void *pSubclause)
return pQueryInfo; return pQueryInfo;
} }
void setCreatedTableName(SSqlInfo *pInfo, SStrToken *pMeterName, SStrToken *pIfNotExists) { void setCreatedTableName(SSqlInfo *pInfo, SStrToken *pTableNameToken, SStrToken *pIfNotExists) {
pInfo->pCreateTableInfo->name = *pMeterName; pInfo->pCreateTableInfo->name = *pTableNameToken;
pInfo->pCreateTableInfo->existCheck = (pIfNotExists->n != 0); pInfo->pCreateTableInfo->existCheck = (pIfNotExists->n != 0);
} }

1781
src/query/src/sql.c
View File

File diff suppressed because it is too large Load Diff

2
src/util/inc/tstoken.h
View File

@ -31,7 +31,7 @@ extern "C" {
typedef struct SStrToken { typedef struct SStrToken {
uint32_t n; uint32_t n;
uint32_t type; uint32_t type;
char * z; char *z;
} SStrToken; } SStrToken;
/** /**