Merge pull request #20970 from taosdata/fix/TS-3116
opti:escape logic in schemaless
This commit is contained in:
commit
3f432cca2e
|
@ -341,6 +341,8 @@ typedef struct {
|
|||
float f;
|
||||
};
|
||||
size_t length;
|
||||
bool keyEscaped;
|
||||
bool valueEscaped;
|
||||
} SSmlKv;
|
||||
|
||||
#define QUERY_ASC_FORWARD_STEP 1
|
||||
|
|
|
@ -107,6 +107,7 @@ typedef struct {
|
|||
int32_t colsLen;
|
||||
int32_t timestampLen;
|
||||
|
||||
bool measureEscaped;
|
||||
SArray *colArray;
|
||||
} SSmlLineInfo;
|
||||
|
||||
|
@ -206,6 +207,19 @@ typedef struct {
|
|||
|
||||
#define IS_SAME_KEY (maxKV->keyLen == kv.keyLen && memcmp(maxKV->key, kv.key, kv.keyLen) == 0)
|
||||
|
||||
#define IS_SLASH_LETTER_IN_MEASUREMENT(sql) \
|
||||
(*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE))
|
||||
|
||||
#define MOVE_FORWARD_ONE(sql, len) (memmove((void *)((sql)-1), (sql), len))
|
||||
|
||||
#define PROCESS_SLASH_IN_MEASUREMENT(key, keyLen) \
|
||||
for (int i = 1; i < keyLen; ++i) { \
|
||||
if (IS_SLASH_LETTER_IN_MEASUREMENT(key + i)) { \
|
||||
MOVE_FORWARD_ONE(key + i, keyLen - i); \
|
||||
keyLen--; \
|
||||
} \
|
||||
}
|
||||
|
||||
extern int64_t smlFactorNS[3];
|
||||
extern int64_t smlFactorS[3];
|
||||
|
||||
|
@ -237,6 +251,7 @@ uint8_t smlGetTimestampLen(int64_t num);
|
|||
void clearColValArray(SArray* pCols);
|
||||
void smlDestroyTableInfo(SSmlHandle *info, SSmlTableInfo *tag);
|
||||
|
||||
void freeSSmlKv(void* data);
|
||||
int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
|
||||
int32_t smlParseTelnetString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLineInfo *elements);
|
||||
int32_t smlParseJSON(SSmlHandle *info, char *payload);
|
||||
|
|
|
@ -765,8 +765,12 @@ static int32_t smlModifyDBSchemas(SSmlHandle *info) {
|
|||
|
||||
size_t superTableLen = 0;
|
||||
void *superTable = taosHashGetKey(tmp, &superTableLen);
|
||||
char* measure = taosMemoryMalloc(superTableLen);
|
||||
memcpy(measure, superTable, superTableLen);
|
||||
PROCESS_SLASH_IN_MEASUREMENT(measure, superTableLen);
|
||||
memset(pName.tname, 0, TSDB_TABLE_NAME_LEN);
|
||||
memcpy(pName.tname, superTable, superTableLen);
|
||||
memcpy(pName.tname, measure, superTableLen);
|
||||
taosMemoryFree(measure);
|
||||
|
||||
code = catalogGetSTableMeta(info->pCatalog, &conn, &pName, &pTableMeta);
|
||||
|
||||
|
@ -1049,7 +1053,7 @@ void smlDestroyTableInfo(SSmlHandle *info, SSmlTableInfo *tag) {
|
|||
// }
|
||||
// taosMemoryFree(tag->key);
|
||||
taosArrayDestroy(tag->cols);
|
||||
taosArrayDestroy(tag->tags);
|
||||
taosArrayDestroyEx(tag->tags, freeSSmlKv);
|
||||
taosMemoryFree(tag);
|
||||
}
|
||||
|
||||
|
@ -1063,6 +1067,12 @@ void clearColValArray(SArray *pCols) {
|
|||
}
|
||||
}
|
||||
|
||||
void freeSSmlKv(void* data){
|
||||
SSmlKv *kv = (SSmlKv*)data;
|
||||
if(kv->keyEscaped) taosMemoryFree((void*)(kv->key));
|
||||
if(kv->valueEscaped) taosMemoryFree((void*)(kv->value));
|
||||
}
|
||||
|
||||
void smlDestroyInfo(SSmlHandle *info) {
|
||||
if (!info) return;
|
||||
qDestroyQuery(info->pQuery);
|
||||
|
@ -1098,11 +1108,11 @@ void smlDestroyInfo(SSmlHandle *info) {
|
|||
}
|
||||
taosArrayDestroy(info->valueJsonArray);
|
||||
|
||||
taosArrayDestroy(info->preLineTagKV);
|
||||
taosArrayDestroyEx(info->preLineTagKV, freeSSmlKv);
|
||||
|
||||
if (!info->dataFormat) {
|
||||
for (int i = 0; i < info->lineNum; i++) {
|
||||
taosArrayDestroy(info->lines[i].colArray);
|
||||
taosArrayDestroyEx(info->lines[i].colArray, freeSSmlKv);
|
||||
if (info->parseJsonByLib) {
|
||||
taosMemoryFree(info->lines[i].tags);
|
||||
}
|
||||
|
@ -1165,8 +1175,10 @@ static int32_t smlPushCols(SArray *colsArray, SArray *cols) {
|
|||
}
|
||||
for (size_t i = 0; i < taosArrayGetSize(cols); i++) {
|
||||
SSmlKv *kv = (SSmlKv *)taosArrayGet(cols, i);
|
||||
terrno = 0;
|
||||
taosHashPut(kvHash, kv->key, kv->keyLen, &kv, POINTER_BYTES);
|
||||
if (terrno == TSDB_CODE_DUP_KEY) {
|
||||
taosHashCleanup(kvHash);
|
||||
return terrno;
|
||||
}
|
||||
}
|
||||
|
@ -1240,12 +1252,13 @@ static int32_t smlParseLineBottom(SSmlHandle *info) {
|
|||
uDebug("SML:0x%" PRIx64 " smlParseLineBottom add meta, format:%d, linenum:%d", info->id, info->dataFormat,
|
||||
info->lineNum);
|
||||
SSmlSTableMeta *meta = smlBuildSTableMeta(info->dataFormat);
|
||||
taosHashPut(info->superTables, elements->measure, elements->measureLen, &meta, POINTER_BYTES);
|
||||
terrno = 0;
|
||||
smlInsertMeta(meta->tagHash, meta->tags, tinfo->tags);
|
||||
if (terrno == TSDB_CODE_DUP_KEY) {
|
||||
return terrno;
|
||||
}
|
||||
smlInsertMeta(meta->colHash, meta->cols, elements->colArray);
|
||||
taosHashPut(info->superTables, elements->measure, elements->measureLen, &meta, POINTER_BYTES);
|
||||
}
|
||||
}
|
||||
uDebug("SML:0x%" PRIx64 " smlParseLineBottom end, format:%d, linenum:%d", info->id, info->dataFormat, info->lineNum);
|
||||
|
@ -1304,9 +1317,15 @@ static int32_t smlInsertData(SSmlHandle *info) {
|
|||
uDebug("SML:0x%" PRIx64 " smlInsertData table:%s, uid:%" PRIu64 ", format:%d", info->id, pName.tname,
|
||||
tableData->uid, info->dataFormat);
|
||||
|
||||
int measureLen = tableData->sTableNameLen;
|
||||
char* measure = (char*)taosMemoryMalloc(tableData->sTableNameLen);
|
||||
memcpy(measure, tableData->sTableName, tableData->sTableNameLen);
|
||||
PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen);
|
||||
|
||||
code = smlBindData(info->pQuery, info->dataFormat, tableData->tags, (*pMeta)->cols, tableData->cols,
|
||||
(*pMeta)->tableMeta, tableData->childTableName, tableData->sTableName, tableData->sTableNameLen,
|
||||
(*pMeta)->tableMeta, tableData->childTableName, measure, measureLen,
|
||||
info->ttl, info->msgBuf.buf, info->msgBuf.len);
|
||||
taosMemoryFree(measure);
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
uError("SML:0x%" PRIx64 " smlBindData failed", info->id);
|
||||
return code;
|
||||
|
@ -1420,14 +1439,14 @@ static int32_t smlParseLine(SSmlHandle *info, char *lines[], char *rawLine, char
|
|||
|
||||
char cTmp = 0; // for print tmp if is raw
|
||||
if (info->isRawLine) {
|
||||
cTmp = tmp[len - 1];
|
||||
tmp[len - 1] = '\0';
|
||||
cTmp = tmp[len];
|
||||
tmp[len] = '\0';
|
||||
}
|
||||
|
||||
uDebug("SML:0x%" PRIx64 " smlParseLine israw:%d, numLines:%d, protocol:%d, len:%d, sql:%s", info->id,
|
||||
info->isRawLine, numLines, info->protocol, len, tmp);
|
||||
if (info->isRawLine) {
|
||||
tmp[len - 1] = cTmp;
|
||||
tmp[len] = cTmp;
|
||||
}
|
||||
|
||||
if (info->protocol == TSDB_SML_LINE_PROTOCOL) {
|
||||
|
@ -1449,6 +1468,7 @@ static int32_t smlParseLine(SSmlHandle *info, char *lines[], char *rawLine, char
|
|||
code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE;
|
||||
}
|
||||
if (code != TSDB_CODE_SUCCESS) {
|
||||
tmp[len] = '\0';
|
||||
uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, tmp);
|
||||
return code;
|
||||
}
|
||||
|
|
|
@ -21,32 +21,33 @@
|
|||
#include "clientSml.h"
|
||||
|
||||
// comma ,
|
||||
// #define IS_SLASH_COMMA(sql) (*(sql) == COMMA && *((sql)-1) == SLASH)
|
||||
#define IS_COMMA(sql) (*(sql) == COMMA && *((sql)-1) != SLASH)
|
||||
// space
|
||||
// #define IS_SLASH_SPACE(sql) (*(sql) == SPACE && *((sql)-1) == SLASH)
|
||||
#define IS_SPACE(sql) (*(sql) == SPACE && *((sql)-1) != SLASH)
|
||||
// equal =
|
||||
// #define IS_SLASH_EQUAL(sql) (*(sql) == EQUAL && *((sql)-1) == SLASH)
|
||||
#define IS_EQUAL(sql) (*(sql) == EQUAL && *((sql)-1) != SLASH)
|
||||
// quote "
|
||||
// #define IS_SLASH_QUOTE(sql) (*(sql) == QUOTE && *((sql)-1) == SLASH)
|
||||
#define IS_QUOTE(sql) (*(sql) == QUOTE && *((sql)-1) != SLASH)
|
||||
//#define IS_QUOTE(sql) (*(sql) == QUOTE && *((sql)-1) != SLASH)
|
||||
// SLASH
|
||||
// #define IS_SLASH_SLASH(sql) (*(sql) == SLASH && *((sql)-1) == SLASH)
|
||||
|
||||
#define IS_SLASH_LETTER(sql) \
|
||||
(*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE || *(sql) == EQUAL || *(sql) == QUOTE || \
|
||||
*(sql) == SLASH)) // (IS_SLASH_COMMA(sql) || IS_SLASH_SPACE(sql) || IS_SLASH_EQUAL(sql) ||
|
||||
// IS_SLASH_QUOTE(sql) || IS_SLASH_SLASH(sql))
|
||||
#define IS_SLASH_LETTER_IN_FIELD_VALUE(sql) \
|
||||
(*((sql)-1) == SLASH && (*(sql) == QUOTE || *(sql) == SLASH))
|
||||
|
||||
#define MOVE_FORWARD_ONE(sql, len) (memmove((void *)((sql)-1), (sql), len))
|
||||
#define IS_SLASH_LETTER_IN_TAG_FIELD_KEY(sql) \
|
||||
(*((sql)-1) == SLASH && (*(sql) == COMMA || *(sql) == SPACE || *(sql) == EQUAL))
|
||||
|
||||
#define PROCESS_SLASH(key, keyLen) \
|
||||
for (int i = 1; i < keyLen; ++i) { \
|
||||
if (IS_SLASH_LETTER(key + i)) { \
|
||||
#define PROCESS_SLASH_IN_FIELD_VALUE(key, keyLen) \
|
||||
for (int i = 1; i < keyLen; ++i) { \
|
||||
if (IS_SLASH_LETTER_IN_FIELD_VALUE(key + i)) { \
|
||||
MOVE_FORWARD_ONE(key + i, keyLen - i); \
|
||||
keyLen--; \
|
||||
} \
|
||||
}
|
||||
|
||||
#define PROCESS_SLASH_IN_TAG_FIELD_KEY(key, keyLen) \
|
||||
for (int i = 1; i < keyLen; ++i) { \
|
||||
if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(key + i)) { \
|
||||
MOVE_FORWARD_ONE(key + i, keyLen - i); \
|
||||
i--; \
|
||||
keyLen--; \
|
||||
} \
|
||||
}
|
||||
|
@ -151,7 +152,17 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
|
||||
SSmlSTableMeta *sMeta = NULL;
|
||||
if (unlikely(tmp == NULL)) {
|
||||
STableMeta *pTableMeta = smlGetMeta(info, currElement->measure, currElement->measureLen);
|
||||
char* measure = currElement->measure;
|
||||
int measureLen = currElement->measureLen;
|
||||
if(currElement->measureEscaped){
|
||||
measure = (char*)taosMemoryMalloc(currElement->measureLen);
|
||||
memcpy(measure, currElement->measure, currElement->measureLen);
|
||||
PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen);
|
||||
}
|
||||
STableMeta *pTableMeta = smlGetMeta(info, measure, measureLen);
|
||||
if(currElement->measureEscaped){
|
||||
taosMemoryFree(measure);
|
||||
}
|
||||
if (pTableMeta == NULL) {
|
||||
info->dataFormat = false;
|
||||
info->reRun = true;
|
||||
|
@ -171,17 +182,18 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
info->maxTagKVs = (*tmp)->tags;
|
||||
}
|
||||
}
|
||||
taosArrayClear(preLineKV);
|
||||
taosArrayClearEx(preLineKV, freeSSmlKv);
|
||||
|
||||
while (*sql < sqlEnd) {
|
||||
if (unlikely(IS_SPACE(*sql))) {
|
||||
break;
|
||||
}
|
||||
|
||||
bool hasSlash = false;
|
||||
// parse key
|
||||
const char *key = *sql;
|
||||
size_t keyLen = 0;
|
||||
bool keyEscaped = false;
|
||||
size_t keyLenEscaped = 0;
|
||||
while (*sql < sqlEnd) {
|
||||
if (unlikely(IS_COMMA(*sql))) {
|
||||
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
|
||||
|
@ -192,16 +204,14 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
(*sql)++;
|
||||
break;
|
||||
}
|
||||
if (!hasSlash) {
|
||||
hasSlash = (*(*sql) == SLASH);
|
||||
if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
|
||||
keyLenEscaped++;
|
||||
keyEscaped = true;
|
||||
}
|
||||
(*sql)++;
|
||||
}
|
||||
if (unlikely(hasSlash)) {
|
||||
PROCESS_SLASH(key, keyLen)
|
||||
}
|
||||
|
||||
if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
|
||||
if (unlikely(IS_INVALID_COL_LEN(keyLen - keyLenEscaped))) {
|
||||
smlBuildInvalidDataMsg(&info->msgBuf, "invalid key or key is too long than 64", key);
|
||||
return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
|
||||
}
|
||||
|
@ -209,7 +219,8 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
// parse value
|
||||
const char *value = *sql;
|
||||
size_t valueLen = 0;
|
||||
hasSlash = false;
|
||||
bool valueEscaped = false;
|
||||
size_t valueLenEscaped = 0;
|
||||
while (*sql < sqlEnd) {
|
||||
// parse value
|
||||
if (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql))) {
|
||||
|
@ -219,8 +230,9 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
return TSDB_CODE_SML_INVALID_DATA;
|
||||
}
|
||||
|
||||
if (!hasSlash) {
|
||||
hasSlash = (*(*sql) == SLASH);
|
||||
if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
|
||||
valueLenEscaped++;
|
||||
valueEscaped = true;
|
||||
}
|
||||
|
||||
(*sql)++;
|
||||
|
@ -232,15 +244,24 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
return TSDB_CODE_SML_INVALID_DATA;
|
||||
}
|
||||
|
||||
if (unlikely(hasSlash)) {
|
||||
PROCESS_SLASH(value, valueLen)
|
||||
}
|
||||
|
||||
if (unlikely(valueLen > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
|
||||
if (unlikely(valueLen - valueLenEscaped > (TSDB_MAX_NCHAR_LEN - VARSTR_HEADER_SIZE) / TSDB_NCHAR_SIZE)) {
|
||||
return TSDB_CODE_PAR_INVALID_VAR_COLUMN_LEN;
|
||||
}
|
||||
|
||||
SSmlKv kv = {.key = key, .keyLen = keyLen, .type = TSDB_DATA_TYPE_NCHAR, .value = value, .length = valueLen};
|
||||
if (keyEscaped){
|
||||
char *tmp = (char*)taosMemoryMalloc(keyLen);
|
||||
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);
|
||||
PROCESS_SLASH_IN_TAG_FIELD_KEY(tmp, valueLen);
|
||||
value = tmp;
|
||||
}
|
||||
SSmlKv kv = {.key = key, .keyLen = keyLen, .type = TSDB_DATA_TYPE_NCHAR, .value = value, .length = valueLen, .keyEscaped = keyEscaped, .valueEscaped = valueEscaped};
|
||||
taosArrayPush(preLineKV, &kv);
|
||||
if (info->dataFormat) {
|
||||
if (unlikely(cnt + 1 > info->currSTableMeta->tableInfo.numOfTags)) {
|
||||
info->dataFormat = false;
|
||||
|
@ -266,7 +287,6 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
info->needModifySchema = true;
|
||||
}
|
||||
}
|
||||
taosArrayPush(preLineKV, &kv);
|
||||
|
||||
cnt++;
|
||||
if (IS_SPACE(*sql)) {
|
||||
|
@ -285,6 +305,11 @@ static int32_t smlParseTagKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
return TSDB_CODE_OUT_OF_MEMORY;
|
||||
}
|
||||
tinfo->tags = taosArrayDup(preLineKV, NULL);
|
||||
for(size_t i = 0; i < taosArrayGetSize(preLineKV); i++){
|
||||
SSmlKv *kv = (SSmlKv *)taosArrayGet(preLineKV, i);
|
||||
if(kv->keyEscaped)kv->key = NULL;
|
||||
if(kv->valueEscaped)kv->value = NULL;
|
||||
}
|
||||
|
||||
smlSetCTableName(tinfo);
|
||||
tinfo->uid = info->uid++;
|
||||
|
@ -321,7 +346,17 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
SSmlSTableMeta **tmp =
|
||||
(SSmlSTableMeta **)taosHashGet(info->superTables, currElement->measure, currElement->measureLen);
|
||||
if (unlikely(tmp == NULL)) {
|
||||
STableMeta *pTableMeta = smlGetMeta(info, currElement->measure, currElement->measureLen);
|
||||
char* measure = currElement->measure;
|
||||
int measureLen = currElement->measureLen;
|
||||
if(currElement->measureEscaped){
|
||||
measure = (char*)taosMemoryMalloc(currElement->measureLen);
|
||||
memcpy(measure, currElement->measure, currElement->measureLen);
|
||||
PROCESS_SLASH_IN_MEASUREMENT(measure, measureLen);
|
||||
}
|
||||
STableMeta *pTableMeta = smlGetMeta(info, measure, measureLen);
|
||||
if(currElement->measureEscaped){
|
||||
taosMemoryFree(measure);
|
||||
}
|
||||
if (pTableMeta == NULL) {
|
||||
info->dataFormat = false;
|
||||
info->reRun = true;
|
||||
|
@ -352,10 +387,11 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
break;
|
||||
}
|
||||
|
||||
bool hasSlash = false;
|
||||
// parse key
|
||||
const char *key = *sql;
|
||||
size_t keyLen = 0;
|
||||
bool keyEscaped = false;
|
||||
size_t keyLenEscaped = 0;
|
||||
while (*sql < sqlEnd) {
|
||||
if (unlikely(IS_COMMA(*sql))) {
|
||||
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
|
||||
|
@ -366,16 +402,14 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
(*sql)++;
|
||||
break;
|
||||
}
|
||||
if (!hasSlash) {
|
||||
hasSlash = (*(*sql) == SLASH);
|
||||
if (IS_SLASH_LETTER_IN_TAG_FIELD_KEY(*sql)) {
|
||||
keyLenEscaped++;
|
||||
keyEscaped = true;
|
||||
}
|
||||
(*sql)++;
|
||||
}
|
||||
if (unlikely(hasSlash)) {
|
||||
PROCESS_SLASH(key, keyLen)
|
||||
}
|
||||
|
||||
if (unlikely(IS_INVALID_COL_LEN(keyLen))) {
|
||||
if (unlikely(IS_INVALID_COL_LEN(keyLen - keyLenEscaped))) {
|
||||
smlBuildInvalidDataMsg(&info->msgBuf, "invalid key or key is too long than 64", key);
|
||||
return TSDB_CODE_TSC_INVALID_COLUMN_LENGTH;
|
||||
}
|
||||
|
@ -383,11 +417,13 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
// parse value
|
||||
const char *value = *sql;
|
||||
size_t valueLen = 0;
|
||||
hasSlash = false;
|
||||
bool isInQuote = false;
|
||||
bool valueEscaped = false;
|
||||
size_t valueLenEscaped = 0;
|
||||
bool isInQuote = false;
|
||||
const char *escapeChar = NULL;
|
||||
while (*sql < sqlEnd) {
|
||||
// parse value
|
||||
if (unlikely(IS_QUOTE(*sql))) {
|
||||
if (unlikely(*(*sql) == QUOTE && (*(*sql - 1) != SLASH || (*sql - 1) == escapeChar))) {
|
||||
isInQuote = !isInQuote;
|
||||
(*sql)++;
|
||||
continue;
|
||||
|
@ -395,13 +431,12 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
if (!isInQuote) {
|
||||
if (unlikely(IS_SPACE(*sql) || IS_COMMA(*sql))) {
|
||||
break;
|
||||
} else if (unlikely(IS_EQUAL(*sql))) {
|
||||
smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", *sql);
|
||||
return TSDB_CODE_SML_INVALID_DATA;
|
||||
}
|
||||
}
|
||||
if (!hasSlash) {
|
||||
hasSlash = (*(*sql) == SLASH);
|
||||
if (IS_SLASH_LETTER_IN_FIELD_VALUE(*sql) && (*sql - 1) != escapeChar) {
|
||||
escapeChar = *sql;
|
||||
valueEscaped = true;
|
||||
valueLenEscaped++;
|
||||
}
|
||||
|
||||
(*sql)++;
|
||||
|
@ -416,9 +451,6 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", value);
|
||||
return TSDB_CODE_SML_INVALID_DATA;
|
||||
}
|
||||
if (unlikely(hasSlash)) {
|
||||
PROCESS_SLASH(value, valueLen)
|
||||
}
|
||||
|
||||
SSmlKv kv = {.key = key, .keyLen = keyLen, .value = value, .length = valueLen};
|
||||
int32_t ret = smlParseValue(&kv, &info->msgBuf);
|
||||
|
@ -427,11 +459,28 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
return ret;
|
||||
}
|
||||
|
||||
if (keyEscaped){
|
||||
char *tmp = (char*)taosMemoryMalloc(kv.keyLen);
|
||||
memcpy(tmp, key, kv.keyLen);
|
||||
PROCESS_SLASH_IN_TAG_FIELD_KEY(tmp, kv.keyLen);
|
||||
kv.key = tmp;
|
||||
kv.keyEscaped = keyEscaped;
|
||||
}
|
||||
|
||||
if (valueEscaped){
|
||||
char *tmp = (char*)taosMemoryMalloc(kv.length);
|
||||
memcpy(tmp, kv.value, kv.length);
|
||||
PROCESS_SLASH_IN_FIELD_VALUE(tmp, kv.length);
|
||||
kv.value = tmp;
|
||||
kv.valueEscaped = valueEscaped;
|
||||
}
|
||||
|
||||
if (info->dataFormat) {
|
||||
// cnt begin 0, add ts so + 2
|
||||
if (unlikely(cnt + 2 > info->currSTableMeta->tableInfo.numOfColumns)) {
|
||||
info->dataFormat = false;
|
||||
info->reRun = true;
|
||||
freeSSmlKv(&kv);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
// bind data
|
||||
|
@ -440,22 +489,26 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
uDebug("smlBuildCol error, retry");
|
||||
info->dataFormat = false;
|
||||
info->reRun = true;
|
||||
freeSSmlKv(&kv);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
if (cnt >= taosArrayGetSize(info->masColKVs)) {
|
||||
info->dataFormat = false;
|
||||
info->reRun = true;
|
||||
freeSSmlKv(&kv);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
SSmlKv *maxKV = (SSmlKv *)taosArrayGet(info->masColKVs, cnt);
|
||||
if (kv.type != maxKV->type) {
|
||||
info->dataFormat = false;
|
||||
info->reRun = true;
|
||||
freeSSmlKv(&kv);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
if (unlikely(!IS_SAME_KEY)) {
|
||||
info->dataFormat = false;
|
||||
info->reRun = true;
|
||||
freeSSmlKv(&kv);
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
|
@ -463,6 +516,7 @@ static int32_t smlParseColKv(SSmlHandle *info, char **sql, char *sqlEnd, SSmlLin
|
|||
maxKV->length = kv.length;
|
||||
info->needModifySchema = true;
|
||||
}
|
||||
freeSSmlKv(&kv);
|
||||
} else {
|
||||
if (currElement->colArray == NULL) {
|
||||
currElement->colArray = taosArrayInit_s(sizeof(SSmlKv), 1);
|
||||
|
@ -487,10 +541,12 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
|
|||
elements->measure = sql;
|
||||
|
||||
// parse measure
|
||||
size_t measureLenEscaped = 0;
|
||||
while (sql < sqlEnd) {
|
||||
if (unlikely((sql != elements->measure) && IS_SLASH_LETTER(sql))) {
|
||||
MOVE_FORWARD_ONE(sql, sqlEnd - sql);
|
||||
sqlEnd--;
|
||||
if (unlikely((sql != elements->measure) && IS_SLASH_LETTER_IN_MEASUREMENT(sql))) {
|
||||
elements->measureEscaped = true;
|
||||
measureLenEscaped++;
|
||||
sql++;
|
||||
continue;
|
||||
}
|
||||
if (unlikely(IS_COMMA(sql))) {
|
||||
|
@ -503,7 +559,7 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
|
|||
sql++;
|
||||
}
|
||||
elements->measureLen = sql - elements->measure;
|
||||
if (unlikely(IS_INVALID_TABLE_LEN(elements->measureLen))) {
|
||||
if (unlikely(IS_INVALID_TABLE_LEN(elements->measureLen - measureLenEscaped))) {
|
||||
smlBuildInvalidDataMsg(&info->msgBuf, "measure is empty or too large than 192", NULL);
|
||||
return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH;
|
||||
}
|
||||
|
@ -581,7 +637,9 @@ int32_t smlParseInfluxString(SSmlHandle *info, char *sql, char *sqlEnd, SSmlLine
|
|||
.keyLen = TS_LEN,
|
||||
.type = TSDB_DATA_TYPE_TIMESTAMP,
|
||||
.i = ts,
|
||||
.length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes};
|
||||
.length = (size_t)tDataTypes[TSDB_DATA_TYPE_TIMESTAMP].bytes,
|
||||
.keyEscaped = false,
|
||||
.valueEscaped = false};
|
||||
if (info->dataFormat) {
|
||||
uDebug("SML:0x%" PRIx64 " smlParseInfluxString format true, ts:%" PRId64, info->id, ts);
|
||||
ret = smlBuildCol(info->currTableDataCtx, info->currSTableMeta->schema, &kv, 0);
|
||||
|
|
|
@ -1243,9 +1243,6 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
|
|||
taosMemoryFree(pParam);
|
||||
|
||||
if (code != 0) {
|
||||
tscWarn("consumer:0x%" PRIx64 " msg from vgId:%d discarded, epoch %d, since %s, reqId:0x%" PRIx64, tmq->consumerId,
|
||||
vgId, epoch, tstrerror(code), requestId);
|
||||
|
||||
if (pMsg->pData) taosMemoryFree(pMsg->pData);
|
||||
if (pMsg->pEpSet) taosMemoryFree(pMsg->pEpSet);
|
||||
|
||||
|
@ -1267,6 +1264,9 @@ int32_t tmqPollCb(void* param, SDataBuf* pMsg, int32_t code) {
|
|||
taosWriteQitem(tmq->mqueue, pRspWrapper);
|
||||
} else if (code == TSDB_CODE_WAL_LOG_NOT_EXIST) { // poll data while insert
|
||||
taosMsleep(500);
|
||||
} else{
|
||||
tscError("consumer:0x%" PRIx64 " msg from vgId:%d discarded, epoch %d, since %s, reqId:0x%" PRIx64, tmq->consumerId,
|
||||
vgId, epoch, tstrerror(code), requestId);
|
||||
}
|
||||
|
||||
goto CREATE_MSG_FAIL;
|
||||
|
|
|
@ -50,8 +50,9 @@ TEST(testCase, smlParseInfluxString_Test) {
|
|||
int ret = smlParseInfluxString(info, sql, sql + strlen(sql), &elements);
|
||||
ASSERT_EQ(ret, 0);
|
||||
ASSERT_EQ(elements.measure, sql);
|
||||
ASSERT_EQ(elements.measureLen, strlen(",st"));
|
||||
ASSERT_EQ(elements.measureTagsLen, strlen(",st,t1=3,t2=4,t3=t3"));
|
||||
ASSERT_EQ(elements.measureLen, strlen("\\,st"));
|
||||
ASSERT_EQ(elements.measureEscaped, true);
|
||||
ASSERT_EQ(elements.measureTagsLen, strlen("\\,st,t1=3,t2=4,t3=t3"));
|
||||
|
||||
ASSERT_EQ(elements.tags, sql + elements.measureLen + 1);
|
||||
ASSERT_EQ(elements.tagsLen, strlen("t1=3,t2=4,t3=t3"));
|
||||
|
@ -204,7 +205,26 @@ TEST(testCase, smlParseCols_Error_Test) {
|
|||
"st,t=1 c=-3.402823466e+39u64 1626006833639000000",
|
||||
"st,t=1 c=-339u64 1626006833639000000",
|
||||
"st,t=1 c=18446744073709551616u64 1626006833639000000",
|
||||
"st,t=1 c=1=2 1626006833639000000"};
|
||||
"st,t=1 c=1=2 1626006833639000000,",
|
||||
// escape error test
|
||||
// measure comma,space
|
||||
"s,t,t=1 c=1 1626006833639000000,",
|
||||
"s t,t=1 c=1 1626006833639000000,",
|
||||
//tag key comma,equal,space
|
||||
"st,t,t=1 c=2 1626006833639000000,",
|
||||
"st,t=t=1 c=2 1626006833639000000,",
|
||||
"st,t t=1 c=2 1626006833639000000,",
|
||||
//tag value comma,equal,space
|
||||
"st,tt=a,a c=2 1626006833639000000,",
|
||||
"st,t=t=a a c=2 1626006833639000000,",
|
||||
"st,t t=a=a c=2 1626006833639000000,",
|
||||
//field key comma,equal,space
|
||||
"st,tt=aa c,1=2 1626006833639000000,",
|
||||
"st,tt=aa c=1=2 1626006833639000000,",
|
||||
"st,tt=aa c 1=2 1626006833639000000,",
|
||||
//field value double quote,slash
|
||||
"st,tt=aa c=\"a\"a\" 1626006833639000000,",
|
||||
};
|
||||
|
||||
SSmlHandle *info = smlBuildSmlInfo(NULL);
|
||||
info->protocol = TSDB_SML_LINE_PROTOCOL;
|
||||
|
@ -256,16 +276,18 @@ TEST(testCase, smlParseCols_Test) {
|
|||
ASSERT_EQ(strncasecmp(kv->key, "cb=in", 5), 0);
|
||||
ASSERT_EQ(kv->keyLen, 5);
|
||||
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_BINARY);
|
||||
ASSERT_EQ(kv->length, 17);
|
||||
ASSERT_EQ(strncasecmp(kv->value, "pass,it ", 8), 0);
|
||||
ASSERT_EQ(kv->length, 18);
|
||||
ASSERT_EQ(kv->keyEscaped, true);
|
||||
ASSERT_EQ(kv->valueEscaped, false);
|
||||
ASSERT_EQ(strncasecmp(kv->value, "pass\\,it ", 9), 0);
|
||||
|
||||
// nchar
|
||||
kv = (SSmlKv *)taosArrayGet(elements.colArray, 2);
|
||||
ASSERT_EQ(strncasecmp(kv->key, "cnch", 4), 0);
|
||||
ASSERT_EQ(kv->keyLen, 4);
|
||||
ASSERT_EQ(kv->type, TSDB_DATA_TYPE_NCHAR);
|
||||
ASSERT_EQ(kv->length, 8);
|
||||
ASSERT_EQ(strncasecmp(kv->value, "ii=sd", 5), 0);
|
||||
ASSERT_EQ(kv->length, 9);
|
||||
ASSERT_EQ(strncasecmp(kv->value, "ii\\=sd", 5), 0);
|
||||
|
||||
// bool
|
||||
kv = (SSmlKv *)taosArrayGet(elements.colArray, 3);
|
||||
|
|
|
@ -198,7 +198,7 @@ int32_t tsTransPullupInterval = 2;
|
|||
int32_t tsMqRebalanceInterval = 2;
|
||||
int32_t tsStreamCheckpointTickInterval = 1;
|
||||
int32_t tsTtlUnit = 86400;
|
||||
int32_t tsTtlPushInterval = 86400;
|
||||
int32_t tsTtlPushInterval = 3600;
|
||||
int32_t tsGrantHBInterval = 60;
|
||||
int32_t tsUptimeInterval = 300; // seconds
|
||||
char tsUdfdResFuncs[512] = ""; // udfd resident funcs that teardown when udfd exits
|
||||
|
|
|
@ -895,6 +895,63 @@ int smlProcess_18784_Test() {
|
|||
return code;
|
||||
}
|
||||
|
||||
int sml_escape_Test() {
|
||||
TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
|
||||
|
||||
TAOS_RES *pRes = taos_query(taos, "create database if not exists db_escape");
|
||||
taos_free_result(pRes);
|
||||
|
||||
pRes = taos_query(taos, "use db_escape");
|
||||
taos_free_result(pRes);
|
||||
|
||||
const char *sql[] = {
|
||||
"d\\,i=\\ s\\k\",dev\"i\\,\\=\\ ce=s\"i\\,\\=\\ dc inode\"i\\,\\=\\ s_used=176059i,total=1076048383523889174i 1661943960000000000",
|
||||
"d\\,i=\\ s\\k\",dev\"i\\,\\=\\ ce=s\"i\\,\\=\\ dc inode\"i\\,\\=\\ s_f\\\\ree=\"\\\"id,= ei\\\\\\f\" 1661943960000000000",
|
||||
};
|
||||
pRes = taos_schemaless_insert(taos, (char **)sql, sizeof(sql) / sizeof(sql[0]), TSDB_SML_LINE_PROTOCOL, 0);
|
||||
printf("%s result:%s, rows:%d\n", __FUNCTION__, taos_errstr(pRes), taos_affected_rows(pRes));
|
||||
int code = taos_errno(pRes);
|
||||
ASSERT(!code);
|
||||
ASSERT(taos_affected_rows(pRes) == 1);
|
||||
taos_free_result(pRes);
|
||||
|
||||
pRes = taos_query(taos, "select * from `d,i= s\\k\"`"); //check stable name
|
||||
ASSERT(pRes);
|
||||
int fieldNum = taos_field_count(pRes);
|
||||
ASSERT(fieldNum == 5);
|
||||
printf("fieldNum:%d\n", fieldNum);
|
||||
|
||||
int numFields = taos_num_fields(pRes);
|
||||
TAOS_FIELD *fields = taos_fetch_fields(pRes);
|
||||
ASSERT(numFields == 5);
|
||||
ASSERT(strncmp(fields[1].name, "inode\"i,= s_used", sizeof("inode\"i,= s_used") - 1) == 0);
|
||||
ASSERT(strncmp(fields[2].name, "total", sizeof("total") - 1) == 0);
|
||||
ASSERT(strncmp(fields[3].name, "inode\"i,= s_f\\\\ree", sizeof("inode\"i,= s_f\\\\ree") - 1) == 0);
|
||||
ASSERT(strncmp(fields[4].name, "dev\"i,= ce", sizeof("dev\"i,= ce") - 1) == 0);
|
||||
|
||||
TAOS_ROW row = NULL;
|
||||
int32_t rowIndex = 0;
|
||||
while ((row = taos_fetch_row(pRes)) != NULL) {
|
||||
int64_t ts = *(int64_t *)row[0];
|
||||
int64_t used = *(int64_t *)row[1];
|
||||
int64_t total = *(int64_t *)row[2];
|
||||
|
||||
if (rowIndex == 0) {
|
||||
ASSERT(ts == 1661943960000);
|
||||
ASSERT(used == 176059);
|
||||
ASSERT(total == 1076048383523889174);
|
||||
ASSERT(strncmp(row[3], "\"id,= ei\\\\f", sizeof("\"id,= ei\\\\f") - 1) == 0);
|
||||
ASSERT(strncmp(row[4], "s\"i,= dc", sizeof("s\"i,= dc") - 1) == 0);
|
||||
|
||||
}
|
||||
rowIndex++;
|
||||
}
|
||||
taos_free_result(pRes);
|
||||
taos_close(taos);
|
||||
|
||||
return code;
|
||||
}
|
||||
|
||||
int sml_19221_Test() {
|
||||
TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
|
||||
|
||||
|
@ -961,6 +1018,55 @@ int sml_ts2164_Test() {
|
|||
return code;
|
||||
}
|
||||
|
||||
|
||||
int sml_ts3116_Test() {
|
||||
TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
|
||||
|
||||
TAOS_RES *pRes =
|
||||
taos_query(taos, "DROP DATABASE IF EXISTS ts3116");
|
||||
taos_free_result(pRes);
|
||||
|
||||
pRes = taos_query(taos, "CREATE DATABASE IF NOT EXISTS ts3116 BUFFER 384 MINROWS 1000 PAGES 256 PRECISION 'ns'");
|
||||
taos_free_result(pRes);
|
||||
|
||||
char *sql = {
|
||||
"meters,location=la,groupid=ca current=11.8,voltage=221",
|
||||
};
|
||||
|
||||
pRes = taos_query(taos, "use ts3116");
|
||||
taos_free_result(pRes);
|
||||
int32_t totalRows = 0;
|
||||
char *tmp = (char *)taosMemoryCalloc(1024, 1);
|
||||
memcpy(tmp, sql, strlen(sql));
|
||||
totalRows = 0;
|
||||
pRes = taos_schemaless_insert_raw(taos, tmp, strlen(tmp), &totalRows, TSDB_SML_LINE_PROTOCOL,
|
||||
TSDB_SML_TIMESTAMP_MILLI_SECONDS);
|
||||
taosMemoryFree(tmp);
|
||||
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
|
||||
int code = taos_errno(pRes);
|
||||
taos_free_result(pRes);
|
||||
|
||||
char *sql1 = {
|
||||
"meters,location=la,groupid=ca\\=3 current=11.8,voltage=221\nmeters,location=la,groupid=ca current=11.8,voltage=221,phase=0.27",
|
||||
};
|
||||
|
||||
pRes = taos_query(taos, "use ts3116");
|
||||
taos_free_result(pRes);
|
||||
|
||||
tmp = (char *)taosMemoryCalloc(1024, 1);
|
||||
memcpy(tmp, sql1, strlen(sql1));
|
||||
totalRows = 0;
|
||||
pRes = taos_schemaless_insert_raw(taos, tmp, strlen(tmp), &totalRows, TSDB_SML_LINE_PROTOCOL,
|
||||
TSDB_SML_TIMESTAMP_MILLI_SECONDS);
|
||||
taosMemoryFree(tmp);
|
||||
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
|
||||
code = taos_errno(pRes);
|
||||
taos_free_result(pRes);
|
||||
taos_close(taos);
|
||||
|
||||
return code;
|
||||
}
|
||||
|
||||
int sml_td22898_Test() {
|
||||
TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0);
|
||||
|
||||
|
@ -1195,6 +1301,10 @@ int main(int argc, char *argv[]) {
|
|||
}
|
||||
|
||||
int ret = 0;
|
||||
ret = sml_escape_Test();
|
||||
ASSERT(!ret);
|
||||
ret = sml_ts3116_Test();
|
||||
ASSERT(!ret);
|
||||
ret = sml_ts2385_Test(); // this test case need config sml table name using ./sml_test config_file
|
||||
ASSERT(!ret);
|
||||
// for(int i = 0; i < sizeof(str)/sizeof(str[0]); i++){
|
||||
|
|
Loading…
Reference in New Issue