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

fix:<TD-18784> data lost in schemaless if timestamp is same

This commit is contained in:
wangmm0220 2022-09-01 13:49:05 +08:00
parent f778854302
commit d1bdea6917
3 changed files with 117 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -2450,9 +2450,11 @@ static void smlInsertCallback(void *param, void *res, int32_t code) {
uDebug("SML:0x%" PRIx64 " result. code:%d, msg:%s", info->id, pRequest->code, pRequest->msgBuf);
// lock
taosThreadSpinLock(&info->params->lock);
info->params->request->body.resInfo.numOfRows += rows;
if (code != TSDB_CODE_SUCCESS) {
info->params->request->code = code;
info->params->request->body.resInfo.numOfRows += rows;
}else{
info->params->request->body.resInfo.numOfRows += info->affectedRows;
}
taosThreadSpinUnlock(&info->params->lock);
// unlock

13
source/libs/parser/src/parInsert.c
View File

@ -2253,7 +2253,7 @@ static int32_t smlBoundColumnData(SArray* cols, SParsedDataColInfo* pColList, SS
SToken sToken = {.n = kv->keyLen, .z = (char*)kv->key};
col_id_t t = lastColIdx + 1;
col_id_t index = ((t == 0 && !isTag) ? 0 : findCol(&sToken, t, nCols, pSchema));
uDebug("SML, index:%d, t:%d, ncols:%d, kv->name:%s", index, t, nCols, kv->key);
uDebug("SML, index:%d, t:%d, ncols:%d", index, t, nCols);
if (index < 0 && t > 0) {
index = findCol(&sToken, 0, t, pSchema);
isOrdered = false;
@ -2474,9 +2474,7 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols
if (p) kv = *p;
}
if (!kv || kv->length == 0) {
MemRowAppend(&pBuf, NULL, 0, &param);
} else {
if (kv){
int32_t colLen = kv->length;
if (pColSchema->type == TSDB_DATA_TYPE_TIMESTAMP) {
// uError("SML:data before:%" PRId64 ", precision:%d", kv->i, pTableMeta->tableInfo.precision);
@ -2489,6 +2487,8 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols
} else {
MemRowAppend(&pBuf, &(kv->value), colLen, &param);
}
}else{
pBuilder->hasNone = true;
}
if (PRIMARYKEY_TIMESTAMP_COL_ID == pColSchema->colId) {
@ -2497,11 +2497,6 @@ int32_t smlBindData(void* handle, SArray* tags, SArray* colsSchema, SArray* cols
}
}
// set the null value for the columns that do not assign values
if ((spd->numOfBound < spd->numOfCols) && TD_IS_TP_ROW(row)) {
pBuilder->hasNone = true;
}
tdSRowEnd(pBuilder);
pDataBlock->size += extendedRowSize;
}

123
tests/test/c/sml_test.c
View File

@ -63,6 +63,46 @@ int smlProcess_influx_Test() {
printf("%s result:%s\n", __FUNCTION__, taos_errstr(pRes));
int code = taos_errno(pRes);
taos_free_result(pRes);
// case 1
pRes = taos_query(taos, "select * from t_91e0b182be80332b5c530cbf872f760e");
ASSERT(pRes);
int fieldNum = taos_field_count(pRes);
ASSERT(fieldNum == 11);
printf("fieldNum:%d\n", fieldNum);
TAOS_ROW row = NULL;
int32_t rowIndex = 0;
while((row = taos_fetch_row(pRes)) != NULL) {
int64_t ts = *(int64_t*)row[0];
double load_capacity = *(double*)row[1];
double fuel_capacity = *(double*)row[2];
double nominal_fuel_consumption = *(double*)row[3];
double latitude = *(double*)row[4];
double longitude = *(double*)row[5];
double elevation = *(double*)row[6];
double velocity = *(double*)row[7];
double heading = *(double*)row[8];
double grade = *(double*)row[9];
double fuel_consumption = *(double*)row[10];
if(rowIndex == 0){
ASSERT(ts ==1451606407000);
// ASSERT_EQ(load_capacity, 2000);
// ASSERT_EQ(fuel_capacity, 200);
// ASSERT_EQ(nominal_fuel_consumption, 15);
// ASSERT_EQ(latitude, 24.5208);
// ASSERT_EQ(longitude, 28.09377);
// ASSERT_EQ(elevation, 428);
// ASSERT_EQ(velocity, 0);
// ASSERT_EQ(heading, 304);
// ASSERT_EQ(grade, 0);
// ASSERT_EQ(fuel_consumption, 25);
}else{
// ASSERT(0);
}
rowIndex++;
}
taos_free_result(pRes);
return code;
}
@ -1100,34 +1140,91 @@ int sml_add_tag_col_Test() {
return code;
}
int smlProcess_18784_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);
pRes = taos_query(taos, "use sml_db");
taos_free_result(pRes);
const char *sql[] = {
"disk,device=sdc inodes_used=176059i,total=1081101176832i 1661943960000000000",
"disk,device=sdc inodes_free=66932805i 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) == 2);
taos_free_result(pRes);
pRes = taos_query(taos, "select * from disk");
ASSERT(pRes);
int fieldNum = taos_field_count(pRes);
ASSERT(fieldNum == 5);
printf("fieldNum:%d\n", fieldNum);
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];
int64_t freed = *(int64_t*)row[3];
if(rowIndex == 0){
ASSERT(ts == 1661943960000);
ASSERT(used == 176059);
ASSERT(total == 1081101176832);
ASSERT(freed == 66932805);
// ASSERT_EQ(latitude, 24.5208);
// ASSERT_EQ(longitude, 28.09377);
// ASSERT_EQ(elevation, 428);
// ASSERT_EQ(velocity, 0);
// ASSERT_EQ(heading, 304);
// ASSERT_EQ(grade, 0);
// ASSERT_EQ(fuel_consumption, 25);
}else{
// ASSERT(0);
}
rowIndex++;
}
taos_free_result(pRes);
return code;
}
int main(int argc, char *argv[]) {
int ret = 0;
ret = smlProcess_influx_Test();
if(ret) return ret;
ASSERT(!ret);
ret = smlProcess_telnet_Test();
if(ret) return ret;
ASSERT(!ret);
ret = smlProcess_json1_Test();
if(ret) return ret;
ASSERT(!ret);
ret = smlProcess_json2_Test();
if(ret) return ret;
ASSERT(!ret);
ret = smlProcess_json3_Test();
if(ret) return ret;
ASSERT(!ret);
ret = smlProcess_json4_Test();
if(ret) return ret;
ASSERT(!ret);
ret = sml_TD15662_Test();
if(ret) return ret;
ASSERT(!ret);
ret = sml_TD15742_Test();
if(ret) return ret;
ASSERT(!ret);
ret = sml_16384_Test();
if(ret) return ret;
ASSERT(!ret);
ret = sml_oom_Test();
if(ret) return ret;
ASSERT(!ret);
ret = sml_16368_Test();
if(ret) return ret;
ASSERT(!ret);
ret = sml_dup_time_Test();
if(ret) return ret;
ASSERT(!ret);
ret = sml_16960_Test();
if(ret) return ret;
ASSERT(!ret);
ret = sml_add_tag_col_Test();
ASSERT(!ret);
ret = smlProcess_18784_Test();
ASSERT(!ret);
return ret;
}