diff --git a/include/client/taos.h b/include/client/taos.h index 270b647a77d18fd10e97954e435000799bdd6007..1182a9c2f7d4ff085d04ae72238644314d50e03b 100644 --- a/include/client/taos.h +++ b/include/client/taos.h @@ -198,6 +198,7 @@ DLL_EXPORT const void *taos_get_raw_block(TAOS_RES *res); DLL_EXPORT int taos_load_table_info(TAOS *taos, const char *tableNameList); DLL_EXPORT TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision); +DLL_EXPORT TAOS_RES *taos_schemaless_insert_raw(TAOS* taos, char* lines, int len, int32_t *totalRows, int protocol, int precision); /* --------------------------TMQ INTERFACE------------------------------- */ diff --git a/source/client/src/clientSml.c b/source/client/src/clientSml.c index 5c37822222ece2e83db0d958074d7e36913a35ea..f2739c9ab892a5a895e45ca714dee7d467615023 100644 --- a/source/client/src/clientSml.c +++ b/source/client/src/clientSml.c @@ -28,8 +28,8 @@ #define QUOTE '"' #define SLASH '\\' -#define JUMP_SPACE(sql) \ - while (*sql != '\0') { \ +#define JUMP_SPACE(sql, sqlEnd) \ + while (sql < sqlEnd) { \ if (*sql == SPACE) \ sql++; \ else \ @@ -917,16 +917,17 @@ static int32_t smlParseValue(SSmlKv *pVal, SSmlMsgBuf *msg) { return TSDB_CODE_TSC_INVALID_VALUE; } -static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSmlMsgBuf *msg) { +static int32_t smlParseInfluxString(const char *sql, const char *sqlEnd, SSmlLineInfo *elements, SSmlMsgBuf *msg) { if (!sql) return TSDB_CODE_SML_INVALID_DATA; - JUMP_SPACE(sql) + JUMP_SPACE(sql, sqlEnd) if (*sql == COMMA) return TSDB_CODE_SML_INVALID_DATA; elements->measure = sql; // parse measure - while (*sql != '\0') { + while (sql < sqlEnd) { if ((sql != elements->measure) && IS_SLASH_LETTER(sql)) { - MOVE_FORWARD_ONE(sql, strlen(sql) + 1); + MOVE_FORWARD_ONE(sql, sqlEnd - sql); + sqlEnd--; continue; } if (IS_COMMA(sql)) { @@ -950,7 +951,7 @@ static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSm } else { if (*sql == COMMA) sql++; elements->tags = sql; - while (*sql != '\0') { + while (sql < sqlEnd) { if (IS_SPACE(sql)) { break; } @@ -961,10 +962,10 @@ static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSm elements->measureTagsLen = sql - elements->measure; // parse cols - JUMP_SPACE(sql) + JUMP_SPACE(sql, sqlEnd) elements->cols = sql; bool isInQuote = false; - while (*sql != '\0') { + while (sql < sqlEnd) { if (IS_QUOTE(sql)) { isInQuote = !isInQuote; } @@ -984,10 +985,10 @@ static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSm } // parse timestamp - JUMP_SPACE(sql) + JUMP_SPACE(sql, sqlEnd) elements->timestamp = sql; - while (*sql != '\0') { - if (*sql == SPACE) { + while (sql < sqlEnd) { + if (isspace(*sql)) { break; } sql++; @@ -997,8 +998,8 @@ static int32_t smlParseInfluxString(const char *sql, SSmlLineInfo *elements, SSm return TSDB_CODE_SUCCESS; } -static void smlParseTelnetElement(const char **sql, const char **data, int32_t *len) { - while (**sql != '\0') { +static void smlParseTelnetElement(const char **sql, const char *sqlEnd, const char **data, int32_t *len) { + while (*sql < sqlEnd) { if (**sql != SPACE && !(*data)) { *data = *sql; } else if (**sql == SPACE && *data) { @@ -1009,20 +1010,20 @@ static void smlParseTelnetElement(const char **sql, const char **data, int32_t * } } -static int32_t smlParseTelnetTags(const char *data, SArray *cols, char *childTableName, SHashObj *dumplicateKey, +static int32_t smlParseTelnetTags(const char *data, const char *sqlEnd, SArray *cols, char *childTableName, SHashObj *dumplicateKey, SSmlMsgBuf *msg) { if(!cols) return TSDB_CODE_OUT_OF_MEMORY; const char *sql = data; size_t childTableNameLen = strlen(tsSmlChildTableName); - while (*sql != '\0') { - JUMP_SPACE(sql) + while (sql < sqlEnd) { + JUMP_SPACE(sql, sqlEnd) if (*sql == '\0') break; const char *key = sql; int32_t keyLen = 0; // parse key - while (*sql != '\0') { + while (sql < sqlEnd) { if (*sql == SPACE) { smlBuildInvalidDataMsg(msg, "invalid data", sql); return TSDB_CODE_SML_INVALID_DATA; @@ -1047,7 +1048,7 @@ static int32_t smlParseTelnetTags(const char *data, SArray *cols, char *childTab // parse value const char *value = sql; int32_t valueLen = 0; - while (*sql != '\0') { + while (sql < sqlEnd) { // parse value if (*sql == SPACE) { break; @@ -1092,11 +1093,11 @@ static int32_t smlParseTelnetTags(const char *data, SArray *cols, char *childTab } // format: =[ =] -static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTableInfo *tinfo, SArray *cols) { +static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, const char *sqlEnd, SSmlTableInfo *tinfo, SArray *cols) { if (!sql) return TSDB_CODE_SML_INVALID_DATA; // parse metric - smlParseTelnetElement(&sql, &tinfo->sTableName, &tinfo->sTableNameLen); + smlParseTelnetElement(&sql, sqlEnd, &tinfo->sTableName, &tinfo->sTableNameLen); if (!(tinfo->sTableName) || IS_INVALID_TABLE_LEN(tinfo->sTableNameLen)) { smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql); return TSDB_CODE_TSC_INVALID_TABLE_ID_LENGTH; @@ -1105,7 +1106,7 @@ static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTable // parse timestamp const char *timestamp = NULL; int32_t tLen = 0; - smlParseTelnetElement(&sql, ×tamp, &tLen); + smlParseTelnetElement(&sql, sqlEnd, ×tamp, &tLen); if (!timestamp || tLen == 0) { smlBuildInvalidDataMsg(&info->msgBuf, "invalid timestamp", sql); return TSDB_CODE_SML_INVALID_DATA; @@ -1120,7 +1121,7 @@ static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTable // parse value const char *value = NULL; int32_t valueLen = 0; - smlParseTelnetElement(&sql, &value, &valueLen); + smlParseTelnetElement(&sql, sqlEnd, &value, &valueLen); if (!value || valueLen == 0) { smlBuildInvalidDataMsg(&info->msgBuf, "invalid value", sql); return TSDB_CODE_TSC_INVALID_VALUE; @@ -1138,7 +1139,7 @@ static int32_t smlParseTelnetString(SSmlHandle *info, const char *sql, SSmlTable } // parse tags - ret = smlParseTelnetTags(sql, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf); + ret = smlParseTelnetTags(sql, sqlEnd, tinfo->tags, tinfo->childTableName, info->dumplicateKey, &info->msgBuf); if (ret != TSDB_CODE_SUCCESS) { smlBuildInvalidDataMsg(&info->msgBuf, "invalid data", sql); return ret; @@ -2073,11 +2074,11 @@ static int32_t smlParseJSONString(SSmlHandle *info, cJSON *root, SSmlTableInfo * } /************* TSDB_SML_JSON_PROTOCOL function end **************/ -static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql) { +static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql, const int len) { SSmlLineInfo elements = {0}; uDebug("SML:0x%" PRIx64 " smlParseInfluxLine sql:%s, hello", info->id, sql); - int ret = smlParseInfluxString(sql, &elements, &info->msgBuf); + int ret = smlParseInfluxString(sql, sql + len, &elements, &info->msgBuf); if (ret != TSDB_CODE_SUCCESS) { uError("SML:0x%" PRIx64 " smlParseInfluxLine failed", info->id); return ret; @@ -2184,7 +2185,7 @@ static int32_t smlParseInfluxLine(SSmlHandle *info, const char *sql) { return TSDB_CODE_SUCCESS; } -static int32_t smlParseTelnetLine(SSmlHandle *info, void *data) { +static int32_t smlParseTelnetLine(SSmlHandle *info, void *data, const int len) { int ret = TSDB_CODE_SUCCESS; SSmlTableInfo *tinfo = smlBuildTableInfo(); if (!tinfo) { @@ -2198,7 +2199,7 @@ static int32_t smlParseTelnetLine(SSmlHandle *info, void *data) { } if (info->protocol == TSDB_SML_TELNET_PROTOCOL) { - ret = smlParseTelnetString(info, (const char *)data, tinfo, cols); + ret = smlParseTelnetString(info, (const char *)data, (char*)data + len, tinfo, cols); } else if (info->protocol == TSDB_SML_JSON_PROTOCOL) { ret = smlParseJSONString(info, (cJSON *)data, tinfo, cols); } else { @@ -2289,7 +2290,7 @@ static int32_t smlParseJSON(SSmlHandle *info, char *payload) { for (int32_t i = 0; i < payloadNum; ++i) { cJSON *dataPoint = (payloadNum == 1 && cJSON_IsObject(root)) ? root : cJSON_GetArrayItem(root, i); - ret = smlParseTelnetLine(info, dataPoint); + ret = smlParseTelnetLine(info, dataPoint, -1); if (ret != TSDB_CODE_SUCCESS) { uError("SML:0x%" PRIx64 " Invalid JSON Payload", info->id); goto end; @@ -2378,10 +2379,14 @@ static void smlPrintStatisticInfo(SSmlHandle *info) { info->cost.endTime - info->cost.insertRpcTime, info->cost.endTime - info->cost.parseTime); } -static int32_t smlParseLine(SSmlHandle *info, char *lines[], int numLines) { +static int32_t smlParseLine(SSmlHandle *info, char *lines[], char* rawLine, char* rawLineEnd, int numLines) { int32_t code = TSDB_CODE_SUCCESS; if (info->protocol == TSDB_SML_JSON_PROTOCOL) { - code = smlParseJSON(info, *lines); + if(lines){ + code = smlParseJSON(info, *lines); + }else if(rawLine){ + code = smlParseJSON(info, rawLine); + } if (code != TSDB_CODE_SUCCESS) { uError("SML:0x%" PRIx64 " smlParseJSON failed:%s", info->id, *lines); return code; @@ -2390,28 +2395,46 @@ static int32_t smlParseLine(SSmlHandle *info, char *lines[], int numLines) { } for (int32_t i = 0; i < numLines; ++i) { + char *tmp = NULL; + int len = 0; + if(lines){ + tmp = lines[i]; + len = strlen(tmp); + }else if(rawLine){ + tmp = rawLine; + while(rawLine < rawLineEnd){ + if(*(rawLine++) == '\n'){ + break; + } + len++; + } + if(info->protocol == TSDB_SML_LINE_PROTOCOL && tmp[0] == '#'){ // this line is comment + continue; + } + } + if (info->protocol == TSDB_SML_LINE_PROTOCOL) { - code = smlParseInfluxLine(info, lines[i]); + code = smlParseInfluxLine(info, tmp, len); } else if (info->protocol == TSDB_SML_TELNET_PROTOCOL) { - code = smlParseTelnetLine(info, lines[i]); + code = smlParseTelnetLine(info, tmp, len); } else { ASSERT(0); } if (code != TSDB_CODE_SUCCESS) { - uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, lines[i]); + uError("SML:0x%" PRIx64 " smlParseLine failed. line %d : %s", info->id, i, tmp); return code; } } return code; } -static int smlProcess(SSmlHandle *info, char *lines[], int numLines) { +static int smlProcess(SSmlHandle *info, char *lines[], char* rawLine, char* rawLineEnd, int numLines) { int32_t code = TSDB_CODE_SUCCESS; int32_t retryNum = 0; info->cost.parseTime = taosGetTimestampUs(); - code = smlParseLine(info, lines, numLines); + code = smlParseLine(info, lines, rawLine, rawLineEnd, numLines); if (code != 0) { uError("SML:0x%" PRIx64 " smlParseLine error : %s", info->id, tstrerror(code)); return code; @@ -2504,39 +2527,8 @@ static void smlInsertCallback(void *param, void *res, int32_t code) { smlDestroyInfo(info); } -/** - * taos_schemaless_insert() parse and insert data points into database according to - * different protocol. - * - * @param $lines input array may contain multiple lines, each line indicates a data point. - * If protocol=2 is used input array should contain single JSON - * string(e.g. char *lines[] = {"$JSON_string"}). If need to insert - * multiple data points in JSON format, should include them in $JSON_string - * as a JSON array. - * @param $numLines indicates how many data points in $lines. - * If protocol = 2 is used this param will be ignored as $lines should - * contain single JSON string. - * @param $protocol indicates which protocol to use for parsing: - * 0 - influxDB line protocol - * 1 - OpenTSDB telnet line protocol - * 2 - OpenTSDB JSON format protocol - * @return return zero for successful insertion. Otherwise return none-zero error code of - * failure reason. - * - */ - -TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) { - if (NULL == taos) { - terrno = TSDB_CODE_TSC_DISCONNECTED; - return NULL; - } - - SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT); - if (!request) { - uError("SML:taos_schemaless_insert error request is null"); - return NULL; - } +TAOS_RES *taos_schemaless_insert_inner(SRequestObj *request, char *lines[], char *rawLine, char *rawLineEnd, int numLines, int protocol, int precision) { int batchs = 0; STscObj *pTscObj = request->pTscObj; @@ -2560,12 +2552,6 @@ TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int pr goto end; } - if (!lines) { - request->code = TSDB_CODE_SML_INVALID_DATA; - smlBuildInvalidDataMsg(&msg, "lines is null", NULL); - goto end; - } - if (protocol < TSDB_SML_LINE_PROTOCOL || protocol > TSDB_SML_JSON_PROTOCOL) { request->code = TSDB_CODE_SML_INVALID_PROTOCOL_TYPE; smlBuildInvalidDataMsg(&msg, "protocol invalidate", NULL); @@ -2616,15 +2602,28 @@ TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int pr info->affectedRows = perBatch; info->pRequest->body.queryFp = smlInsertCallback; info->pRequest->body.param = info; - int32_t code = smlProcess(info, lines, perBatch); - lines += perBatch; + int32_t code = smlProcess(info, lines, rawLine, rawLineEnd, perBatch); + if(lines){ + lines += perBatch; + } + if(rawLine){ + int num = 0; + while(rawLine < rawLineEnd){ + if(*(rawLine++) == '\n'){ + num++; + } + if(num == perBatch){ + break; + } + } + } if (code != TSDB_CODE_SUCCESS) { info->pRequest->body.queryFp(info, req, code); } } tsem_wait(¶ms.sem); -end: + end: taosThreadSpinDestroy(¶ms.lock); tsem_destroy(¶ms.sem); // ((STscObj *)taos)->schemalessType = 0; @@ -2632,3 +2631,80 @@ end: uDebug("resultend:%s", request->msgBuf); return (TAOS_RES *)request; } + +/** + * taos_schemaless_insert() parse and insert data points into database according to + * different protocol. + * + * @param $lines input array may contain multiple lines, each line indicates a data point. + * If protocol=2 is used input array should contain single JSON + * string(e.g. char *lines[] = {"$JSON_string"}). If need to insert + * multiple data points in JSON format, should include them in $JSON_string + * as a JSON array. + * @param $numLines indicates how many data points in $lines. + * If protocol = 2 is used this param will be ignored as $lines should + * contain single JSON string. + * @param $protocol indicates which protocol to use for parsing: + * 0 - influxDB line protocol + * 1 - OpenTSDB telnet line protocol + * 2 - OpenTSDB JSON format protocol + * @return return zero for successful insertion. Otherwise return none-zero error code of + * failure reason. + * + */ + +TAOS_RES *taos_schemaless_insert(TAOS *taos, char *lines[], int numLines, int protocol, int precision) { + if (NULL == taos) { + terrno = TSDB_CODE_TSC_DISCONNECTED; + return NULL; + } + + SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT); + if (!request) { + uError("SML:taos_schemaless_insert error request is null"); + return NULL; + } + + if (!lines) { + SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf}; + request->code = TSDB_CODE_SML_INVALID_DATA; + smlBuildInvalidDataMsg(&msg, "lines is null", NULL); + return (TAOS_RES *)request; + } + + return taos_schemaless_insert_inner(request, lines, NULL, NULL, numLines, protocol, precision); +} + +TAOS_RES *taos_schemaless_insert_raw(TAOS* taos, char* lines, int len, int32_t *totalRows, int protocol, int precision){ + if (NULL == taos) { + terrno = TSDB_CODE_TSC_DISCONNECTED; + return NULL; + } + + SRequestObj *request = (SRequestObj *)createRequest(*(int64_t *)taos, TSDB_SQL_INSERT); + if (!request) { + uError("SML:taos_schemaless_insert error request is null"); + return NULL; + } + + if (!lines || len <= 0) { + SSmlMsgBuf msg = {ERROR_MSG_BUF_DEFAULT_SIZE, request->msgBuf}; + request->code = TSDB_CODE_SML_INVALID_DATA; + smlBuildInvalidDataMsg(&msg, "lines is null", NULL); + return (TAOS_RES *)request; + } + + int numLines = 0; + *totalRows = 0; + char *tmp = lines; + for(int i = 0; i < len; i++){ + if(lines[i] == '\n' || i == len - 1){ + numLines++; + if(tmp[0] != '#' || protocol != TSDB_SML_LINE_PROTOCOL){ //ignore comment + (*totalRows)++; + } + tmp = lines + i + 1; + } + } + return taos_schemaless_insert_inner(request, NULL, lines, lines + len, numLines, protocol, precision); +} diff --git a/source/client/test/smlTest.cpp b/source/client/test/smlTest.cpp index ab4442eca7abda74f851aedf2926cdb3f64ec7a7..54778e87a788190c55fafc0e4d0db732677d795d 100644 --- a/source/client/test/smlTest.cpp +++ b/source/client/test/smlTest.cpp @@ -44,7 +44,7 @@ TEST(testCase, smlParseInfluxString_Test) { 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); - int ret = smlParseInfluxString(sql, &elements, &msgBuf); + int ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf); ASSERT_EQ(ret, 0); ASSERT_EQ(elements.measure, sql); ASSERT_EQ(elements.measureLen, strlen(",st")); @@ -63,14 +63,14 @@ TEST(testCase, smlParseInfluxString_Test) { 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)); - ret = smlParseInfluxString(sql, &elements, &msgBuf); + ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf); ASSERT_NE(ret, 0); // case 3 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)); - ret = smlParseInfluxString(sql, &elements, &msgBuf); + ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf); ASSERT_EQ(ret, 0); ASSERT_EQ(elements.cols, sql + elements.measureTagsLen + 1); ASSERT_EQ(elements.colsLen, strlen("t1=3,t2=4,t3=t3")); @@ -79,7 +79,7 @@ TEST(testCase, smlParseInfluxString_Test) { 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)); - ret = smlParseInfluxString(sql, &elements, &msgBuf); + ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf); ASSERT_EQ(ret, 0); ASSERT_EQ(elements.measure, sql); ASSERT_EQ(elements.measureLen, strlen("st")); @@ -98,7 +98,7 @@ TEST(testCase, smlParseInfluxString_Test) { 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)); - ret = smlParseInfluxString(sql, &elements, &msgBuf); + ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf); ASSERT_EQ(ret, 0); ASSERT_EQ(elements.measure, sql + 1); ASSERT_EQ(elements.measureLen, strlen("st")); @@ -116,21 +116,21 @@ TEST(testCase, smlParseInfluxString_Test) { tmp = " st c1=3i64,c3=\"passit hello,c1=2\",c2=false,c4=4f64 "; memcpy(sql, tmp, strlen(tmp) + 1); memset(&elements, 0, sizeof(SSmlLineInfo)); - ret = smlParseInfluxString(sql, &elements, &msgBuf); + ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf); ASSERT_EQ(ret, 0); // case 7 tmp = " st , "; memcpy(sql, tmp, strlen(tmp) + 1); memset(&elements, 0, sizeof(SSmlLineInfo)); - ret = smlParseInfluxString(sql, &elements, &msgBuf); + ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf); ASSERT_EQ(ret, 0); // case 8 false tmp = ", st , "; memcpy(sql, tmp, strlen(tmp) + 1); memset(&elements, 0, sizeof(SSmlLineInfo)); - ret = smlParseInfluxString(sql, &elements, &msgBuf); + ret = smlParseInfluxString(sql, sql + strlen(sql), &elements, &msgBuf); ASSERT_NE(ret, 0); taosMemoryFree(sql); } @@ -542,7 +542,7 @@ TEST(testCase, smlParseTelnetLine_error_Test) { "sys.procs.running 1479496100 42 host= web01", }; for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) { - int ret = smlParseTelnetLine(info, (void *)sql[i]); + int ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i])); ASSERT_NE(ret, 0); } @@ -561,7 +561,7 @@ TEST(testCase, smlParseTelnetLine_diff_type_Test) { int ret = TSDB_CODE_SUCCESS; for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) { - ret = smlParseTelnetLine(info, (void *)sql[i]); + ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i])); if (ret != TSDB_CODE_SUCCESS) break; } ASSERT_NE(ret, 0); @@ -617,7 +617,7 @@ TEST(testCase, smlParseTelnetLine_json_error_Test) { int ret = TSDB_CODE_SUCCESS; for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) { - ret = smlParseTelnetLine(info, (void *)sql[i]); + ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i])); ASSERT_NE(ret, 0); } @@ -653,7 +653,7 @@ TEST(testCase, smlParseTelnetLine_diff_json_type1_Test) { int ret = TSDB_CODE_SUCCESS; for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) { - ret = smlParseTelnetLine(info, (void *)sql[i]); + ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i])); if (ret != TSDB_CODE_SUCCESS) break; } ASSERT_NE(ret, 0); @@ -688,7 +688,7 @@ TEST(testCase, smlParseTelnetLine_diff_json_type2_Test) { }; int ret = TSDB_CODE_SUCCESS; for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) { - ret = smlParseTelnetLine(info, (void *)sql[i]); + ret = smlParseTelnetLine(info, (void *)sql[i], strlen(sql[i])); if (ret != TSDB_CODE_SUCCESS) break; } ASSERT_NE(ret, 0); @@ -1002,7 +1002,7 @@ TEST(testCase, sml_col_4096_Test) { int ret = TSDB_CODE_SUCCESS; for (int i = 0; i < sizeof(sql) / sizeof(sql[0]); i++) { - ret = smlParseInfluxLine(info, sql[i]); + ret = smlParseInfluxLine(info, sql[i], strlen(sql[i])); if (ret != TSDB_CODE_SUCCESS) break; } ASSERT_NE(ret, 0); diff --git a/utils/test/c/sml_test.c b/utils/test/c/sml_test.c index b6d8d75ba01c8930787f8fad434348ce9ee571d1..f7fd59cd7ba4a6b5208acce53b2d96dd10bb7586 100644 --- a/utils/test/c/sml_test.c +++ b/utils/test/c/sml_test.c @@ -63,6 +63,7 @@ int smlProcess_influx_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); return code; } @@ -86,6 +87,8 @@ int smlProcess_telnet_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -125,6 +128,8 @@ int smlProcess_json1_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -165,6 +170,8 @@ int smlProcess_json2_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -233,6 +240,8 @@ int smlProcess_json3_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -292,6 +301,8 @@ int smlProcess_json4_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -313,6 +324,8 @@ int sml_TD15662_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -333,6 +346,8 @@ int sml_TD15742_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -362,6 +377,8 @@ int sml_16384_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -781,6 +798,8 @@ int sml_oom_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -825,6 +844,8 @@ int sml_16368_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -862,6 +883,8 @@ int sml_dup_time_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -1068,6 +1091,8 @@ int sml_16960_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); int code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); + return code; } @@ -1097,6 +1122,7 @@ int sml_add_tag_col_Test() { printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); code = taos_errno(pRes); taos_free_result(pRes); + taos_close(taos); return code; } @@ -1151,6 +1177,36 @@ int smlProcess_18784_Test() { rowIndex++; } taos_free_result(pRes); + taos_close(taos); + + return code; +} + +int sml_19221_Test() { + TAOS *taos = taos_connect("localhost", "root", "taosdata", NULL, 0); + + TAOS_RES *pRes = taos_query(taos, "create database if not exists sml_db schemaless 1"); + taos_free_result(pRes); + + const char *sql[] = { + "qelhxo,id=pnnqhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833639000000\nqelhxo,id=pnnhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833639000000\n#comment\nqelhxo,id=pnqhsa,t0=t,t1=127i8 c11=L\"ncharColValue\",c0=t,c1=127i8 1626006833639000000", + }; + + pRes = taos_query(taos, "use sml_db"); + taos_free_result(pRes); + + char* tmp = (char*)taosMemoryCalloc(1024, 1); + memcpy(tmp, sql[0], strlen(sql[0])); + *(char*)(tmp+44) = 0; + int32_t totalRows = 0; + pRes = taos_schemaless_insert_raw(taos, tmp, strlen(sql[0]), &totalRows, TSDB_SML_LINE_PROTOCOL, TSDB_SML_TIMESTAMP_NANO_SECONDS); + + ASSERT(totalRows == 3); + printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes)); + int code = taos_errno(pRes); + taos_free_result(pRes); + taos_close(taos); + taosMemoryFree(tmp); return code; } @@ -1187,5 +1243,7 @@ int main(int argc, char *argv[]) { ASSERT(!ret); ret = smlProcess_18784_Test(); ASSERT(!ret); + ret = sml_19221_Test(); + ASSERT(!ret); return ret; }