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

feat:[TD-30975]process return value in schemaless

This commit is contained in:
wangmm0220 2024-07-15 19:07:20 +08:00
parent 5957b2c19a
commit aed8a75e35
12 changed files with 744 additions and 376 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/tname.h
View File

@ -81,7 +81,7 @@ typedef struct {
// uint64_t uid; // child table uid, may be useful
} RandTableName;
void buildChildTableName(RandTableName* rName);
int32_t buildChildTableName(RandTableName* rName);
#ifdef __cplusplus
}

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

@ -149,10 +149,10 @@ int32_t qCreateSName(SName* pName, const char* pTableName, int32_t acctId, char*
void qDestroyBoundColInfo(void* pInfo);
SQuery* smlInitHandle();
int32_t smlInitHandle(SQuery** query);
int32_t smlBuildRow(STableDataCxt* pTableCxt);
int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* kv, int32_t index);
STableDataCxt* smlInitTableDataCtx(SQuery* query, STableMeta* pTableMeta);
int32_t smlInitTableDataCtx(SQuery* query, STableMeta* pTableMeta, STableDataCxt** cxt);
void clearColValArraySml(SArray* pCols);
int32_t smlBindData(SQuery* handle, bool dataFormat, SArray* tags, SArray* colsSchema, SArray* cols,

24
source/client/inc/clientSml.h
View File

@ -82,7 +82,7 @@ extern "C" {
#define IS_SLASH_LETTER_IN_MEASUREMENT(sql) \
(*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE || *(sql) == SLASH))
#define MOVE_FORWARD_ONE(sql, len) (memmove((void *)((sql)-1), (sql), len))
#define MOVE_FORWARD_ONE(sql, len) ((void)memmove((void *)((sql)-1), (sql), len))
#define PROCESS_SLASH_IN_MEASUREMENT(key, keyLen) \
for (int i = 1; i < keyLen; ++i) { \
@ -211,19 +211,19 @@ extern int64_t smlFactorS[];
typedef int32_t (*_equal_fn_sml)(const void *, const void *);
SSmlHandle *smlBuildSmlInfo(TAOS *taos);
int32_t smlBuildSmlInfo(TAOS *taos, SSmlHandle **handle);
void smlDestroyInfo(SSmlHandle *info);
int smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset);
int smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset);
bool smlParseNumberOld(SSmlKv *kvVal, SSmlMsgBuf *msg);
int32_t smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2);
bool smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg);
void smlBuildInvalidDataMsg(SSmlMsgBuf *pBuf, const char *msg1, const char *msg2);
int32_t smlParseNumber(SSmlKv *kvVal, SSmlMsgBuf *msg);
int64_t smlGetTimeValue(const char *value, int32_t len, uint8_t fromPrecision, uint8_t toPrecision);
SSmlTableInfo* smlBuildTableInfo(int numRows, const char* measure, int32_t measureLen);
SSmlSTableMeta* smlBuildSTableMeta(bool isDataFormat);
int32_t smlBuildTableInfo(int numRows, const char* measure, int32_t measureLen, SSmlTableInfo** tInfo);
int32_t smlBuildSTableMeta(bool isDataFormat, SSmlSTableMeta** sMeta);
int32_t smlSetCTableName(SSmlTableInfo *oneTable);
void getTableUid(SSmlHandle *info, SSmlLineInfo *currElement, SSmlTableInfo *tinfo);
STableMeta* smlGetMeta(SSmlHandle *info, const void* measure, int32_t measureLen);
int32_t getTableUid(SSmlHandle *info, SSmlLineInfo *currElement, SSmlTableInfo *tinfo);
int32_t smlGetMeta(SSmlHandle *info, const void* measure, int32_t measureLen, STableMeta **pTableMeta);
int32_t is_same_child_table_telnet(const void *a, const void *b);
int64_t smlParseOpenTsdbTime(SSmlHandle *info, const char *data, int32_t len);
int32_t smlClearForRerun(SSmlHandle *info);
@ -236,7 +236,7 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
int32_t smlParseJSON(SSmlHandle *info, char *payload);
SSmlSTableMeta* smlBuildSuperTableInfo(SSmlHandle *info, SSmlLineInfo *currElement);
int32_t smlBuildSuperTableInfo(SSmlHandle *info, SSmlLineInfo *currElement, SSmlSTableMeta** sMeta);
bool isSmlTagAligned(SSmlHandle *info, int cnt, SSmlKv *kv);
bool isSmlColAligned(SSmlHandle *info, int cnt, SSmlKv *kv);
int32_t smlProcessChildTable(SSmlHandle *info, SSmlLineInfo *elements);
@ -246,9 +246,9 @@ void smlBuildTsKv(SSmlKv *kv, int64_t ts);
int32_t smlParseEndTelnetJson(SSmlHandle *info, SSmlLineInfo *elements, SSmlKv *kvTs, SSmlKv *kv);
int32_t smlParseEndLine(SSmlHandle *info, SSmlLineInfo *elements, SSmlKv *kvTs);
static inline bool smlDoubleToInt64OverFlow(double num) {
if (num >= (double)INT64_MAX || num <= (double)INT64_MIN) return true;
return false;
static inline int32_t smlDoubleToInt64OverFlow(double num) {
if (num >= (double)INT64_MAX || num <= (double)INT64_MIN) return 0;
return -1;
}
static inline void smlStrReplace(char* src, int32_t len){

23
source/client/src/clientRawBlockWrite.c
View File

@ -1572,9 +1572,8 @@ int taos_write_raw_block_with_fields_with_reqid(TAOS* taos, int rows, char* pDat
}
// uError("td23101 0vgId:%d, vgId:%d, name:%s, uid:%"PRIu64, vgData.vgId, pTableMeta->vgId, tbname, pTableMeta->uid);
pQuery = smlInitHandle();
if (pQuery == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
code = smlInitHandle(&pQuery);
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
@ -1656,9 +1655,9 @@ int taos_write_raw_block_with_reqid(TAOS* taos, int rows, char* pData, const cha
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
pQuery = smlInitHandle();
if (pQuery == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
code = smlInitHandle(&pQuery);
if (code != TSDB_CODE_SUCCESS) {
goto end;
}
pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
@ -1755,12 +1754,12 @@ static int32_t tmqWriteRawDataImpl(TAOS* taos, void* data, int32_t dataLen) {
conn.requestObjRefId = pRequest->self;
conn.mgmtEps = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);
pQuery = smlInitHandle();
if (pQuery == NULL) {
code = TSDB_CODE_OUT_OF_MEMORY;
code = smlInitHandle(&pQuery);
if (code != TSDB_CODE_SUCCESS) {
SET_ERROR_MSG("init sml handle failed");
goto end;
}
pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
while (++rspObj.common.resIter < rspObj.rsp.common.blockNum) {
void* pRetrieve = taosArrayGetP(rspObj.rsp.common.blockData, rspObj.common.resIter);
@ -1903,12 +1902,12 @@ static int32_t tmqWriteRawMetaDataImpl(TAOS* taos, void* data, int32_t dataLen)
conn.requestObjRefId = pRequest->self;
conn.mgmtEps = getEpSet_s(&pRequest->pTscObj->pAppInfo->mgmtEp);
pQuery = smlInitHandle();
if (pQuery == NULL) {
code = smlInitHandle(&pQuery);
if (code != TSDB_CODE_SUCCESS) {
SET_ERROR_MSG("init sml handle failed");
code = TSDB_CODE_OUT_OF_MEMORY;
goto end;
}
pVgHash = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
uDebug(LOG_ID_TAG " write raw metadata block num:%d", LOG_ID_VALUE, rspObj.rsp.common.blockNum);

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

File diff suppressed because it is too large Load Diff

105
source/client/src/clientSmlJson.c
View File

@ -29,29 +29,29 @@
(start)++; \
}
static char *smlJsonGetObj(char *payload) {
static int32_t smlJsonGetObj(char **payload) {
int leftBracketCnt = 0;
bool isInQuote = false;
while (*payload) {
if (*payload == '"' && *(payload - 1) != '\\') {
while (**payload) {
if (**payload == '"' && *((*payload) - 1) != '\\') {
isInQuote = !isInQuote;
} else if (!isInQuote && unlikely(*payload == '{')) {
} else if (!isInQuote && unlikely(**payload == '{')) {
leftBracketCnt++;
payload++;
(*payload)++;
continue;
} else if (!isInQuote && unlikely(*payload == '}')) {
} else if (!isInQuote && unlikely(**payload == '}')) {
leftBracketCnt--;
payload++;
(*payload)++;
if (leftBracketCnt == 0) {
return payload;
return 0;
} else if (leftBracketCnt < 0) {
return NULL;
return -1;
}
continue;
}
payload++;
(*payload)++;
}
return NULL;
return -1;
}
int smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset) {
@ -99,8 +99,9 @@ int smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset) {
offset[index++] = *start - sTmp;
element->timestamp = (*start);
if (*(*start) == '{') {
char *tmp = smlJsonGetObj((*start));
if (tmp) {
char *tmp = *start;
int32_t code = smlJsonGetObj(&tmp);
if (code == 0) {
element->timestampLen = tmp - (*start);
*start = tmp;
}
@ -127,8 +128,9 @@ int smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset) {
offset[index++] = *start - sTmp;
element->cols = (*start);
if (*(*start) == '{') {
char *tmp = smlJsonGetObj((*start));
if (tmp) {
char *tmp = *start;
int32_t code = smlJsonGetObj(&tmp);
if (code == 0) {
element->colsLen = tmp - (*start);
*start = tmp;
}
@ -153,8 +155,9 @@ int smlJsonParseObjFirst(char **start, SSmlLineInfo *element, int8_t *offset) {
JUMP_JSON_SPACE((*start))
offset[index++] = *start - sTmp;
element->tags = (*start);
char *tmp = smlJsonGetObj((*start));
if (tmp) {
char *tmp = *start;
int32_t code = smlJsonGetObj(&tmp);
if (code == 0) {
element->tagsLen = tmp - (*start);
*start = tmp;
}
@ -209,8 +212,9 @@ int smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset) {
(*start) += offset[index++];
element->timestamp = *start;
if (*(*start) == '{') {
char *tmp = smlJsonGetObj((*start));
if (tmp) {
char *tmp = *start;
int32_t code = smlJsonGetObj(&tmp);
if (code == 0) {
element->timestampLen = tmp - (*start);
*start = tmp;
}
@ -227,8 +231,9 @@ int smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset) {
(*start) += offset[index++];
element->cols = *start;
if (*(*start) == '{') {
char *tmp = smlJsonGetObj((*start));
if (tmp) {
char *tmp = *start;
int32_t code = smlJsonGetObj(&tmp);
if (code == 0) {
element->colsLen = tmp - (*start);
*start = tmp;
}
@ -244,8 +249,9 @@ int smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset) {
} else if ((*start)[1] == 't' && (*start)[2] == 'a') {
(*start) += offset[index++];
element->tags = (*start);
char *tmp = smlJsonGetObj((*start));
if (tmp) {
char *tmp = *start;
int32_t code = smlJsonGetObj(&tmp);
if (code == 0) {
element->tagsLen = tmp - (*start);
*start = tmp;
}
@ -261,7 +267,7 @@ int smlJsonParseObj(char **start, SSmlLineInfo *element, int8_t *offset) {
uError("elements != %d", OTD_JSON_FIELDS_NUM);
return TSDB_CODE_TSC_INVALID_JSON;
}
return 0;
return TSDB_CODE_SUCCESS;
}
static inline int32_t smlParseMetricFromJSON(SSmlHandle *info, cJSON *metric, SSmlLineInfo *elements) {
@ -492,20 +498,17 @@ static int32_t smlProcessTagJson(SSmlHandle *info, cJSON *tags){
int32_t tagNum = cJSON_GetArraySize(tags);
if (unlikely(tagNum == 0)) {
uError("SML:Tag should not be empty");
terrno = TSDB_CODE_TSC_INVALID_JSON;
return -1;
return TSDB_CODE_TSC_INVALID_JSON;
}
for (int32_t i = 0; i < tagNum; ++i) {
cJSON *tag = cJSON_GetArrayItem(tags, i);
if (unlikely(tag == NULL)) {
terrno = TSDB_CODE_TSC_INVALID_JSON;
return -1;
return TSDB_CODE_TSC_INVALID_JSON;
}
size_t keyLen = strlen(tag->string);
if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
uError("OTD:Tag key length is 0 or too large than 64");
terrno = TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
return -1;
return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
}
// add kv to SSmlKv
@ -516,19 +519,17 @@ static int32_t smlProcessTagJson(SSmlHandle *info, cJSON *tags){
// value
int32_t ret = smlParseValueFromJSON(tag, &kv);
if (unlikely(ret != TSDB_CODE_SUCCESS)) {
terrno = ret;
return -1;
return ret;
}
taosArrayPush(preLineKV, &kv);
if (info->dataFormat && !isSmlTagAligned(info, cnt, &kv)) {
terrno = TSDB_CODE_SUCCESS;
return -1;
return TSDB_CODE_TSC_INVALID_JSON;
}
cnt++;
}
return 0;
return TSDB_CODE_SUCCESS;
}
static int32_t smlParseTagsFromJSON(SSmlHandle *info, cJSON *tags, SSmlLineInfo *elements) {
@ -536,12 +537,18 @@ static int32_t smlParseTagsFromJSON(SSmlHandle *info, cJSON *tags, SSmlLineInfo
if(info->dataFormat){
ret = smlProcessSuperTable(info, elements);
if(ret != 0){
return terrno;
if(info->reRun){
return TSDB_CODE_SUCCESS;
}
return ret;
}
}
ret = smlProcessTagJson(info, tags);
if(ret != 0){
return terrno;
if(info->reRun){
return TSDB_CODE_SUCCESS;
}
return ret;
}
ret = smlJoinMeasureTag(elements);
if(ret != 0){
@ -703,6 +710,9 @@ static int32_t smlParseJSONStringExt(SSmlHandle *info, cJSON *root, SSmlLineInfo
// Parse tags
bool needFree = info->dataFormat;
elements->tags = cJSON_PrintUnformatted(tagsJson);
if (elements->tags == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
elements->tagsLen = strlen(elements->tags);
if (is_same_child_table_telnet(elements, &info->preLine) != 0) {
ret = smlParseTagsFromJSON(info, tagsJson, elements);
@ -823,7 +833,7 @@ static int32_t smlParseJSONString(SSmlHandle *info, char **start, SSmlLineInfo *
}
if (ret != TSDB_CODE_SUCCESS) {
return TSDB_CODE_TSC_INVALID_VALUE;
return ret;
}
if (unlikely(**start == '\0' && elements->measure == NULL)) return TSDB_CODE_SUCCESS;
@ -844,18 +854,23 @@ static int32_t smlParseJSONString(SSmlHandle *info, char **start, SSmlLineInfo *
cJSON *valueJson = cJSON_Parse(elements->cols);
if (unlikely(valueJson == NULL)) {
uError("SML:0x%" PRIx64 " parse json cols failed:%s", info->id, elements->cols);
elements->cols[elements->colsLen] = tmp;
return TSDB_CODE_TSC_INVALID_JSON;
}
taosArrayPush(info->tagJsonArray, &valueJson);
if (taosArrayPush(info->tagJsonArray, &valueJson) == NULL){
cJSON_Delete(valueJson);
elements->cols[elements->colsLen] = tmp;
return TSDB_CODE_OUT_OF_MEMORY;
}
ret = smlParseValueFromJSONObj(valueJson, &kv);
if (ret != TSDB_CODE_SUCCESS) {
uError("SML:Failed to parse value from JSON Obj:%s", elements->cols);
uError("SML:0x%" PRIx64 " Failed to parse value from JSON Obj:%s", info->id, elements->cols);
elements->cols[elements->colsLen] = tmp;
return TSDB_CODE_TSC_INVALID_VALUE;
}
elements->cols[elements->colsLen] = tmp;
} else if (smlParseValue(&kv, &info->msgBuf) != TSDB_CODE_SUCCESS) {
uError("SML:cols invalidate:%s", elements->cols);
uError("SML:0x%" PRIx64 " cols invalidate:%s", info->id, elements->cols);
return TSDB_CODE_TSC_INVALID_VALUE;
}
@ -870,7 +885,11 @@ static int32_t smlParseJSONString(SSmlHandle *info, char **start, SSmlLineInfo *
return TSDB_CODE_TSC_INVALID_JSON;
}
taosArrayPush(info->tagJsonArray, &tagsJson);
if (taosArrayPush(info->tagJsonArray, &tagsJson) == NULL){
cJSON_Delete(tagsJson);
uError("SML:0x%" PRIx64 " taosArrayPush failed", info->id);
return TSDB_CODE_OUT_OF_MEMORY;
}
ret = smlParseTagsFromJSON(info, tagsJson, elements);
if (unlikely(ret)) {
uError("OTD:0x%" PRIx64 " Unable to parse tags from JSON payload", info->id);
@ -937,7 +956,7 @@ int32_t smlParseJSON(SSmlHandle *info, char *payload) {
return ret;
}
info->lines = (SSmlLineInfo *)tmp;
memset(info->lines + cnt, 0, (payloadNum - cnt) * sizeof(SSmlLineInfo));
(void)memset(info->lines + cnt, 0, (payloadNum - cnt) * sizeof(SSmlLineInfo));
}
ret = smlParseJSONString(info, &dataPointStart, info->lines + cnt);
if ((info->lines + cnt)->measure == NULL) break;

66
source/client/src/clientSmlLine.c
View File

@ -103,7 +103,10 @@ int32_t smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) {
return code;
}
char* tmp = taosMemoryCalloc(pVal->length, 1);
memcpy(tmp, pVal->value + NCHAR_ADD_LEN - 1, pVal->length - NCHAR_ADD_LEN);
if (tmp == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
(void)memcpy(tmp, pVal->value + NCHAR_ADD_LEN - 1, pVal->length - NCHAR_ADD_LEN);
code = doGeomFromText(tmp, (unsigned char **)&pVal->value, &pVal->length);
taosMemoryFree(tmp);
if (code != TSDB_CODE_SUCCESS) {
@ -212,8 +215,7 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
while (*sql < sqlEnd) {
if (unlikely(IS_SPACE(*sql,escapeChar) || IS_COMMA(*sql,escapeChar))) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
terrno = TSDB_CODE_SML_INVALID_DATA;
return -1;
return TSDB_CODE_SML_INVALID_DATA;
}
if (unlikely(IS_EQUAL(*sql,escapeChar))) {
keyLen = *sql - key;
@ -230,8 +232,7 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
if (unlikely(IS_INVALID_COL_LEN(keyLen - keyLenEscaped))) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid key or key is too long than 64", key);
terrno = TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
return -1;
return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
}
// parse value
@ -245,8 +246,7 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
break;
} else if (unlikely(IS_EQUAL(*sql,escapeChar))) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
terrno = TSDB_CODE_SML_INVALID_DATA;
return -1;
return TSDB_CODE_SML_INVALID_DATA;
}
if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
@ -261,24 +261,28 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
if (unlikely(valueLen == 0)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", value);
terrno = TSDB_CODE_SML_INVALID_DATA;
return -1;
return TSDB_CODE_SML_INVALID_DATA;
}
if (unlikely(valueLen - valueLenEscaped > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
terrno = TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
return -1;
return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
}
if (keyEscaped) {
char *tmp = (char *)taosMemoryMalloc(keyLen);
memcpy(tmp, key, keyLen);
if (tmp == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
(void)memcpy(tmp, key, keyLen);
PROCESS_SLASH_IN_TAG_FIELD_KEY(tmp, keyLen);
key = tmp;
}
if (valueEscaped) {
char *tmp = (char *)taosMemoryMalloc(valueLen);
memcpy(tmp, value, valueLen);
if (tmp == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
(void)memcpy(tmp, value, valueLen);
PROCESS_SLASH_IN_TAG_FIELD_KEY(tmp, valueLen);
value = tmp;
}
@ -289,11 +293,12 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
.length = valueLen,
.keyEscaped = keyEscaped,
.valueEscaped = valueEscaped};
taosArrayPush(preLineKV, &kv);
if(taosArrayPush(preLineKV, &kv) == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
if (info->dataFormat && !isSmlTagAligned(info, cnt, &kv)) {
terrno = TSDB_CODE_SUCCESS;
return -1;
return TSDB_CODE_TSC_INVALID_JSON;
}
cnt++;
@ -302,7 +307,7 @@ static int32_t smlProcessTagLine(SSmlHandle *info, char **sql, char *sqlEnd){
}
(*sql)++;
}
return 0;
return TSDB_CODE_SUCCESS;
}
static int32_t smlParseTagLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLineInfo *elements) {
@ -315,13 +320,19 @@ static int32_t smlParseTagLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlL
if(info->dataFormat){
ret = smlProcessSuperTable(info, elements);
if(ret != 0){
return terrno;
if(info->reRun){
return TSDB_CODE_SUCCESS;
}
return ret;
}
}
ret = smlProcessTagLine(info, sql, sqlEnd);
if(ret != 0){
return terrno;
if (info->reRun){
return TSDB_CODE_SUCCESS;
}
return ret;
}
return smlProcessChildTable(info, elements);
@ -410,7 +421,10 @@ static int32_t smlParseColLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlL
if (keyEscaped) {
char *tmp = (char *)taosMemoryMalloc(kv.keyLen);
memcpy(tmp, key, kv.keyLen);
if (tmp == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
(void)memcpy(tmp, key, kv.keyLen);
PROCESS_SLASH_IN_TAG_FIELD_KEY(tmp, kv.keyLen);
kv.key = tmp;
kv.keyEscaped = keyEscaped;
@ -418,7 +432,10 @@ static int32_t smlParseColLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlL
if (valueEscaped) {
char *tmp = (char *)taosMemoryMalloc(kv.length);
memcpy(tmp, kv.value, kv.length);
if (tmp == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
(void)memcpy(tmp, kv.value, kv.length);
PROCESS_SLASH_IN_FIELD_VALUE(tmp, kv.length);
ASSERT(kv.type != TSDB_DATA_TYPE_GEOMETRY);
if(kv.type == TSDB_DATA_TYPE_VARBINARY){
@ -437,8 +454,13 @@ static int32_t smlParseColLine(SSmlHandle *info, char **sql, char *sqlEnd, SSmlL
} else {
if (currElement->colArray == NULL) {
currElement->colArray = taosArrayInit_s(sizeof(SSmlKv), 1);
if (currElement->colArray == NULL) {
return TSDB_CODE_OUT_OF_MEMORY;
}
}
if (taosArrayPush(currElement->colArray, &kv) == NULL){ // reserve for timestamp
return TSDB_CODE_OUT_OF_MEMORY;
}
taosArrayPush(currElement->colArray, &kv); // reserve for timestamp
}
cnt++;

30
source/client/src/clientSmlTelnet.c
View File

@ -84,8 +84,7 @@ static int32_t smlProcessTagTelnet(SSmlHandle *info, char *data, char *sqlEnd){
while (sql < sqlEnd) {
if (unlikely(*sql == SPACE)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
terrno = TSDB_CODE_SML_INVALID_DATA;
return -1;
return TSDB_CODE_SML_INVALID_DATA;
}
if (unlikely(*sql == EQUAL)) {
keyLen = sql - key;
@ -97,8 +96,7 @@ static int32_t smlProcessTagTelnet(SSmlHandle *info, char *data, char *sqlEnd){
if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid key or key is too long than 64", key);
terrno = TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
return -1;
return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
}
// parse value
@ -111,8 +109,7 @@ static int32_t smlProcessTagTelnet(SSmlHandle *info, char *data, char *sqlEnd){
}
if (unlikely(*sql == EQUAL)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql);
terrno = TSDB_CODE_SML_INVALID_DATA;
return -1;
return TSDB_CODE_SML_INVALID_DATA;
}
sql++;
}
@ -120,13 +117,11 @@ static int32_t smlProcessTagTelnet(SSmlHandle *info, char *data, char *sqlEnd){
if (unlikely(valueLen == 0)) {
smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", value);
terrno = TSDB_CODE_TSC_INVALID_VALUE;
return -1;
return TSDB_CODE_TSC_INVALID_VALUE;
}
if (unlikely(valueLen > (TSDB_MAX_TAGS_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
terrno = TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
return -1;
return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
}
SSmlKv kv = {.key = key,
@ -138,12 +133,11 @@ static int32_t smlProcessTagTelnet(SSmlHandle *info, char *data, char *sqlEnd){
.valueEscaped = false};
taosArrayPush(preLineKV, &kv);
if (info->dataFormat && !isSmlTagAligned(info, cnt, &kv)) {
terrno = TSDB_CODE_SUCCESS;
return -1;
return TSDB_CODE_SML_INVALID_DATA;
}
cnt++;
}
return 0;
return TSDB_CODE_SUCCESS;
}
static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SSmlLineInfo *elements) {
@ -156,13 +150,19 @@ static int32_t smlParseTelnetTags(SSmlHandle *info, char *data, char *sqlEnd, SS
if(info->dataFormat){
ret = smlProcessSuperTable(info, elements);
if(ret != 0){
return terrno;
if(info->reRun){
return TSDB_CODE_SUCCESS;
}
return ret;
}
}
ret = smlProcessTagTelnet(info, data, sqlEnd);
if(ret != 0){
return terrno;
if (info->reRun){
return TSDB_CODE_SUCCESS;
}
return ret;
}
ret = smlJoinMeasureTag(elements);

114
source/client/test/smlTest.cpp
View File

@ -40,13 +40,16 @@ TEST(testCase, smlParseInfluxString_Test) {
msgBuf.len = 256;
SSmlLineInfo elements = {0};
SSmlHandle *info = smlBuildSmlInfo(NULL);
SSmlHandle *info = nullptr;
int32_t code = smlBuildSmlInfo(nullptr, &info);
ASSERT_EQ(code, 0);
info->protocol = TSDB_SML_LINE_PROTOCOL;
info->dataFormat = false;
// case 1
char *tmp = "\\,st,t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000 ,32,c=3";
char *sql = (char *)taosMemoryCalloc(256, 1);
memcpy(sql, tmp, strlen(tmp) + 1);
ASSERT_NE(sql, nullptr);
(void)memcpy(sql, tmp, strlen(tmp) + 1);
int ret = smlParseInfluxString(info, sql, sql + strlen(sql), &elements);
ASSERT_EQ(ret, 0);
ASSERT_EQ(elements.measure, sql);
@ -62,22 +65,22 @@ TEST(testCase, smlParseInfluxString_Test) {
ASSERT_EQ(elements.timestamp, sql + elements.measureTagsLen + 1 + elements.colsLen + 1);
ASSERT_EQ(elements.timestampLen, strlen("1626006833639000000"));
taosArrayDestroy(elements.colArray);
elements.colArray = NULL;
(void)taosArrayDestroy(elements.colArray);
elements.colArray = nullptr;
// case 2 false
tmp = "st,t1=3,t2=4,t3=t3 c1=3i64,c3=\"passit hello,c1=2,c2=false,c4=4f64 1626006833639000000";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
(void)memcpy(sql, tmp, strlen(tmp) + 1);
(void)memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(info, sql, sql + strlen(sql), &elements);
ASSERT_NE(ret, 0);
taosArrayDestroy(elements.colArray);
elements.colArray = NULL;
(void)taosArrayDestroy(elements.colArray);
elements.colArray = nullptr;
// case 4 tag is null
tmp = "st, c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
(void)memcpy(sql, tmp, strlen(tmp) + 1);
(void)memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(info, sql, sql + strlen(sql), &elements);
ASSERT_EQ(ret, 0);
ASSERT_EQ(elements.measure, sql);
@ -92,13 +95,13 @@ TEST(testCase, smlParseInfluxString_Test) {
ASSERT_EQ(elements.timestamp, sql + elements.measureTagsLen + 1 + elements.colsLen + 1);
ASSERT_EQ(elements.timestampLen, strlen("1626006833639000000"));
taosArrayDestroy(elements.colArray);
elements.colArray = NULL;
(void)taosArrayDestroy(elements.colArray);
elements.colArray = nullptr;
// case 5 tag is null
tmp = " st c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 1626006833639000000 ";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
(void)memcpy(sql, tmp, strlen(tmp) + 1);
(void)memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(info, sql, sql + strlen(sql), &elements);
ASSERT_EQ(ret, 0);
ASSERT_EQ(elements.measure, sql + 1);
@ -112,36 +115,36 @@ TEST(testCase, smlParseInfluxString_Test) {
ASSERT_EQ(elements.timestamp, sql + 1 + elements.measureTagsLen + 3 + elements.colsLen + 2);
ASSERT_EQ(elements.timestampLen, strlen("1626006833639000000"));
taosArrayDestroy(elements.colArray);
elements.colArray = NULL;
(void)taosArrayDestroy(elements.colArray);
elements.colArray = nullptr;
// case 6
tmp = " st c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 ";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
(void)memcpy(sql, tmp, strlen(tmp) + 1);
(void)memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(info, sql, sql + strlen(sql), &elements);
ASSERT_EQ(ret, 0);
taosArrayDestroy(elements.colArray);
elements.colArray = NULL;
smlClearForRerun(info);
(void)taosArrayDestroy(elements.colArray);
elements.colArray = nullptr;
ASSERT_EQ(smlClearForRerun(info), 0);
// case 7
tmp = " st , ";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
(void)memcpy(sql, tmp, strlen(tmp) + 1);
(void)memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(info, sql, sql + strlen(sql), &elements);
ASSERT_NE(ret, 0);
taosArrayDestroy(elements.colArray);
elements.colArray = NULL;
(void)taosArrayDestroy(elements.colArray);
elements.colArray = nullptr;
// case 8 false
tmp = ", st , ";
memcpy(sql, tmp, strlen(tmp) + 1);
memset(&elements, 0, sizeof(SSmlLineInfo));
(void)memcpy(sql, tmp, strlen(tmp) + 1);
(void)memset(&elements, 0, sizeof(SSmlLineInfo));
ret = smlParseInfluxString(info, sql, sql + strlen(sql), &elements);
ASSERT_NE(ret, 0);
taosArrayDestroy(elements.colArray);
elements.colArray = NULL;
(void)taosArrayDestroy(elements.colArray);
elements.colArray = nullptr;
taosMemoryFree(sql);
smlDestroyInfo(info);
@ -228,7 +231,9 @@ TEST(testCase, smlParseCols_Error_Test) {
"escape_test,tag1=\"tag1_value\",tag2=\"tag2_value\" col0=\"co\"l\"0_value\",col1=\"col1_value\" 1680918783010000000"
};
SSmlHandle *info = smlBuildSmlInfo(NULL);
SSmlHandle *info = nullptr;
int32_t code = smlBuildSmlInfo(nullptr, &info);
ASSERT_EQ(code, TSDB_CODE_SUCCESS);
info->protocol = TSDB_SML_LINE_PROTOCOL;
info->dataFormat = false;
for (int i = 0; i < sizeof(data) / sizeof(data[0]); i++) {
@ -238,13 +243,14 @@ TEST(testCase, smlParseCols_Error_Test) {
msgBuf.len = 256;
int32_t len = strlen(data[i]);
char *sql = (char *)taosMemoryCalloc(256, 1);
memcpy(sql, data[i], len + 1);
ASSERT_NE(sql, nullptr);
(void)memcpy(sql, data[i], len + 1);
SSmlLineInfo elements = {0};
int32_t ret = smlParseInfluxString(info, sql, sql + len, &elements);
// printf("i:%d\n", i);
ASSERT_NE(ret, TSDB_CODE_SUCCESS);
taosMemoryFree(sql);
taosArrayDestroy(elements.colArray);
(void)taosArrayDestroy(elements.colArray);
}
smlDestroyInfo(info);
}
@ -254,7 +260,9 @@ TEST(testCase, smlParseCols_Test) {
SSmlMsgBuf msgBuf;
msgBuf.buf = msg;
msgBuf.len = 256;
SSmlHandle *info = smlBuildSmlInfo(NULL);
SSmlHandle *info = nullptr;
int32_t code = smlBuildSmlInfo(nullptr, &info);
ASSERT_EQ(code, TSDB_CODE_SUCCESS);
info->protocol = TSDB_SML_LINE_PROTOCOL;
info->dataFormat = false;
SSmlLineInfo elements = {0};
@ -267,7 +275,8 @@ TEST(testCase, smlParseCols_Test) {
"cboolf=f,cnch_=l\"iuwq\" 1626006833639000000";
int32_t len = strlen(data);
char *sql = (char *)taosMemoryCalloc(1024, 1);
memcpy(sql, data, len + 1);
ASSERT_NE(sql, nullptr);
(void)memcpy(sql, data, len + 1);
int32_t ret = smlParseInfluxString(info, sql, sql + len, &elements);
ASSERT_EQ(ret, TSDB_CODE_SUCCESS);
int32_t size = taosArrayGetSize(elements.colArray);
@ -275,6 +284,7 @@ TEST(testCase, smlParseCols_Test) {
// binary
SSmlKv *kv = (SSmlKv *)taosArrayGet(elements.colArray, 1);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cb=in", 5), 0);
ASSERT_EQ(kv->keyLen, 5);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BINARY);
@ -285,6 +295,7 @@ TEST(testCase, smlParseCols_Test) {
// nchar
kv = (SSmlKv *)taosArrayGet(elements.colArray, 2);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cnch", 4), 0);
ASSERT_EQ(kv->keyLen, 4);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
@ -293,6 +304,7 @@ TEST(testCase, smlParseCols_Test) {
// bool
kv = (SSmlKv *)taosArrayGet(elements.colArray, 3);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cbool", 5), 0);
ASSERT_EQ(kv->keyLen, 5);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BOOL);
@ -301,6 +313,7 @@ TEST(testCase, smlParseCols_Test) {
// double
kv = (SSmlKv *)taosArrayGet(elements.colArray, 4);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cf64", 4), 0);
ASSERT_EQ(kv->keyLen, 4);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_DOUBLE);
@ -310,6 +323,7 @@ TEST(testCase, smlParseCols_Test) {
// float
kv = (SSmlKv *)taosArrayGet(elements.colArray, 5);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cf64_", 5), 0);
ASSERT_EQ(kv->keyLen, 5);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_DOUBLE);
@ -319,6 +333,7 @@ TEST(testCase, smlParseCols_Test) {
// float
kv = (SSmlKv *)taosArrayGet(elements.colArray, 6);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cf32", 4), 0);
ASSERT_EQ(kv->keyLen, 4);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_FLOAT);
@ -328,6 +343,7 @@ TEST(testCase, smlParseCols_Test) {
// tiny int
kv = (SSmlKv *)taosArrayGet(elements.colArray, 7);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "ci8", 3), 0);
ASSERT_EQ(kv->keyLen, 3);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_TINYINT);
@ -336,6 +352,7 @@ TEST(testCase, smlParseCols_Test) {
// unsigned tiny int
kv = (SSmlKv *)taosArrayGet(elements.colArray, 8);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cu8", 3), 0);
ASSERT_EQ(kv->keyLen, 3);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_UTINYINT);
@ -344,6 +361,7 @@ TEST(testCase, smlParseCols_Test) {
// small int
kv = (SSmlKv *)taosArrayGet(elements.colArray, 9);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "ci16", 4), 0);
ASSERT_EQ(kv->keyLen, 4);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_SMALLINT);
@ -352,6 +370,7 @@ TEST(testCase, smlParseCols_Test) {
// unsigned smallint
kv = (SSmlKv *)taosArrayGet(elements.colArray, 10);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cu16", 4), 0);
ASSERT_EQ(kv->keyLen, 4);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_USMALLINT);
@ -360,6 +379,7 @@ TEST(testCase, smlParseCols_Test) {
// int
kv = (SSmlKv *)taosArrayGet(elements.colArray, 11);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "ci32", 4), 0);
ASSERT_EQ(kv->keyLen, 4);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_INT);
@ -368,6 +388,7 @@ TEST(testCase, smlParseCols_Test) {
// unsigned int
kv = (SSmlKv *)taosArrayGet(elements.colArray, 12);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cu32", 4), 0);
ASSERT_EQ(kv->keyLen, 4);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_UINT);
@ -376,6 +397,7 @@ TEST(testCase, smlParseCols_Test) {
// bigint
kv = (SSmlKv *)taosArrayGet(elements.colArray, 13);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "ci64", 4), 0);
ASSERT_EQ(kv->keyLen, 4);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BIGINT);
@ -384,6 +406,7 @@ TEST(testCase, smlParseCols_Test) {
// bigint
kv = (SSmlKv *)taosArrayGet(elements.colArray, 14);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "ci", 2), 0);
ASSERT_EQ(kv->keyLen, 2);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BIGINT);
@ -392,6 +415,7 @@ TEST(testCase, smlParseCols_Test) {
// unsigned bigint
kv = (SSmlKv *)taosArrayGet(elements.colArray, 15);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cu64", 4), 0);
ASSERT_EQ(kv->keyLen, 4);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_UBIGINT);
@ -400,6 +424,7 @@ TEST(testCase, smlParseCols_Test) {
// bool
kv = (SSmlKv *)taosArrayGet(elements.colArray, 16);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cbooltrue", 9), 0);
ASSERT_EQ(kv->keyLen, 9);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BOOL);
@ -408,6 +433,7 @@ TEST(testCase, smlParseCols_Test) {
// bool
kv = (SSmlKv *)taosArrayGet(elements.colArray, 17);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cboolt", 6), 0);
ASSERT_EQ(kv->keyLen, 6);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BOOL);
@ -416,6 +442,7 @@ TEST(testCase, smlParseCols_Test) {
// bool
kv = (SSmlKv *)taosArrayGet(elements.colArray, 18);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cboolf", 6), 0);
ASSERT_EQ(kv->keyLen, 6);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BOOL);
@ -424,13 +451,14 @@ TEST(testCase, smlParseCols_Test) {
// nchar
kv = (SSmlKv *)taosArrayGet(elements.colArray, 19);
ASSERT_NE(kv, nullptr);
ASSERT_EQ(strncasecmp(kv->key, "cnch_", 5), 0);
ASSERT_EQ(kv->keyLen, 5);
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
ASSERT_EQ(kv->length, 4);
ASSERT_EQ(strncasecmp(kv->value, "iuwq", 4), 0);
taosArrayDestroy(elements.colArray);
(void)taosArrayDestroy(elements.colArray);
taosMemoryFree(sql);
smlDestroyInfo(info);
}
@ -471,7 +499,9 @@ TEST(testCase, smlParseNumber_Test) {
}
TEST(testCase, smlParseTelnetLine_error_Test) {
SSmlHandle *info = smlBuildSmlInfo(NULL);
SSmlHandle *info = nullptr;
int32_t code = smlBuildSmlInfo(nullptr, &info);
ASSERT_EQ(code, TSDB_CODE_SUCCESS);
info->dataFormat = false;
info->protocol = TSDB_SML_TELNET_PROTOCOL;
ASSERT_NE(info, nullptr);
@ -509,7 +539,9 @@ TEST(testCase, smlParseTelnetLine_error_Test) {
}
TEST(testCase, smlParseTelnetLine_Test) {
SSmlHandle *info = smlBuildSmlInfo(NULL);
SSmlHandle *info = nullptr;
int32_t code = smlBuildSmlInfo(nullptr, &info);
ASSERT_EQ(code, TSDB_CODE_SUCCESS);
info->dataFormat = false;
info->protocol = TSDB_SML_TELNET_PROTOCOL;
ASSERT_NE(info, nullptr);
@ -528,7 +560,9 @@ TEST(testCase, smlParseTelnetLine_Test) {
}
//TEST(testCase, smlParseTelnetLine_diff_json_type2_Test) {
// SSmlHandle *info = smlBuildSmlInfo(NULL);
// SSmlHandle *info = nullptr;
// int32_t code = smlBuildSmlInfo(nullptr, &info);
// ASSERT_EQ(code, TSDB_CODE_SUCCESS);
// info->protocol = TSDB_SML_JSON_PROTOCOL;
// ASSERT_NE(info, nullptr);
//
@ -568,7 +602,7 @@ TEST(testCase, smlParseNumber_performance_Test) {
for (int j = 0; j < 10000000; ++j) {
kv.value = str[i];
kv.length = strlen(str[i]);
smlParseNumber(&kv, &msgBuf);
(void)smlParseNumber(&kv, &msgBuf); //ignore result
}
printf("smlParseNumber:%s cost:%" PRId64, str[i], taosGetTimestampUs() - t1);
printf("\n");
@ -576,7 +610,7 @@ TEST(testCase, smlParseNumber_performance_Test) {
for (int j = 0; j < 10000000; ++j) {
kv.value = str[i];
kv.length = strlen(str[i]);
smlParseNumberOld(&kv, &msgBuf);
(void)smlParseNumberOld(&kv, &msgBuf); //ignore result
}
printf("smlParseNumberOld:%s cost:%" PRId64, str[i], taosGetTimestampUs() - t2);
printf("\n\n");

5
source/common/src/tdatablock.c
View File

@ -2486,7 +2486,10 @@ int32_t buildCtbNameByGroupIdImpl(const char* stbFullName, uint64_t groupId, cha
RandTableName rname = {
.tags = tags, .stbFullName = stbFullName, .stbFullNameLen = strlen(stbFullName), .ctbShortName = cname};
buildChildTableName(&rname);
int32_t code = buildChildTableName(&rname);
if(code != TSDB_CODE_SUCCESS){
return code;
}
taosArrayDestroy(tags);
if ((rname.ctbShortName && rname.ctbShortName[0]) == 0) {

13
source/common/src/tname.c
View File

@ -293,17 +293,19 @@ static int compareKv(const void* p1, const void* p2) {
/*
* use stable name and tags to grearate child table name
*/
void buildChildTableName(RandTableName* rName) {
int32_t buildChildTableName(RandTableName* rName) {
SStringBuilder sb = {0};
taosStringBuilderAppendStringLen(&sb, rName->stbFullName, rName->stbFullNameLen);
if (sb.buf == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY;
return;
return TSDB_CODE_OUT_OF_MEMORY;
}
taosArraySort(rName->tags, compareKv);
for (int j = 0; j < taosArrayGetSize(rName->tags); ++j) {
taosStringBuilderAppendChar(&sb, ',');
SSmlKv* tagKv = taosArrayGet(rName->tags, j);
if(tagKv == NULL) {
return TSDB_CODE_SML_INVALID_DATA;
}
taosStringBuilderAppendStringLen(&sb, tagKv->key, tagKv->keyLen);
taosStringBuilderAppendChar(&sb, '=');
if (IS_VAR_DATA_TYPE(tagKv->type)) {
@ -323,8 +325,9 @@ void buildChildTableName(RandTableName* rName) {
rName->ctbShortName[0] = 't';
rName->ctbShortName[1] = '_';
for (int i = 0; i < 16; i++) {
sprintf(temp, "%02x", context.digest[i]);
strcat(rName->ctbShortName, temp);
(void)sprintf(temp, "%02x", context.digest[i]);
(void)strcat(rName->ctbShortName, temp);
}
taosStringBuilderDestroy(&sb);
return TSDB_CODE_SUCCESS;
}

101
source/libs/parser/src/parInsertSml.c
View File

@ -112,7 +112,11 @@ static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchem
for (int i = 0; i < tags->numOfBound; ++i) {
SSchema* pTagSchema = &pSchema[tags->pColIndex[i]];
SSmlKv* kv = taosArrayGet(cols, i);
if (kv == NULL){
code = TSDB_CODE_SML_INVALID_DATA;
uError("SML smlBuildTagRow error kv is null");
goto end;
}
if (kv->keyLen != strlen(pTagSchema->name) || memcmp(kv->key, pTagSchema->name, kv->keyLen) != 0 ||
kv->type != pTagSchema->type) {
code = TSDB_CODE_SML_INVALID_DATA;
@ -120,7 +124,11 @@ static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchem
goto end;
}
taosArrayPush(*tagName, pTagSchema->name);
if (taosArrayPush(*tagName, pTagSchema->name) == NULL){
code = TSDB_CODE_OUT_OF_MEMORY;
uError("SML smlBuildTagRow error push tag name");
goto end;
}
STagVal val = {.cid = pTagSchema->colId, .type = pTagSchema->type};
// strcpy(val.colName, pTagSchema->name);
if (pTagSchema->type == TSDB_DATA_TYPE_BINARY || pTagSchema->type == TSDB_DATA_TYPE_VARBINARY ||
@ -141,7 +149,7 @@ static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchem
goto end;
}
char buf[512] = {0};
snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
(void)snprintf(buf, tListLen(buf), " taosMbsToUcs4 error:%s", strerror(errno));
taosMemoryFree(p);
code = buildSyntaxErrMsg(msg, buf, kv->value);
goto end;
@ -149,9 +157,13 @@ static int32_t smlBuildTagRow(SArray* cols, SBoundColInfo* tags, SSchema* pSchem
val.pData = p;
val.nData = output;
} else {
memcpy(&val.i64, &(kv->value), kv->length);
(void)memcpy(&val.i64, &(kv->value), kv->length);
}
if (taosArrayPush(pTagArray, &val) == NULL){
code = TSDB_CODE_OUT_OF_MEMORY;
uError("SML smlBuildTagRow error push tag array");
goto end;
}
taosArrayPush(pTagArray, &val);
}
code = tTagNew(pTagArray, 1, false, ppTag);
@ -166,20 +178,19 @@ end:
return code;
}
STableDataCxt* smlInitTableDataCtx(SQuery* query, STableMeta* pTableMeta) {
STableDataCxt* pTableCxt = NULL;
int32_t smlInitTableDataCtx(SQuery* query, STableMeta* pTableMeta, STableDataCxt** cxt) {
SVCreateTbReq* pCreateTbReq = NULL;
int ret = insGetTableDataCxt(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj, &pTableMeta->uid,
sizeof(pTableMeta->uid), pTableMeta, &pCreateTbReq, &pTableCxt, false, false);
sizeof(pTableMeta->uid), pTableMeta, &pCreateTbReq, cxt, false, false);
if (ret != TSDB_CODE_SUCCESS) {
return NULL;
return ret;
}
ret = initTableColSubmitData(pTableCxt);
ret = initTableColSubmitData(*cxt);
if (ret != TSDB_CODE_SUCCESS) {
return NULL;
return ret;
}
return pTableCxt;
return TSDB_CODE_SUCCESS;
}
void clearColValArraySml(SArray* pCols) {
@ -197,6 +208,9 @@ void clearColValArraySml(SArray* pCols) {
int32_t smlBuildRow(STableDataCxt* pTableCxt) {
SRow** pRow = taosArrayReserve(pTableCxt->pData->aRowP, 1);
if (pRow == NULL){
return TSDB_CODE_OUT_OF_MEMORY;
}
int ret = tRowBuild(pTableCxt->pValues, pTableCxt->pSchema, pRow);
if (TSDB_CODE_SUCCESS != ret) {
return ret;
@ -211,18 +225,22 @@ int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* data, int32
int ret = TSDB_CODE_SUCCESS;
SSchema* pColSchema = schema + index;
SColVal* pVal = taosArrayGet(pTableCxt->pValues, index);
if (pVal == NULL) {
return TSDB_CODE_SUCCESS;
}
SSmlKv* kv = (SSmlKv*)data;
if (kv->keyLen != strlen(pColSchema->name) || memcmp(kv->key, pColSchema->name, kv->keyLen) != 0 ||
kv->type != pColSchema->type) {
ret = TSDB_CODE_SML_INVALID_DATA;
char* tmp = taosMemoryCalloc(kv->keyLen + 1, 1);
if (tmp) {
memcpy(tmp, kv->key, kv->keyLen);
(void)memcpy(tmp, kv->key, kv->keyLen);
uInfo("SML data(name:%s type:%s) is not same like the db data(name:%s type:%s)", tmp, tDataTypes[kv->type].name,
pColSchema->name, tDataTypes[pColSchema->type].name);
taosMemoryFree(tmp);
} else {
uError("SML smlBuildCol out of memory");
ret = TSDB_CODE_OUT_OF_MEMORY;
}
goto end;
}
@ -256,9 +274,9 @@ int32_t smlBuildCol(STableDataCxt* pTableCxt, SSchema* schema, void* data, int32
} else if (kv->type == TSDB_DATA_TYPE_GEOMETRY || kv->type == TSDB_DATA_TYPE_VARBINARY) {
pVal->value.nData = kv->length;
pVal->value.pData = taosMemoryMalloc(kv->length);
memcpy(pVal->value.pData, (uint8_t*)kv->value, kv->length);
(void)memcpy(pVal->value.pData, (uint8_t*)kv->value, kv->length);
} else {
memcpy(&pVal->value.val, &(kv->value), kv->length);
(void)memcpy(&pVal->value.val, &(kv->value), kv->length);
}
pVal->flag = CV_FLAG_VALUE;
@ -276,8 +294,13 @@ int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSc
SVCreateTbReq* pCreateTblReq = NULL;
SArray* tagName = NULL;
insInitBoundColsInfo(getNumOfTags(pTableMeta), &bindTags);
int ret = smlBoundColumnData(tags, &bindTags, pTagsSchema, true);
int ret = insInitBoundColsInfo(getNumOfTags(pTableMeta), &bindTags);
if (ret != TSDB_CODE_SUCCESS) {
buildInvalidOperationMsg(&pBuf, "init bound cols error");
goto end;
}
ret = smlBoundColumnData(tags, &bindTags, pTagsSchema, true);
if (ret != TSDB_CODE_SUCCESS) {
buildInvalidOperationMsg(&pBuf, "bound tags error");
goto end;
@ -299,7 +322,11 @@ int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSc
ttl);
pCreateTblReq->ctb.stbName = taosMemoryCalloc(1, sTableNameLen + 1);
memcpy(pCreateTblReq->ctb.stbName, sTableName, sTableNameLen);
if (pCreateTblReq->ctb.stbName == NULL){
ret = TSDB_CODE_OUT_OF_MEMORY;
goto end;
}
(void)memcpy(pCreateTblReq->ctb.stbName, sTableName, sTableNameLen);
if (dataFormat) {
STableDataCxt** pTableCxt = (STableDataCxt**)taosHashGet(((SVnodeModifyOpStmt*)(query->pRoot))->pTableBlockHashObj,
@ -345,11 +372,18 @@ int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSc
for (int32_t r = 0; r < rowNum; ++r) {
void* rowData = taosArrayGetP(cols, r);
if (rowData == NULL) {
ret = TSDB_CODE_SML_INVALID_DATA;
goto end;
}
// 1. set the parsed value from sql string
for (int c = 0; c < pTableCxt->boundColsInfo.numOfBound; ++c) {
SSchema* pColSchema = &pSchema[pTableCxt->boundColsInfo.pColIndex[c]];
SColVal* pVal = taosArrayGet(pTableCxt->pValues, pTableCxt->boundColsInfo.pColIndex[c]);
if (pVal == NULL) {
ret = TSDB_CODE_SML_INVALID_DATA;
goto end;
}
void** p = taosHashGet(rowData, pColSchema->name, strlen(pColSchema->name));
if (p == NULL) {
continue;
@ -388,14 +422,22 @@ int32_t smlBindData(SQuery* query, bool dataFormat, SArray* tags, SArray* colsSc
} else if (kv->type == TSDB_DATA_TYPE_GEOMETRY || kv->type == TSDB_DATA_TYPE_VARBINARY) {
pVal->value.nData = kv->length;
pVal->value.pData = taosMemoryMalloc(kv->length);
memcpy(pVal->value.pData, (uint8_t*)kv->value, kv->length);
if (NULL == pVal->value.pData) {
ret = TSDB_CODE_OUT_OF_MEMORY;
goto end;
}
(void)memcpy(pVal->value.pData, (uint8_t*)kv->value, kv->length);
} else {
memcpy(&pVal->value.val, &(kv->value), kv->length);
(void)memcpy(&pVal->value.val, &(kv->value), kv->length);
}
pVal->flag = CV_FLAG_VALUE;
}
SRow** pRow = taosArrayReserve(pTableCxt->pData->aRowP, 1);
if (NULL == pRow) {
ret = TSDB_CODE_OUT_OF_MEMORY;
goto end;
}
ret = tRowBuild(pTableCxt->pValues, pTableCxt->pSchema, pRow);
if (TSDB_CODE_SUCCESS != ret) {
buildInvalidOperationMsg(&pBuf, "tRowBuild error");
@ -415,11 +457,12 @@ end:
return ret;
}
SQuery* smlInitHandle() {
int32_t smlInitHandle(SQuery** query) {
*query = NULL;
SQuery* pQuery = (SQuery*)nodesMakeNode(QUERY_NODE_QUERY);
if (NULL == pQuery) {
uError("create pQuery error");
return NULL;
return TSDB_CODE_OUT_OF_MEMORY;
}
pQuery->execMode = QUERY_EXEC_MODE_SCHEDULE;
pQuery->haveResultSet = false;
@ -428,14 +471,22 @@ SQuery* smlInitHandle() {
if (NULL == stmt) {
uError("create SVnodeModifyOpStmt error");
qDestroyQuery(pQuery);
return NULL;
return TSDB_CODE_OUT_OF_MEMORY;
}
stmt->pTableBlockHashObj = taosHashInit(16, taosGetDefaultHashFunction(TSDB_DATA_TYPE_BIGINT), true, HASH_NO_LOCK);
if (stmt->pTableBlockHashObj == NULL){
uError("create pTableBlockHashObj error");
qDestroyQuery(pQuery);
nodesDestroyNode((SNode*)stmt);
return TSDB_CODE_OUT_OF_MEMORY;
}
stmt->freeHashFunc = insDestroyTableDataCxtHashMap;
stmt->freeArrayFunc = insDestroyVgroupDataCxtList;
pQuery->pRoot = (SNode*)stmt;
return pQuery;
*query = pQuery;
return TSDB_CODE_SUCCESS;
}
int32_t smlBuildOutput(SQuery* handle, SHashObj* pVgHash) {