From 98c026fd7c1e521c50c235f49d0ad1c5ba37c369 Mon Sep 17 00:00:00 2001 From: Shuduo Sang Date: Sat, 17 Jul 2021 17:53:48 +0800 Subject: [PATCH] Hotfix/sangshuduo/td 5302 taosdemo stmt interlace for master (#6900) * [TD-5302]:taosdemo stmt interlace. * fix the logic interlace rows compare insert rows. --- src/kit/taosdemo/taosdemo.c | 160 +++++++++++++++++++++++++----------- 1 file changed, 111 insertions(+), 49 deletions(-) diff --git a/src/kit/taosdemo/taosdemo.c b/src/kit/taosdemo/taosdemo.c index 8b3f5743de..e35e54751f 100644 --- a/src/kit/taosdemo/taosdemo.c +++ b/src/kit/taosdemo/taosdemo.c @@ -106,9 +106,9 @@ enum TEST_MODE { typedef enum CREATE_SUB_TALBE_MOD_EN { - PRE_CREATE_SUBTBL, - AUTO_CREATE_SUBTBL, - NO_CREATE_SUBTBL + PRE_CREATE_SUBTBL, + AUTO_CREATE_SUBTBL, + NO_CREATE_SUBTBL } CREATE_SUB_TALBE_MOD_EN; typedef enum TALBE_EXISTS_EN { @@ -637,7 +637,7 @@ static FILE * g_fpOfInsertResult = NULL; #define performancePrint(fmt, ...) \ do { if (g_args.performance_print) \ - fprintf(stderr, "VERB: "fmt, __VA_ARGS__); } while(0) + fprintf(stderr, "PERF: "fmt, __VA_ARGS__); } while(0) #define errorPrint(fmt, ...) \ do { fprintf(stderr, "ERROR: "fmt, __VA_ARGS__); } while(0) @@ -3184,8 +3184,10 @@ static void createChildTables() { if (g_Dbs.db[i].superTblCount > 0) { // with super table for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) { - if ((AUTO_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable) - || (TBL_ALREADY_EXISTS == g_Dbs.db[i].superTbls[j].childTblExists)) { + if ((AUTO_CREATE_SUBTBL + == g_Dbs.db[i].superTbls[j].autoCreateTable) + || (TBL_ALREADY_EXISTS + == g_Dbs.db[i].superTbls[j].childTblExists)) { continue; } verbosePrint("%s() LN%d: %s\n", __func__, __LINE__, @@ -4157,6 +4159,22 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { goto PARSE_OVER; } */ + cJSON* insertRows = cJSON_GetObjectItem(stbInfo, "insert_rows"); + if (insertRows && insertRows->type == cJSON_Number) { + if (insertRows->valueint < 0) { + errorPrint("%s() LN%d, failed to read json, insert_rows input mistake\n", + __func__, __LINE__); + goto PARSE_OVER; + } + g_Dbs.db[i].superTbls[j].insertRows = insertRows->valueint; + } else if (!insertRows) { + g_Dbs.db[i].superTbls[j].insertRows = 0x7FFFFFFFFFFFFFFF; + } else { + errorPrint("%s() LN%d, failed to read json, insert_rows input mistake\n", + __func__, __LINE__); + goto PARSE_OVER; + } + cJSON* stbInterlaceRows = cJSON_GetObjectItem(stbInfo, "interlace_rows"); if (stbInterlaceRows && stbInterlaceRows->type == cJSON_Number) { if (stbInterlaceRows->valueint < 0) { @@ -4165,15 +4183,15 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { goto PARSE_OVER; } g_Dbs.db[i].superTbls[j].interlaceRows = stbInterlaceRows->valueint; - // rows per table need be less than insert batch - if (g_Dbs.db[i].superTbls[j].interlaceRows > g_args.num_of_RPR) { - printf("NOTICE: db[%d].superTbl[%d]'s interlace rows value %u > num_of_records_per_req %u\n\n", - i, j, g_Dbs.db[i].superTbls[j].interlaceRows, - g_args.num_of_RPR); - printf(" interlace rows value will be set to num_of_records_per_req %u\n\n", - g_args.num_of_RPR); - prompt(); - g_Dbs.db[i].superTbls[j].interlaceRows = g_args.num_of_RPR; + + if (g_Dbs.db[i].superTbls[j].interlaceRows > g_Dbs.db[i].superTbls[j].insertRows) { + printf("NOTICE: db[%d].superTbl[%d]'s interlace rows value %u > insert_rows %"PRId64"\n\n", + i, j, g_Dbs.db[i].superTbls[j].interlaceRows, + g_Dbs.db[i].superTbls[j].insertRows); + printf(" interlace rows value will be set to insert_rows %"PRId64"\n\n", + g_Dbs.db[i].superTbls[j].insertRows); + prompt(); + g_Dbs.db[i].superTbls[j].interlaceRows = g_Dbs.db[i].superTbls[j].insertRows; } } else if (!stbInterlaceRows) { g_Dbs.db[i].superTbls[j].interlaceRows = 0; // 0 means progressive mode, > 0 mean interlace mode. max value is less or equ num_of_records_per_req @@ -4210,22 +4228,6 @@ static bool getMetaFromInsertJsonFile(cJSON* root) { goto PARSE_OVER; } - cJSON* insertRows = cJSON_GetObjectItem(stbInfo, "insert_rows"); - if (insertRows && insertRows->type == cJSON_Number) { - if (insertRows->valueint < 0) { - errorPrint("%s() LN%d, failed to read json, insert_rows input mistake\n", - __func__, __LINE__); - goto PARSE_OVER; - } - g_Dbs.db[i].superTbls[j].insertRows = insertRows->valueint; - } else if (!insertRows) { - g_Dbs.db[i].superTbls[j].insertRows = 0x7FFFFFFFFFFFFFFF; - } else { - errorPrint("%s() LN%d, failed to read json, insert_rows input mistake\n", - __func__, __LINE__); - goto PARSE_OVER; - } - cJSON* insertInterval = cJSON_GetObjectItem(stbInfo, "insert_interval"); if (insertInterval && insertInterval->type == cJSON_Number) { g_Dbs.db[i].superTbls[j].insertInterval = insertInterval->valueint; @@ -5152,12 +5154,18 @@ static int32_t generateDataTailWithoutStb( char **data_type = g_args.datatype; int lenOfBinary = g_args.len_of_binary; - retLen = generateData(data, data_type, - startTime + getTSRandTail( - (int64_t) DEFAULT_TIMESTAMP_STEP, k, - g_args.disorderRatio, - g_args.disorderRange), - lenOfBinary); + if (g_args.disorderRatio) { + retLen = generateData(data, data_type, + startTime + getTSRandTail( + (int64_t) DEFAULT_TIMESTAMP_STEP, k, + g_args.disorderRatio, + g_args.disorderRange), + lenOfBinary); + } else { + retLen = generateData(data, data_type, + startTime + (int64_t) (DEFAULT_TIMESTAMP_STEP* k), + lenOfBinary); + } if (len > remainderBufLen) break; @@ -5209,14 +5217,14 @@ static int32_t generateStbDataTail( bool tsRand; if (0 == strncasecmp(superTblInfo->dataSource, "rand", strlen("rand"))) { - tsRand = true; + tsRand = true; } else { - tsRand = false; + tsRand = false; } verbosePrint("%s() LN%d batch=%u buflen=%"PRId64"\n", __func__, __LINE__, batch, remainderBufLen); - int32_t k = 0; + int32_t k; for (k = 0; k < batch;) { char data[MAX_DATA_SIZE]; memset(data, 0, MAX_DATA_SIZE); @@ -5665,10 +5673,15 @@ static int32_t prepareStmtWithoutStb( bind_ts = (int64_t *)ptr; bind->buffer_type = TSDB_DATA_TYPE_TIMESTAMP; - *bind_ts = startTime + getTSRandTail( - (int64_t)DEFAULT_TIMESTAMP_STEP, k, - g_args.disorderRatio, - g_args.disorderRange); + + if (g_args.disorderRatio) { + *bind_ts = startTime + getTSRandTail( + (int64_t)DEFAULT_TIMESTAMP_STEP, k, + g_args.disorderRatio, + g_args.disorderRange); + } else { + *bind_ts = startTime + (int64_t)(DEFAULT_TIMESTAMP_STEP * k); + } bind->buffer_length = sizeof(int64_t); bind->buffer = bind_ts; bind->length = &bind->buffer_length; @@ -5753,7 +5766,7 @@ static int32_t prepareStbStmt( bind_ts = (int64_t *)ptr; bind->buffer_type = TSDB_DATA_TYPE_TIMESTAMP; - if (sourceRand) { + if (stbInfo->disorderRatio) { *bind_ts = startTime + getTSRandTail( stbInfo->timeStampStep, k, stbInfo->disorderRatio, @@ -5820,6 +5833,7 @@ static int32_t prepareStbStmt( if (!sourceRand) { (*pSamplePos) ++; } + if (recordFrom >= insertRows) { break; } @@ -5829,6 +5843,43 @@ static int32_t prepareStbStmt( free(bindArray); return k; } + +static int32_t prepareStbStmtInterlace( + SSuperTable *stbInfo, + TAOS_STMT *stmt, + char *tableName, uint32_t batch, + uint64_t insertRows, + uint64_t recordFrom, + int64_t startTime, + int64_t *pSamplePos) +{ + return prepareStbStmt( + stbInfo, + stmt, + tableName, + g_args.num_of_RPR, + insertRows, 0, startTime, + pSamplePos); +} + +static int32_t prepareStbStmtProgressive( + SSuperTable *stbInfo, + TAOS_STMT *stmt, + char *tableName, uint32_t batch, + uint64_t insertRows, + uint64_t recordFrom, + int64_t startTime, + int64_t *pSamplePos) +{ + return prepareStbStmt( + stbInfo, + stmt, + tableName, + g_args.num_of_RPR, + insertRows, recordFrom, startTime, + pSamplePos); +} + #endif static int32_t generateStbProgressiveData( @@ -6003,7 +6054,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { uint32_t recOfBatch = 0; - for (uint32_t i = 0; i < batchPerTblTimes; i ++) { + for (uint64_t i = 0; i < batchPerTblTimes; i ++) { char tableName[TSDB_TABLE_NAME_LEN]; getTableName(tableName, pThreadInfo, tableSeq); @@ -6020,7 +6071,7 @@ static void* syncWriteInterlace(threadInfo *pThreadInfo) { if (superTblInfo) { if (superTblInfo->iface == STMT_IFACE) { #if STMT_IFACE_ENABLED == 1 - generated = prepareStbStmt( + generated = prepareStbStmtInterlace( superTblInfo, pThreadInfo->stmt, tableName, @@ -6249,7 +6300,7 @@ static void* syncWriteProgressive(threadInfo *pThreadInfo) { if (superTblInfo) { if (superTblInfo->iface == STMT_IFACE) { #if STMT_IFACE_ENABLED == 1 - generated = prepareStbStmt( + generated = prepareStbStmtProgressive( superTblInfo, pThreadInfo->stmt, tableName, @@ -6677,13 +6728,24 @@ static void startMultiThreadInsertData(int threads, char* db_name, char buffer[3000]; char *pstr = buffer; - pstr += sprintf(pstr, "INSERT INTO ? values(?"); + + if (AUTO_CREATE_SUBTBL == superTblInfo->autoCreateTable) { + pstr += sprintf(pstr, "INSERT INTO ? USING %s TAGS(?", + superTblInfo->sTblName); + for (int tag = 0; tag < (superTblInfo->tagCount - 1); tag ++ ) { + pstr += sprintf(pstr, ",?"); + } + pstr += sprintf(pstr, ") VALUES(?"); + } else { + pstr += sprintf(pstr, "INSERT INTO ? VALUES(?"); + } for (int col = 0; col < columnCount; col ++) { pstr += sprintf(pstr, ",?"); } pstr += sprintf(pstr, ")"); + debugPrint("%s() LN%d, buffer: %s", __func__, __LINE__, buffer); int ret = taos_stmt_prepare(pThreadInfo->stmt, buffer, 0); if (ret != 0){ errorPrint("failed to execute taos_stmt_prepare. return 0x%x. reason: %s\n", -- GitLab