未验证 提交 833c94b8 编写于 作者: sangshuduo's avatar sangshuduo 提交者: GitHub

[TD-5067]<fix>: taosdemo stmt use sample data (#6760)

上级 1d25f5f6
...@@ -3205,13 +3205,6 @@ static int readTagFromCsvFileToMem(SSuperTable * superTblInfo) { ...@@ -3205,13 +3205,6 @@ static int readTagFromCsvFileToMem(SSuperTable * superTblInfo) {
return 0; return 0;
} }
#if 0
int readSampleFromJsonFileToMem(SSuperTable * superTblInfo) {
// TODO
return 0;
}
#endif
/* /*
Read 10000 lines at most. If more than 10000 lines, continue to read after using Read 10000 lines at most. If more than 10000 lines, continue to read after using
*/ */
...@@ -5327,7 +5320,7 @@ static int64_t generateInterlaceDataWithoutStb( ...@@ -5327,7 +5320,7 @@ static int64_t generateInterlaceDataWithoutStb(
#if STMT_IFACE_ENABLED == 1 #if STMT_IFACE_ENABLED == 1
static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
char *dataType, int32_t dataLen, char **ptr) char *dataType, int32_t dataLen, char **ptr, char *value)
{ {
if (0 == strncasecmp(dataType, if (0 == strncasecmp(dataType,
"BINARY", strlen("BINARY"))) { "BINARY", strlen("BINARY"))) {
...@@ -5337,12 +5330,18 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, ...@@ -5337,12 +5330,18 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
return -1; return -1;
} }
char *bind_binary = (char *)*ptr; char *bind_binary = (char *)*ptr;
rand_string(bind_binary, dataLen);
bind->buffer_type = TSDB_DATA_TYPE_BINARY; bind->buffer_type = TSDB_DATA_TYPE_BINARY;
if (value) {
strncpy(bind_binary, value, strlen(value));
bind->buffer_length = strlen(bind_binary);
} else {
rand_string(bind_binary, dataLen);
bind->buffer_length = dataLen; bind->buffer_length = dataLen;
bind->buffer = bind_binary; }
bind->length = &bind->buffer_length; bind->length = &bind->buffer_length;
bind->buffer = bind_binary;
bind->is_null = NULL; bind->is_null = NULL;
*ptr += bind->buffer_length; *ptr += bind->buffer_length;
...@@ -5354,9 +5353,14 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, ...@@ -5354,9 +5353,14 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
return -1; return -1;
} }
char *bind_nchar = (char *)*ptr; char *bind_nchar = (char *)*ptr;
rand_string(bind_nchar, dataLen);
bind->buffer_type = TSDB_DATA_TYPE_NCHAR; bind->buffer_type = TSDB_DATA_TYPE_NCHAR;
if (value) {
strncpy(bind_nchar, value, strlen(value));
} else {
rand_string(bind_nchar, dataLen);
}
bind->buffer_length = strlen(bind_nchar); bind->buffer_length = strlen(bind_nchar);
bind->buffer = bind_nchar; bind->buffer = bind_nchar;
bind->length = &bind->buffer_length; bind->length = &bind->buffer_length;
...@@ -5367,7 +5371,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, ...@@ -5367,7 +5371,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
"INT", strlen("INT"))) { "INT", strlen("INT"))) {
int32_t *bind_int = (int32_t *)*ptr; int32_t *bind_int = (int32_t *)*ptr;
if (value) {
*bind_int = atoi(value);
} else {
*bind_int = rand_int(); *bind_int = rand_int();
}
bind->buffer_type = TSDB_DATA_TYPE_INT; bind->buffer_type = TSDB_DATA_TYPE_INT;
bind->buffer_length = sizeof(int32_t); bind->buffer_length = sizeof(int32_t);
bind->buffer = bind_int; bind->buffer = bind_int;
...@@ -5379,7 +5387,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, ...@@ -5379,7 +5387,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
"BIGINT", strlen("BIGINT"))) { "BIGINT", strlen("BIGINT"))) {
int64_t *bind_bigint = (int64_t *)*ptr; int64_t *bind_bigint = (int64_t *)*ptr;
if (value) {
*bind_bigint = atoll(value);
} else {
*bind_bigint = rand_bigint(); *bind_bigint = rand_bigint();
}
bind->buffer_type = TSDB_DATA_TYPE_BIGINT; bind->buffer_type = TSDB_DATA_TYPE_BIGINT;
bind->buffer_length = sizeof(int64_t); bind->buffer_length = sizeof(int64_t);
bind->buffer = bind_bigint; bind->buffer = bind_bigint;
...@@ -5391,7 +5403,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, ...@@ -5391,7 +5403,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
"FLOAT", strlen("FLOAT"))) { "FLOAT", strlen("FLOAT"))) {
float *bind_float = (float *) *ptr; float *bind_float = (float *) *ptr;
if (value) {
*bind_float = (float)atof(value);
} else {
*bind_float = rand_float(); *bind_float = rand_float();
}
bind->buffer_type = TSDB_DATA_TYPE_FLOAT; bind->buffer_type = TSDB_DATA_TYPE_FLOAT;
bind->buffer_length = sizeof(float); bind->buffer_length = sizeof(float);
bind->buffer = bind_float; bind->buffer = bind_float;
...@@ -5403,7 +5419,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, ...@@ -5403,7 +5419,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
"DOUBLE", strlen("DOUBLE"))) { "DOUBLE", strlen("DOUBLE"))) {
double *bind_double = (double *)*ptr; double *bind_double = (double *)*ptr;
if (value) {
*bind_double = atof(value);
} else {
*bind_double = rand_double(); *bind_double = rand_double();
}
bind->buffer_type = TSDB_DATA_TYPE_DOUBLE; bind->buffer_type = TSDB_DATA_TYPE_DOUBLE;
bind->buffer_length = sizeof(double); bind->buffer_length = sizeof(double);
bind->buffer = bind_double; bind->buffer = bind_double;
...@@ -5415,7 +5435,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, ...@@ -5415,7 +5435,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
"SMALLINT", strlen("SMALLINT"))) { "SMALLINT", strlen("SMALLINT"))) {
int16_t *bind_smallint = (int16_t *)*ptr; int16_t *bind_smallint = (int16_t *)*ptr;
if (value) {
*bind_smallint = (int16_t)atoi(value);
} else {
*bind_smallint = rand_smallint(); *bind_smallint = rand_smallint();
}
bind->buffer_type = TSDB_DATA_TYPE_SMALLINT; bind->buffer_type = TSDB_DATA_TYPE_SMALLINT;
bind->buffer_length = sizeof(int16_t); bind->buffer_length = sizeof(int16_t);
bind->buffer = bind_smallint; bind->buffer = bind_smallint;
...@@ -5427,7 +5451,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, ...@@ -5427,7 +5451,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
"TINYINT", strlen("TINYINT"))) { "TINYINT", strlen("TINYINT"))) {
int8_t *bind_tinyint = (int8_t *)*ptr; int8_t *bind_tinyint = (int8_t *)*ptr;
if (value) {
*bind_tinyint = (int8_t)atoi(value);
} else {
*bind_tinyint = rand_tinyint(); *bind_tinyint = rand_tinyint();
}
bind->buffer_type = TSDB_DATA_TYPE_TINYINT; bind->buffer_type = TSDB_DATA_TYPE_TINYINT;
bind->buffer_length = sizeof(int8_t); bind->buffer_length = sizeof(int8_t);
bind->buffer = bind_tinyint; bind->buffer = bind_tinyint;
...@@ -5450,7 +5478,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind, ...@@ -5450,7 +5478,11 @@ static int32_t prepareStmtBindArrayByType(TAOS_BIND *bind,
"TIMESTAMP", strlen("TIMESTAMP"))) { "TIMESTAMP", strlen("TIMESTAMP"))) {
int64_t *bind_ts2 = (int64_t *) *ptr; int64_t *bind_ts2 = (int64_t *) *ptr;
if (value) {
*bind_ts2 = atoll(value);
} else {
*bind_ts2 = rand_bigint(); *bind_ts2 = rand_bigint();
}
bind->buffer_type = TSDB_DATA_TYPE_TIMESTAMP; bind->buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
bind->buffer_length = sizeof(int64_t); bind->buffer_length = sizeof(int64_t);
bind->buffer = bind_ts2; bind->buffer = bind_ts2;
...@@ -5516,12 +5548,13 @@ static int32_t prepareStmtWithoutStb( ...@@ -5516,12 +5548,13 @@ static int32_t prepareStmtWithoutStb(
ptr += bind->buffer_length; ptr += bind->buffer_length;
for (int i = 0; i < g_args.num_of_CPR; i ++) { for (int i = 0; i < g_args.num_of_CPR; i ++) {
bind = (TAOS_BIND *)((char *)bindArray + (sizeof(TAOS_BIND) * (i + 1))); bind = (TAOS_BIND *)((char *)bindArray
+ (sizeof(TAOS_BIND) * (i + 1)));
if ( -1 == prepareStmtBindArrayByType( if ( -1 == prepareStmtBindArrayByType(
bind, bind,
data_type[i], data_type[i],
g_args.len_of_binary, g_args.len_of_binary,
&ptr)) { &ptr, NULL)) {
return -1; return -1;
} }
} }
...@@ -5540,12 +5573,14 @@ static int32_t prepareStmtWithoutStb( ...@@ -5540,12 +5573,14 @@ static int32_t prepareStmtWithoutStb(
return k; return k;
} }
static int32_t prepareStbStmt(SSuperTable *stbInfo, static int32_t prepareStbStmt(
SSuperTable *stbInfo,
TAOS_STMT *stmt, TAOS_STMT *stmt,
char *tableName, uint32_t batch, char *tableName, uint32_t batch,
uint64_t insertRows, uint64_t insertRows,
uint64_t recordFrom, uint64_t recordFrom,
int64_t startTime, char *buffer) int64_t startTime,
int64_t *pSamplePos)
{ {
int ret = taos_stmt_set_tbname(stmt, tableName); int ret = taos_stmt_set_tbname(stmt, tableName);
if (ret != 0) { if (ret != 0) {
...@@ -5556,16 +5591,24 @@ static int32_t prepareStbStmt(SSuperTable *stbInfo, ...@@ -5556,16 +5591,24 @@ static int32_t prepareStbStmt(SSuperTable *stbInfo,
char *bindArray = malloc(sizeof(TAOS_BIND) * (stbInfo->columnCount + 1)); char *bindArray = malloc(sizeof(TAOS_BIND) * (stbInfo->columnCount + 1));
if (bindArray == NULL) { if (bindArray == NULL) {
errorPrint("Failed to allocate %d bind params\n", errorPrint("%s() LN%d, Failed to allocate %d bind params\n",
(stbInfo->columnCount + 1)); __func__, __LINE__, (stbInfo->columnCount + 1));
return -1; return -1;
} }
bool tsRand; bool sourceRand;
if (0 == strncasecmp(stbInfo->dataSource, "rand", strlen("rand"))) { if (0 == strncasecmp(stbInfo->dataSource, "rand", strlen("rand"))) {
tsRand = true; sourceRand = true;
} else { } else {
tsRand = false; sourceRand = false; // from sample data file
}
char *bindBuffer = malloc(g_args.len_of_binary);
if (bindBuffer == NULL) {
errorPrint("%s() LN%d, Failed to allocate %d bind buffer\n",
__func__, __LINE__, g_args.len_of_binary);
free(bindArray);
return -1;
} }
uint32_t k; uint32_t k;
...@@ -5581,7 +5624,7 @@ static int32_t prepareStbStmt(SSuperTable *stbInfo, ...@@ -5581,7 +5624,7 @@ static int32_t prepareStbStmt(SSuperTable *stbInfo,
bind_ts = (int64_t *)ptr; bind_ts = (int64_t *)ptr;
bind->buffer_type = TSDB_DATA_TYPE_TIMESTAMP; bind->buffer_type = TSDB_DATA_TYPE_TIMESTAMP;
if (tsRand) { if (sourceRand) {
*bind_ts = startTime + getTSRandTail( *bind_ts = startTime + getTSRandTail(
stbInfo->timeStampStep, k, stbInfo->timeStampStep, k,
stbInfo->disorderRatio, stbInfo->disorderRatio,
...@@ -5596,27 +5639,64 @@ static int32_t prepareStbStmt(SSuperTable *stbInfo, ...@@ -5596,27 +5639,64 @@ static int32_t prepareStbStmt(SSuperTable *stbInfo,
ptr += bind->buffer_length; ptr += bind->buffer_length;
int cursor = 0;
for (int i = 0; i < stbInfo->columnCount; i ++) { for (int i = 0; i < stbInfo->columnCount; i ++) {
bind = (TAOS_BIND *)((char *)bindArray + (sizeof(TAOS_BIND) * (i + 1))); bind = (TAOS_BIND *)((char *)bindArray + (sizeof(TAOS_BIND) * (i + 1)));
if (sourceRand) {
if ( -1 == prepareStmtBindArrayByType(
bind,
stbInfo->columns[i].dataType,
stbInfo->columns[i].dataLen,
&ptr,
NULL)) {
free(bindArray);
free(bindBuffer);
return -1;
}
} else {
char *restStr = stbInfo->sampleDataBuf + cursor;
int lengthOfRest = strlen(restStr);
int index = 0;
for (index = 0; index < lengthOfRest; index ++) {
if (restStr[index] == ',') {
break;
}
}
memset(bindBuffer, 0, g_args.len_of_binary);
strncpy(bindBuffer, restStr, index);
cursor += index + 1; // skip ',' too
if ( -1 == prepareStmtBindArrayByType( if ( -1 == prepareStmtBindArrayByType(
bind, bind,
stbInfo->columns[i].dataType, stbInfo->columns[i].dataType,
stbInfo->columns[i].dataLen, stbInfo->columns[i].dataLen,
&ptr)) { &ptr,
bindBuffer)) {
free(bindArray);
free(bindBuffer);
return -1; return -1;
} }
} }
}
taos_stmt_bind_param(stmt, (TAOS_BIND *)bindArray); taos_stmt_bind_param(stmt, (TAOS_BIND *)bindArray);
// if msg > 3MB, break // if msg > 3MB, break
taos_stmt_add_batch(stmt); taos_stmt_add_batch(stmt);
k++; k++;
recordFrom ++; recordFrom ++;
if (!sourceRand) {
(*pSamplePos) ++;
}
if (recordFrom >= insertRows) { if (recordFrom >= insertRows) {
break; break;
} }
} }
free(bindBuffer);
free(bindArray); free(bindArray);
return k; return k;
} }
...@@ -5809,13 +5889,14 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { ...@@ -5809,13 +5889,14 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) {
if (superTblInfo) { if (superTblInfo) {
if (superTblInfo->iface == STMT_IFACE) { if (superTblInfo->iface == STMT_IFACE) {
#if STMT_IFACE_ENABLED == 1 #if STMT_IFACE_ENABLED == 1
generated = prepareStbStmt(superTblInfo, generated = prepareStbStmt(
superTblInfo,
pThreadInfo->stmt, pThreadInfo->stmt,
tableName, tableName,
batchPerTbl, batchPerTbl,
insertRows, i, insertRows, i,
startTime, startTime,
pThreadInfo->buffer); &(pThreadInfo->samplePos));
#else #else
generated = -1; generated = -1;
#endif #endif
...@@ -6040,7 +6121,8 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) { ...@@ -6040,7 +6121,8 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) {
pThreadInfo->stmt, pThreadInfo->stmt,
tableName, tableName,
g_args.num_of_RPR, g_args.num_of_RPR,
insertRows, i, start_time, pstr); insertRows, i, start_time,
&(pThreadInfo->samplePos));
#else #else
generated = -1; generated = -1;
#endif #endif
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册