#4582

Committed 17 Jul 2025 10:48AM UTC coverage: 3.863%. Remained the same

Build # #4582

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #31990 from taosdata/3.0

3.0

Run Details

230 of 7082 branches covered (3.25%)

Branch coverage included in aggregate %.

0 of 62 new or added lines in 6 files covered. (0.0%)

1617 existing lines in 3 files now uncovered.

382 of 8759 relevant lines covered (4.36%)

1.22 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/tools/taos-tools/src/benchCommandOpt.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the MIT license as published by the Free Software
 * Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 */

#include "cus_name.h"  // include/util/
#include <bench.h>
#include "benchLog.h"
#include <toolsdef.h>

extern char      g_configDir[MAX_PATH_LEN];

char *g_aggreFuncDemo[] = {"*",
                           "count(*)",
                           "avg(current)",
                           "sum(current)",
                           "max(current)",
                           "min(current)",
                           "first(current)",
                           "last(current)"};
char *g_aggreFunc[] = {"*",       "count(*)", "avg(C0)",   "sum(C0)",
                       "max(C0)", "min(C0)",  "first(C0)", "last(C0)"};

void printVersion() {
    // version, macro define in src/CMakeLists.txt
    printf("%s\n%sBenchmark version: %s\n", TD_PRODUCT_NAME, CUS_PROMPT, TD_VER_NUMBER);
    printf("git: %s\n", TAOSBENCHMARK_COMMIT_ID);
    printf("build: %s\n", BUILD_INFO);
}


void processSingleToken(char* token, BArray* fields, int index, bool isTag) {
    Field* field = benchCalloc(1, sizeof(Field), true);
    benchArrayPush(fields, field);
    field = benchArrayGet(fields, index);

    regex_t regex;
    regmatch_t pmatch[3];
    int reti;

    // BINARY/NCHAR/VARCHAR/JSON/GEOMETRY/VARBINARY
    reti = regcomp(&regex, "^(BINARY|NCHAR|VARCHAR|JSON|GEOMETRY|VARBINARY)\\(([1-9][0-9]*)\\)$", REG_ICASE | REG_EXTENDED);
    if (!reti) {
        reti = regexec(&regex, token, 3, pmatch, 0);
        if (!reti) {
            char type[DATATYPE_BUFF_LEN] = {0};
            char length[BIGINT_BUFF_LEN] = {0};
            strncpy(type, token + pmatch[1].rm_so, pmatch[1].rm_eo - pmatch[1].rm_so);
            type[pmatch[1].rm_eo - pmatch[1].rm_so] = '\0';
            strncpy(length, token + pmatch[2].rm_so, pmatch[2].rm_eo - pmatch[2].rm_so);
            field->type = convertStringToDatatype(type, 0, NULL);
            field->length = atoi(length);
            regfree(&regex);
            goto SET_PROPS;
        }
        regfree(&regex);
    }

    // DECIMAL
    reti = regcomp(&regex, "^DECIMAL\\s*\\(\\s*(-?[0-9]+)\\s*,\\s*(-?[0-9]+)\\s*\\)$", REG_ICASE | REG_EXTENDED);
    if (!reti) {
        reti = regexec(&regex, token, 3, pmatch, 0);
        if (!reti) {
            char precision[DECIMAL_BUFF_LEN] = {0};
            char scale[DECIMAL_BUFF_LEN] = {0};
            strncpy(precision, token + pmatch[1].rm_so, pmatch[1].rm_eo - pmatch[1].rm_so);
            precision[pmatch[1].rm_eo - pmatch[1].rm_so] = '\0';
            strncpy(scale, token + pmatch[2].rm_so, pmatch[2].rm_eo - pmatch[2].rm_so);
            scale[pmatch[2].rm_eo - pmatch[2].rm_so] = '\0';

            int p = atoi(precision), s = atoi(scale);
            if (p > TSDB_DECIMAL128_MAX_PRECISION || p <= 0) {
                errorPrint("Invalid precision value of decimal type in args, precision: %d, scale: %d\n", p, s);
                exit(EXIT_FAILURE);
            }
            if (s < 0 || s > p) {
                errorPrint("Invalid scale value of decimal type in args, precision: %d, scale: %d\n", p, s);
                exit(EXIT_FAILURE);
            }
            field->precision = p;
            field->scale = s;
            field->type = convertStringToDatatype("DECIMAL", 0, &field->precision);
            field->length = convertTypeToLength(field->type);
            regfree(&regex);

            if (field->type == TSDB_DATA_TYPE_DECIMAL) {
                getDecimal128DefaultMax(p, s, &field->decMax.dec128);
                getDecimal128DefaultMin(p, s, &field->decMin.dec128);
            } else {
                getDecimal64DefaultMax(p, s, &field->decMax.dec64);
                getDecimal64DefaultMin(p, s, &field->decMin.dec64);
            }

            goto SET_PROPS;
        }
        regfree(&regex);
    }

    // other
    field->type = convertStringToDatatype(token, 0, NULL);
    field->length = convertTypeToLength(field->type);

SET_PROPS:
    field->min = convertDatatypeToDefaultMin(field->type);
    field->max = convertDatatypeToDefaultMax(field->type);
    snprintf(field->name, TSDB_COL_NAME_LEN, isTag ? "t%d" : "c%d", index);
}


void parseFieldDatatype(char* dataType, BArray* fields, bool isTag) {
    benchArrayClear(fields);
    if (strstr(dataType, ",") == NULL) {
        processSingleToken(dataType, fields, 0, isTag);
    } else {
        char* dupStr        = strdup(dataType);
        char* start         = dupStr;
        char* current       = start;
        int   bracketDepth  = 0;
        int   index         = 0;

        while (*current != '\0') {
            if (*current == '(') {
                bracketDepth++;
            } else if (*current == ')') {
                if (bracketDepth > 0) bracketDepth--;
                else {
                    errorPrint("Unbalanced parentheses in data type: %s\n", dataType);
                    exit(EXIT_FAILURE);
                }
            } else if (*current == ',' && bracketDepth == 0) {
                *current = '\0';
                processSingleToken(start, fields, index++, isTag);
                start = current + 1;
            }
            current++;
        }

        if (start < current) {
            processSingleToken(start, fields, index, isTag); 
        }
        tmfree(dupStr);
    }
}


static void initStable() {
    SDataBase *database = benchArrayGet(g_arguments->databases, 0);
    database->superTbls = benchArrayInit(1, sizeof(SSuperTable));
    SSuperTable * stbInfo = benchCalloc(1, sizeof(SSuperTable), true);
    benchArrayPush(database->superTbls, stbInfo);
    stbInfo = benchArrayGet(database->superTbls, 0);
    stbInfo->iface = TAOSC_IFACE;
    stbInfo->stbName = "meters";
    stbInfo->childTblPrefix = DEFAULT_TB_PREFIX;
    stbInfo->use_metric = 1;
    stbInfo->max_sql_len = TSDB_MAX_ALLOWED_SQL_LEN;
    stbInfo->cols = benchArrayInit(3, sizeof(Field));
    for (int i = 0; i < 3; ++i) {
        Field *col = benchCalloc(1, sizeof(Field), true);
        benchArrayPush(stbInfo->cols, col);
    }
    Field * c1 = benchArrayGet(stbInfo->cols, 0);
    Field * c2 = benchArrayGet(stbInfo->cols, 1);
    Field * c3 = benchArrayGet(stbInfo->cols, 2);

    c1->type = TSDB_DATA_TYPE_FLOAT;
    c2->type = TSDB_DATA_TYPE_INT;
    c3->type = TSDB_DATA_TYPE_FLOAT;

    c1->length = sizeof(float);
    c2->length = sizeof(int32_t);
    c3->length = sizeof(float);

    TOOLS_STRNCPY(c1->name, "current", TSDB_COL_NAME_LEN + 1);
    TOOLS_STRNCPY(c2->name, "voltage", TSDB_COL_NAME_LEN + 1);
    TOOLS_STRNCPY(c3->name, "phase", TSDB_COL_NAME_LEN + 1);

    c1->min = 9;
    c1->max = 10;    
    //fun = "4*sin(x)+10*random(5)+10"
    c1->funType  = FUNTYPE_SIN;
    c1->multiple = 4;
    c1->random   = 5;
    c1->addend   = 10;
    c1->base     = 10;

    c2->min = 110;
    c2->max = 119;
    //fun = "1*square(0,60,50,0)+100*random(20)+120"
    c2->funType  = FUNTYPE_SQUARE;
    c2->multiple = 1;
    c2->random   = 20;
    c2->addend   = 100;
    c2->base     = 120;

    c3->min = 115;
    c3->max = 125;
    // fun = "1*saw(0,40,40,0)+50*random(10)+30"
    c3->funType  = FUNTYPE_SAW;
    c3->multiple = 1;
    c3->random   = 10;
    c3->addend   = 50;
    c3->base     = 30;

    stbInfo->tags = benchArrayInit(2, sizeof(Field));
    for (int i = 0; i < 2; ++i) {
        Field * tag = benchCalloc(1, sizeof(Field), true);
        benchArrayPush(stbInfo->tags, tag);
    }
    Field * t1 = benchArrayGet(stbInfo->tags, 0);
    Field * t2 = benchArrayGet(stbInfo->tags, 1);

    t1->type = TSDB_DATA_TYPE_INT;
    t2->type = TSDB_DATA_TYPE_BINARY;

    t1->length = sizeof(int32_t);
    t2->length = 24;

    TOOLS_STRNCPY(t1->name, "groupid", TSDB_COL_NAME_LEN + 1);
    TOOLS_STRNCPY(t2->name, "location", TSDB_COL_NAME_LEN + 1);
    TOOLS_STRNCPY(stbInfo->primaryKeyName, "ts", TSDB_COL_NAME_LEN + 1);
    t1->min = 1;
    t1->max = 100000;


    stbInfo->insert_interval = 0;
    stbInfo->timestamp_step = 1;
    stbInfo->angle_step = 1;
    stbInfo->interlaceRows = 0;
    stbInfo->childTblCount = DEFAULT_CHILDTABLES;
    stbInfo->childTblLimit = 0;
    stbInfo->childTblOffset = 0;
    stbInfo->autoTblCreating = false;
    stbInfo->childTblExists = false;
    stbInfo->random_data_source = true;
    stbInfo->lineProtocol = TSDB_SML_LINE_PROTOCOL;

    stbInfo->insertRows = DEFAULT_INSERT_ROWS;
    stbInfo->disorderRange = DEFAULT_DISORDER_RANGE;
    stbInfo->disorderRatio = 0;
    stbInfo->file_factor = -1;
    stbInfo->delay = -1;
    stbInfo->keep_trying = 0;
    stbInfo->trying_interval = 0;
}

static void initDatabase() {
    g_arguments->databases = benchArrayInit(1, sizeof(SDataBase));
    SDataBase *database = benchCalloc(1, sizeof(SDataBase), true);
    benchArrayPush(g_arguments->databases, database);
    database = benchArrayGet(g_arguments->databases, 0);
    database->dbName = DEFAULT_DATABASE;
    database->drop = true;
    database->precision = TSDB_TIME_PRECISION_MILLI;
    database->sml_precision = TSDB_SML_TIMESTAMP_MILLI_SECONDS;
    database->cfgs = benchArrayInit(1, sizeof(SDbCfg));
}

void initArgument() {
    g_arguments = benchCalloc(1, sizeof(SArguments), true);
    if (taos_get_client_info()[0] == '3') {
        g_arguments->taosc_version = 3;
    } else {
        g_arguments->taosc_version = 2;
    }
    g_arguments->test_mode = INSERT_TEST;
    g_arguments->demo_mode = true;
    g_arguments->host = NULL;
    g_arguments->port = 0;
    g_arguments->port_inputted = false;
    g_arguments->telnet_tcp_port = TELNET_TCP_PORT;
    g_arguments->user     = NULL;
    g_arguments->password = NULL;
    g_arguments->answer_yes = 0;
    g_arguments->debug_print = 0;
    g_arguments->binwidth = DEFAULT_BINWIDTH;
    g_arguments->performance_print = 0;
    g_arguments->output_file = DEFAULT_OUTPUT;
    g_arguments->nthreads = DEFAULT_NTHREADS;
    g_arguments->table_threads = DEFAULT_NTHREADS;
    g_arguments->prepared_rand = DEFAULT_PREPARED_RAND;
    g_arguments->reqPerReq = DEFAULT_REQ_PER_REQ;
    g_arguments->totalChildTables = DEFAULT_CHILDTABLES;
    g_arguments->actualChildTables = 0;
    g_arguments->autoCreatedChildTables = 0;
    g_arguments->existedChildTables = 0;
    g_arguments->chinese = false;
    g_arguments->aggr_func = 0;
    g_arguments->terminate = false;

    g_arguments->supplementInsert = false;
    g_arguments->startTimestamp = DEFAULT_START_TIME;
    g_arguments->partialColNum = 0;

    g_arguments->keep_trying = 0;
    g_arguments->trying_interval = 0;
    g_arguments->iface = TAOSC_IFACE;
    g_arguments->rest_server_ver_major = -1;
    g_arguments->inputted_vgroups = -1;

    g_arguments->mistMode = false;
    g_arguments->connMode = CONN_MODE_INVALID;

    initDatabase();
    initStable();
    g_arguments->streams = benchArrayInit(1, sizeof(SSTREAM));
}

void modifyArgument() {
    SDataBase * database = benchArrayGet(g_arguments->databases, 0);
    SSuperTable *superTable = benchArrayGet(database->superTbls, 0);

    superTable->startTimestamp = g_arguments->startTimestamp;

    if (0 != g_arguments->partialColNum) {
        superTable->partialColNum = g_arguments->partialColNum;
    }

    for (int i = 0; i < superTable->cols->size; ++i) {
        Field * col = benchArrayGet(superTable->cols, i);
        if (!g_arguments->demo_mode) {
            snprintf(col->name, TSDB_COL_NAME_LEN, "c%d", i);
            col->min = convertDatatypeToDefaultMin(col->type);
            col->max = convertDatatypeToDefaultMax(col->type);
        }
        if (col->length == 0) {
            col->length = g_arguments->binwidth;
        }
    }

    for (int i = 0; i < superTable->tags->size; ++i) {
        Field* tag = benchArrayGet(superTable->tags, i);
        if (!g_arguments->demo_mode) {
            snprintf(tag->name, TSDB_COL_NAME_LEN, "t%d", i);
        }
        if (tag->length == 0) {
            tag->length = g_arguments->binwidth;
        }
    }

    if (g_arguments->intColumnCount > superTable->cols->size) {
        for (int i = superTable->cols->size;
                i < g_arguments->intColumnCount; ++i) {
            Field * col = benchCalloc(1, sizeof(Field), true);
            benchArrayPush(superTable->cols, col);
            col = benchArrayGet(superTable->cols, i);
            col->type = TSDB_DATA_TYPE_INT;
            col->length = sizeof(int32_t);
            snprintf(col->name, TSDB_COL_NAME_LEN, "c%d", i);
            col->min = convertDatatypeToDefaultMin(col->type);
            col->max = convertDatatypeToDefaultMax(col->type);
        }
    }

    if (g_arguments->keep_trying) {
        superTable->keep_trying = g_arguments->keep_trying;
        superTable->trying_interval = g_arguments->trying_interval;
    }
}

static void *queryStableAggrFunc(void *sarg) {
    threadInfo *pThreadInfo = (threadInfo *)sarg;

    TAOS *taos = NULL;
    if (REST_IFACE != g_arguments->iface) {
        taos = pThreadInfo->conn->taos;
    }
#ifdef LINUX
    prctl(PR_SET_NAME, "queryStableAggrFunc");
#endif
    char *command = benchCalloc(1, TSDB_MAX_ALLOWED_SQL_LEN, false);
    FILE *  fp = g_arguments->fpOfInsertResult;
    SDataBase * database = benchArrayGet(g_arguments->databases, 0);
    SSuperTable * stbInfo = benchArrayGet(database->superTbls, 0);
    int64_t totalData = stbInfo->insertRows * stbInfo->childTblCount;
    char **aggreFunc;
    int    n;

    if (g_arguments->demo_mode) {
        aggreFunc = g_aggreFuncDemo;
        n = sizeof(g_aggreFuncDemo) / sizeof(g_aggreFuncDemo[0]);
    } else {
        aggreFunc = g_aggreFunc;
        n = sizeof(g_aggreFunc) / sizeof(g_aggreFunc[0]);
    }

    infoPrint("total Data: %" PRId64 "\n", totalData);
    if (fp) {
        fprintf(fp, "Querying On %" PRId64 " records:\n", totalData);
    }
    for (int j = 0; j < n; j++) {
        char condition[COND_BUF_LEN] = "\0";
        char tempS[LARGE_BUFF_LEN] = "\0";
        int64_t m = 10 < stbInfo->childTblCount ? 10 : stbInfo->childTblCount;
        for (int64_t i = 1; i <= m; i++) {
            if (i == 1) {
                if (g_arguments->demo_mode) {
                    snprintf(tempS, LARGE_BUFF_LEN,
                             "groupid = %" PRId64, i);
                } else {
                    snprintf(tempS, LARGE_BUFF_LEN,
                             "t0 = %" PRId64, i);
                }
            } else {
                if (g_arguments->demo_mode) {
                    snprintf(tempS, LARGE_BUFF_LEN,
                             " or groupid = %" PRId64 " ", i);
                } else {
                    snprintf(tempS, LARGE_BUFF_LEN,
                             " or t0 = %" PRId64 " ", i);
                }
            }
            strncat(condition, tempS, COND_BUF_LEN - 1);
            snprintf(command, TSDB_MAX_ALLOWED_SQL_LEN,
                     "SELECT %s FROM %s.meters WHERE %s",
                    aggreFunc[j], database->dbName,
                    condition);
            if (fp) {
                fprintf(fp, "%s\n", command);
            }
            double t = (double)toolsGetTimestampUs();
            int32_t code = -1;
            if (REST_IFACE == g_arguments->iface) {
                code = postProcessSql(command, NULL, 0, REST_IFACE,
                                    0, g_arguments->port, 0,
                                    pThreadInfo->sockfd, NULL);
            } else {
                TAOS_RES *res = taos_query(taos, command);
                code = taos_errno(res);
                if (code != 0) {
                    printErrCmdCodeStr(command, code, res);
                    free(command);
                    return NULL;
                }
                int count = 0;
                while (taos_fetch_row(res) != NULL) {
                    count++;
                }
                taos_free_result(res);
            }
            t = toolsGetTimestampUs() - t;
            if (fp) {
                fprintf(fp, "| Speed: %12.2f(per s) | Latency: %.4f(ms) |\n",
                        totalData / (t / 1000), t);
            }
            infoPrint("%s took %.6f second(s)\n\n", command,
                      t / 1000000);
        }
    }
    free(command);
    return NULL;
}

static void *queryNtableAggrFunc(void *sarg) {
    threadInfo *pThreadInfo = (threadInfo *)sarg;
    TAOS *      taos = NULL;
    if (pThreadInfo->conn) {
        taos = pThreadInfo->conn->taos;
    }
#ifdef LINUX
    prctl(PR_SET_NAME, "queryNtableAggrFunc");
#endif
    char *  command = benchCalloc(1, TSDB_MAX_ALLOWED_SQL_LEN, false);
    FILE *  fp = g_arguments->fpOfInsertResult;
    SDataBase * database = benchArrayGet(g_arguments->databases, 0);
    SSuperTable * stbInfo = benchArrayGet(database->superTbls, 0);
    int64_t totalData = stbInfo->childTblCount * stbInfo->insertRows;
    char **aggreFunc;
    int    n;

    if (g_arguments->demo_mode) {
        aggreFunc = g_aggreFuncDemo;
        n = sizeof(g_aggreFuncDemo) / sizeof(g_aggreFuncDemo[0]);
    } else {
        aggreFunc = g_aggreFunc;
        n = sizeof(g_aggreFunc) / sizeof(g_aggreFunc[0]);
    }

    infoPrint("totalData: %" PRId64 "\n", totalData);
    if (fp) {
        fprintf(fp,
                "| QFunctions |    QRecords    |   QSpeed(R/s)   |  "
                "QLatency(ms) |\n");
    }

    for (int j = 0; j < n; j++) {
        double   totalT = 0;
        uint64_t count = 0;
        for (int64_t i = 0; i < stbInfo->childTblCount; i++) {
            snprintf(command,
                    TSDB_MAX_ALLOWED_SQL_LEN,
                    g_arguments->escape_character
                    ? "SELECT %s FROM `%s`.`%s%" PRId64 "` WHERE ts>= %" PRIu64 
                    : "SELECT %s FROM %s.%s%" PRId64 " WHERE ts>= %" PRIu64 ,
                    aggreFunc[j],
                    database->dbName,
                    stbInfo->childTblPrefix, i,
                    (uint64_t) DEFAULT_START_TIME);
            double    t = (double)toolsGetTimestampUs();
            int32_t code = -1;
            if (REST_IFACE == g_arguments->iface) {
                code = postProcessSql(command, NULL, 0, REST_IFACE,
                                    0, g_arguments->port, 0,
                                    pThreadInfo->sockfd, NULL);
            } else {
                TAOS_RES *res = taos_query(taos, command);
                code = taos_errno(res);
                if (code != 0) {
                    printErrCmdCodeStr(command, code, res);
                    free(command);
                    return NULL;
                }
                while (taos_fetch_row(res) != NULL) {
                    count++;
                }
                taos_free_result(res);
            }

            t = toolsGetTimestampUs() - t;
            totalT += t;
        }
        if (fp) {
            fprintf(fp, "|%10s  |   %" PRId64 "   |  %12.2f   |   %10.2f  |\n",
                    (aggreFunc[j][0] == '*')
                        ?("   *   "):(aggreFunc[j]), totalData,
                    (double)(stbInfo->childTblCount*stbInfo->insertRows)/totalT,
                    totalT / 1000000);
        }
        infoPrint("<%s> took %.6f second(s)\n", command,
                  totalT / 1000000);
    }
    free(command);
    return NULL;
}

void queryAggrFunc() {
    pthread_t   read_id;
    threadInfo *pThreadInfo = benchCalloc(1, sizeof(threadInfo), false);
    if (NULL == pThreadInfo) {
        errorPrint("%s() failed to allocate memory\n", __func__);
        return;
    }
    SDataBase * database = benchArrayGet(g_arguments->databases, 0);
    if (NULL == database) {
        errorPrint("%s() failed to get database\n", __func__);
        free(pThreadInfo);
        return;
    }
    SSuperTable * stbInfo = benchArrayGet(database->superTbls, 0);
    if (NULL == stbInfo) {
        errorPrint("%s() failed to get super table\n", __func__);
        free(pThreadInfo);
        return;
    }

    // REST
    if (REST_IFACE != g_arguments->iface) {
        pThreadInfo->conn = initBenchConn();
        if (pThreadInfo->conn == NULL) {
            errorPrint("%s() failed to init connection\n", __func__);
            free(pThreadInfo);
            return;
        }
    } else {
        pThreadInfo->sockfd = createSockFd();
        if (pThreadInfo->sockfd < 0) {
            free(pThreadInfo);
            return;
        }
    }    
    if (stbInfo->use_metric) {
        pthread_create(&read_id, NULL, queryStableAggrFunc, pThreadInfo);
    } else {
        pthread_create(&read_id, NULL, queryNtableAggrFunc, pThreadInfo);
    }
    pthread_join(read_id, NULL);
    // REST
    if (REST_IFACE != g_arguments->iface) {
        closeBenchConn(pThreadInfo->conn);
    } else {
        if (pThreadInfo->sockfd) {
            destroySockFd(pThreadInfo->sockfd);
        }
    }
    free(pThreadInfo);
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the MIT license as published by the Free Software
6	* Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*/
12
13	#include "cus_name.h" // include/util/
14	#include <bench.h>
15	#include "benchLog.h"
16	#include <toolsdef.h>
17
18	extern char g_configDir[MAX_PATH_LEN];
19
20	char g_aggreFuncDemo[] = {"",
21	"count(*)",
22	"avg(current)",
23	"sum(current)",
24	"max(current)",
25	"min(current)",
26	"first(current)",
27	"last(current)"};
28	char g_aggreFunc[] = {"", "count(*)", "avg(C0)", "sum(C0)",
29	"max(C0)", "min(C0)", "first(C0)", "last(C0)"};
30
31	void printVersion() {	×
32	// version, macro define in src/CMakeLists.txt
33	printf("%s\n%sBenchmark version: %s\n", TD_PRODUCT_NAME, CUS_PROMPT, TD_VER_NUMBER);	×
34	printf("git: %s\n", TAOSBENCHMARK_COMMIT_ID);	×
35	printf("build: %s\n", BUILD_INFO);	×
36	}	×
37
38
39	void processSingleToken(char* token, BArray* fields, int index, bool isTag) {	×
40	Field* field = benchCalloc(1, sizeof(Field), true);	×
41	benchArrayPush(fields, field);	×
42	field = benchArrayGet(fields, index);	×
43
44	regex_t regex;
45	regmatch_t pmatch[3];
46	int reti;
47
48	// BINARY/NCHAR/VARCHAR/JSON/GEOMETRY/VARBINARY
49	reti = regcomp(&regex, "^(BINARY\|NCHAR\|VARCHAR\|JSON\|GEOMETRY\|VARBINARY)\\(([1-9][0-9]*)\\)$", REG_ICASE \| REG_EXTENDED);	×
50	if (!reti) {	×
51	reti = regexec(&regex, token, 3, pmatch, 0);	×
52	if (!reti) {	×
53	char type[DATATYPE_BUFF_LEN] = {0};	×
54	char length[BIGINT_BUFF_LEN] = {0};	×
55	strncpy(type, token + pmatch[1].rm_so, pmatch[1].rm_eo - pmatch[1].rm_so);	×
56	type[pmatch[1].rm_eo - pmatch[1].rm_so] = '\0';	×
57	strncpy(length, token + pmatch[2].rm_so, pmatch[2].rm_eo - pmatch[2].rm_so);	×
58	field->type = convertStringToDatatype(type, 0, NULL);	×
59	field->length = atoi(length);	×
60	regfree(&regex);	×
61	goto SET_PROPS;	×
62	}
63	regfree(&regex);	×
64	}
65
66	// DECIMAL
67	reti = regcomp(&regex, "^DECIMAL\\s\\(\\s(-?[0-9]+)\\s,\\s(-?[0-9]+)\\s*\\)$", REG_ICASE \| REG_EXTENDED);	×
68	if (!reti) {	×
69	reti = regexec(&regex, token, 3, pmatch, 0);	×
70	if (!reti) {	×
71	char precision[DECIMAL_BUFF_LEN] = {0};	×
72	char scale[DECIMAL_BUFF_LEN] = {0};	×
73	strncpy(precision, token + pmatch[1].rm_so, pmatch[1].rm_eo - pmatch[1].rm_so);	×
74	precision[pmatch[1].rm_eo - pmatch[1].rm_so] = '\0';	×
75	strncpy(scale, token + pmatch[2].rm_so, pmatch[2].rm_eo - pmatch[2].rm_so);	×
76	scale[pmatch[2].rm_eo - pmatch[2].rm_so] = '\0';	×
77
78	int p = atoi(precision), s = atoi(scale);	×
79	if (p > TSDB_DECIMAL128_MAX_PRECISION \|\| p <= 0) {	×
80	errorPrint("Invalid precision value of decimal type in args, precision: %d, scale: %d\n", p, s);	×
81	exit(EXIT_FAILURE);	×
82	}
83	if (s < 0 \|\| s > p) {	×
84	errorPrint("Invalid scale value of decimal type in args, precision: %d, scale: %d\n", p, s);	×
85	exit(EXIT_FAILURE);	×
86	}
87	field->precision = p;	×
88	field->scale = s;	×
89	field->type = convertStringToDatatype("DECIMAL", 0, &field->precision);	×
90	field->length = convertTypeToLength(field->type);	×
91	regfree(&regex);	×
92
93	if (field->type == TSDB_DATA_TYPE_DECIMAL) {	×
94	getDecimal128DefaultMax(p, s, &field->decMax.dec128);	×
95	getDecimal128DefaultMin(p, s, &field->decMin.dec128);	×
96	} else {
97	getDecimal64DefaultMax(p, s, &field->decMax.dec64);	×
98	getDecimal64DefaultMin(p, s, &field->decMin.dec64);	×
99	}
100
101	goto SET_PROPS;	×
102	}
103	regfree(&regex);	×
104	}
105
106	// other
107	field->type = convertStringToDatatype(token, 0, NULL);	×
108	field->length = convertTypeToLength(field->type);	×
109
110	SET_PROPS:	×
111	field->min = convertDatatypeToDefaultMin(field->type);	×
112	field->max = convertDatatypeToDefaultMax(field->type);	×
113	snprintf(field->name, TSDB_COL_NAME_LEN, isTag ? "t%d" : "c%d", index);	×
114	}	×
115
116
117	void parseFieldDatatype(char* dataType, BArray* fields, bool isTag) {	×
118	benchArrayClear(fields);	×
119	if (strstr(dataType, ",") == NULL) {	×
120	processSingleToken(dataType, fields, 0, isTag);	×
121	} else {
122	char* dupStr = strdup(dataType);	×
123	char* start = dupStr;	×
124	char* current = start;	×
125	int bracketDepth = 0;	×
126	int index = 0;	×
127
128	while (*current != '\0') {	×
129	if (*current == '(') {	×
130	bracketDepth++;	×
131	} else if (*current == ')') {	×
132	if (bracketDepth > 0) bracketDepth--;	×
133	else {
134	errorPrint("Unbalanced parentheses in data type: %s\n", dataType);	×
135	exit(EXIT_FAILURE);	×
136	}
137	} else if (*current == ',' && bracketDepth == 0) {	×
138	*current = '\0';	×
139	processSingleToken(start, fields, index++, isTag);	×
140	start = current + 1;	×
141	}
142	current++;	×
143	}
144
145	if (start < current) {	×
146	processSingleToken(start, fields, index, isTag);	×
147	}
148	tmfree(dupStr);	×
149	}
150	}	×
151
152
153	static void initStable() {	×
154	SDataBase *database = benchArrayGet(g_arguments->databases, 0);	×
155	database->superTbls = benchArrayInit(1, sizeof(SSuperTable));	×
156	SSuperTable * stbInfo = benchCalloc(1, sizeof(SSuperTable), true);	×
157	benchArrayPush(database->superTbls, stbInfo);	×
158	stbInfo = benchArrayGet(database->superTbls, 0);	×
159	stbInfo->iface = TAOSC_IFACE;	×
160	stbInfo->stbName = "meters";	×
161	stbInfo->childTblPrefix = DEFAULT_TB_PREFIX;	×
162	stbInfo->use_metric = 1;	×
163	stbInfo->max_sql_len = TSDB_MAX_ALLOWED_SQL_LEN;	×
164	stbInfo->cols = benchArrayInit(3, sizeof(Field));	×
165	for (int i = 0; i < 3; ++i) {	×
166	Field *col = benchCalloc(1, sizeof(Field), true);	×
167	benchArrayPush(stbInfo->cols, col);	×
168	}
169	Field * c1 = benchArrayGet(stbInfo->cols, 0);	×
170	Field * c2 = benchArrayGet(stbInfo->cols, 1);	×
171	Field * c3 = benchArrayGet(stbInfo->cols, 2);	×
172
173	c1->type = TSDB_DATA_TYPE_FLOAT;	×
174	c2->type = TSDB_DATA_TYPE_INT;	×
175	c3->type = TSDB_DATA_TYPE_FLOAT;	×
176
177	c1->length = sizeof(float);	×
178	c2->length = sizeof(int32_t);	×
179	c3->length = sizeof(float);	×
180
181	TOOLS_STRNCPY(c1->name, "current", TSDB_COL_NAME_LEN + 1);	×
182	TOOLS_STRNCPY(c2->name, "voltage", TSDB_COL_NAME_LEN + 1);	×
183	TOOLS_STRNCPY(c3->name, "phase", TSDB_COL_NAME_LEN + 1);	×
184
185	c1->min = 9;	×
186	c1->max = 10;	×
187	//fun = "4sin(x)+10random(5)+10"
188	c1->funType = FUNTYPE_SIN;	×
189	c1->multiple = 4;	×
190	c1->random = 5;	×
191	c1->addend = 10;	×
192	c1->base = 10;	×
193
194	c2->min = 110;	×
195	c2->max = 119;	×
196	//fun = "1square(0,60,50,0)+100random(20)+120"
197	c2->funType = FUNTYPE_SQUARE;	×
198	c2->multiple = 1;	×
199	c2->random = 20;	×
200	c2->addend = 100;	×
201	c2->base = 120;	×
202
203	c3->min = 115;	×
204	c3->max = 125;	×
205	// fun = "1saw(0,40,40,0)+50random(10)+30"
206	c3->funType = FUNTYPE_SAW;	×
207	c3->multiple = 1;	×
208	c3->random = 10;	×
209	c3->addend = 50;	×
210	c3->base = 30;	×
211
212	stbInfo->tags = benchArrayInit(2, sizeof(Field));	×
213	for (int i = 0; i < 2; ++i) {	×
214	Field * tag = benchCalloc(1, sizeof(Field), true);	×
215	benchArrayPush(stbInfo->tags, tag);	×
216	}
217	Field * t1 = benchArrayGet(stbInfo->tags, 0);	×
218	Field * t2 = benchArrayGet(stbInfo->tags, 1);	×
219
220	t1->type = TSDB_DATA_TYPE_INT;	×
221	t2->type = TSDB_DATA_TYPE_BINARY;	×
222
223	t1->length = sizeof(int32_t);	×
224	t2->length = 24;	×
225
226	TOOLS_STRNCPY(t1->name, "groupid", TSDB_COL_NAME_LEN + 1);	×
227	TOOLS_STRNCPY(t2->name, "location", TSDB_COL_NAME_LEN + 1);	×
NEW 228	TOOLS_STRNCPY(stbInfo->primaryKeyName, "ts", TSDB_COL_NAME_LEN + 1);	×
229	t1->min = 1;	×
230	t1->max = 100000;	×
231
232
233	stbInfo->insert_interval = 0;	×
234	stbInfo->timestamp_step = 1;	×
235	stbInfo->angle_step = 1;	×
236	stbInfo->interlaceRows = 0;	×
237	stbInfo->childTblCount = DEFAULT_CHILDTABLES;	×
238	stbInfo->childTblLimit = 0;	×
239	stbInfo->childTblOffset = 0;	×
240	stbInfo->autoTblCreating = false;	×
241	stbInfo->childTblExists = false;	×
242	stbInfo->random_data_source = true;	×
243	stbInfo->lineProtocol = TSDB_SML_LINE_PROTOCOL;	×
244
245	stbInfo->insertRows = DEFAULT_INSERT_ROWS;	×
246	stbInfo->disorderRange = DEFAULT_DISORDER_RANGE;	×
247	stbInfo->disorderRatio = 0;	×
248	stbInfo->file_factor = -1;	×
249	stbInfo->delay = -1;	×
250	stbInfo->keep_trying = 0;	×
251	stbInfo->trying_interval = 0;	×
252	}	×
253
254	static void initDatabase() {	×
255	g_arguments->databases = benchArrayInit(1, sizeof(SDataBase));	×
256	SDataBase *database = benchCalloc(1, sizeof(SDataBase), true);	×
257	benchArrayPush(g_arguments->databases, database);	×
258	database = benchArrayGet(g_arguments->databases, 0);	×
259	database->dbName = DEFAULT_DATABASE;	×
260	database->drop = true;	×
261	database->precision = TSDB_TIME_PRECISION_MILLI;	×
262	database->sml_precision = TSDB_SML_TIMESTAMP_MILLI_SECONDS;	×
263	database->cfgs = benchArrayInit(1, sizeof(SDbCfg));	×
264	}	×
265
266	void initArgument() {	×
267	g_arguments = benchCalloc(1, sizeof(SArguments), true);	×
268	if (taos_get_client_info()[0] == '3') {	×
269	g_arguments->taosc_version = 3;	×
270	} else {
271	g_arguments->taosc_version = 2;	×
272	}
273	g_arguments->test_mode = INSERT_TEST;	×
274	g_arguments->demo_mode = true;	×
275	g_arguments->host = NULL;	×
276	g_arguments->port = 0;	×
277	g_arguments->port_inputted = false;	×
278	g_arguments->telnet_tcp_port = TELNET_TCP_PORT;	×
279	g_arguments->user = NULL;	×
280	g_arguments->password = NULL;	×
281	g_arguments->answer_yes = 0;	×
282	g_arguments->debug_print = 0;	×
283	g_arguments->binwidth = DEFAULT_BINWIDTH;	×
284	g_arguments->performance_print = 0;	×
285	g_arguments->output_file = DEFAULT_OUTPUT;	×
286	g_arguments->nthreads = DEFAULT_NTHREADS;	×
287	g_arguments->table_threads = DEFAULT_NTHREADS;	×
288	g_arguments->prepared_rand = DEFAULT_PREPARED_RAND;	×
289	g_arguments->reqPerReq = DEFAULT_REQ_PER_REQ;	×
290	g_arguments->totalChildTables = DEFAULT_CHILDTABLES;	×
291	g_arguments->actualChildTables = 0;	×
292	g_arguments->autoCreatedChildTables = 0;	×
293	g_arguments->existedChildTables = 0;	×
294	g_arguments->chinese = false;	×
295	g_arguments->aggr_func = 0;	×
296	g_arguments->terminate = false;	×
297
298	g_arguments->supplementInsert = false;	×
299	g_arguments->startTimestamp = DEFAULT_START_TIME;	×
300	g_arguments->partialColNum = 0;	×
301
302	g_arguments->keep_trying = 0;	×
303	g_arguments->trying_interval = 0;	×
304	g_arguments->iface = TAOSC_IFACE;	×
305	g_arguments->rest_server_ver_major = -1;	×
306	g_arguments->inputted_vgroups = -1;	×
307
308	g_arguments->mistMode = false;	×
309	g_arguments->connMode = CONN_MODE_INVALID;	×
310
311	initDatabase();	×
312	initStable();	×
313	g_arguments->streams = benchArrayInit(1, sizeof(SSTREAM));	×
314	}	×
315
316	void modifyArgument() {	×
317	SDataBase * database = benchArrayGet(g_arguments->databases, 0);	×
318	SSuperTable *superTable = benchArrayGet(database->superTbls, 0);	×
319
320	superTable->startTimestamp = g_arguments->startTimestamp;	×
321
322	if (0 != g_arguments->partialColNum) {	×
323	superTable->partialColNum = g_arguments->partialColNum;	×
324	}
325
326	for (int i = 0; i < superTable->cols->size; ++i) {	×
327	Field * col = benchArrayGet(superTable->cols, i);	×
328	if (!g_arguments->demo_mode) {	×
329	snprintf(col->name, TSDB_COL_NAME_LEN, "c%d", i);	×
330	col->min = convertDatatypeToDefaultMin(col->type);	×
331	col->max = convertDatatypeToDefaultMax(col->type);	×
332	}
333	if (col->length == 0) {	×
334	col->length = g_arguments->binwidth;	×
335	}
336	}
337
338	for (int i = 0; i < superTable->tags->size; ++i) {	×
339	Field* tag = benchArrayGet(superTable->tags, i);	×
340	if (!g_arguments->demo_mode) {	×
341	snprintf(tag->name, TSDB_COL_NAME_LEN, "t%d", i);	×
342	}
343	if (tag->length == 0) {	×
344	tag->length = g_arguments->binwidth;	×
345	}
346	}
347
348	if (g_arguments->intColumnCount > superTable->cols->size) {	×
349	for (int i = superTable->cols->size;	×
350	i < g_arguments->intColumnCount; ++i) {	×
351	Field * col = benchCalloc(1, sizeof(Field), true);	×
352	benchArrayPush(superTable->cols, col);	×
353	col = benchArrayGet(superTable->cols, i);	×
354	col->type = TSDB_DATA_TYPE_INT;	×
355	col->length = sizeof(int32_t);	×
356	snprintf(col->name, TSDB_COL_NAME_LEN, "c%d", i);	×
357	col->min = convertDatatypeToDefaultMin(col->type);	×
358	col->max = convertDatatypeToDefaultMax(col->type);	×
359	}
360	}
361
362	if (g_arguments->keep_trying) {	×
363	superTable->keep_trying = g_arguments->keep_trying;	×
364	superTable->trying_interval = g_arguments->trying_interval;	×
365	}
366	}	×
367
368	static void queryStableAggrFunc(void sarg) {	×
369	threadInfo pThreadInfo = (threadInfo )sarg;	×
370
371	TAOS *taos = NULL;	×
372	if (REST_IFACE != g_arguments->iface) {	×
373	taos = pThreadInfo->conn->taos;	×
374	}
375	#ifdef LINUX
376	prctl(PR_SET_NAME, "queryStableAggrFunc");	×
377	#endif
378	char *command = benchCalloc(1, TSDB_MAX_ALLOWED_SQL_LEN, false);	×
379	FILE * fp = g_arguments->fpOfInsertResult;	×
380	SDataBase * database = benchArrayGet(g_arguments->databases, 0);	×
381	SSuperTable * stbInfo = benchArrayGet(database->superTbls, 0);	×
382	int64_t totalData = stbInfo->insertRows * stbInfo->childTblCount;	×
383	char **aggreFunc;
384	int n;
385
386	if (g_arguments->demo_mode) {	×
387	aggreFunc = g_aggreFuncDemo;	×
388	n = sizeof(g_aggreFuncDemo) / sizeof(g_aggreFuncDemo[0]);	×
389	} else {
390	aggreFunc = g_aggreFunc;	×
391	n = sizeof(g_aggreFunc) / sizeof(g_aggreFunc[0]);	×
392	}
393
394	infoPrint("total Data: %" PRId64 "\n", totalData);	×
395	if (fp) {	×
396	fprintf(fp, "Querying On %" PRId64 " records:\n", totalData);	×
397	}
398	for (int j = 0; j < n; j++) {	×
399	char condition[COND_BUF_LEN] = "\0";	×
400	char tempS[LARGE_BUFF_LEN] = "\0";	×
401	int64_t m = 10 < stbInfo->childTblCount ? 10 : stbInfo->childTblCount;	×
402	for (int64_t i = 1; i <= m; i++) {	×
403	if (i == 1) {	×
404	if (g_arguments->demo_mode) {	×
405	snprintf(tempS, LARGE_BUFF_LEN,	×
406	"groupid = %" PRId64, i);
407	} else {
408	snprintf(tempS, LARGE_BUFF_LEN,	×
409	"t0 = %" PRId64, i);
410	}
411	} else {
412	if (g_arguments->demo_mode) {	×
413	snprintf(tempS, LARGE_BUFF_LEN,	×
414	" or groupid = %" PRId64 " ", i);
415	} else {
416	snprintf(tempS, LARGE_BUFF_LEN,	×
417	" or t0 = %" PRId64 " ", i);
418	}
419	}
420	strncat(condition, tempS, COND_BUF_LEN - 1);	×
421	snprintf(command, TSDB_MAX_ALLOWED_SQL_LEN,	×
422	"SELECT %s FROM %s.meters WHERE %s",
423	aggreFunc[j], database->dbName,	×
424	condition);
425	if (fp) {	×
426	fprintf(fp, "%s\n", command);	×
427	}
428	double t = (double)toolsGetTimestampUs();	×
429	int32_t code = -1;	×
430	if (REST_IFACE == g_arguments->iface) {	×
431	code = postProcessSql(command, NULL, 0, REST_IFACE,	×
432	0, g_arguments->port, 0,	×
433	pThreadInfo->sockfd, NULL);
434	} else {
435	TAOS_RES *res = taos_query(taos, command);	×
436	code = taos_errno(res);	×
437	if (code != 0) {	×
438	printErrCmdCodeStr(command, code, res);	×
439	free(command);	×
440	return NULL;	×
441	}
442	int count = 0;	×
443	while (taos_fetch_row(res) != NULL) {	×
444	count++;	×
445	}
446	taos_free_result(res);	×
447	}
448	t = toolsGetTimestampUs() - t;	×
449	if (fp) {	×
450	fprintf(fp, "\| Speed: %12.2f(per s) \| Latency: %.4f(ms) \|\n",	×
451	totalData / (t / 1000), t);	×
452	}
453	infoPrint("%s took %.6f second(s)\n\n", command,	×
454	t / 1000000);
455	}
456	}
457	free(command);	×
458	return NULL;	×
459	}
460
461	static void queryNtableAggrFunc(void sarg) {	×
462	threadInfo pThreadInfo = (threadInfo )sarg;	×
463	TAOS * taos = NULL;	×
464	if (pThreadInfo->conn) {	×
465	taos = pThreadInfo->conn->taos;	×
466	}
467	#ifdef LINUX
468	prctl(PR_SET_NAME, "queryNtableAggrFunc");	×
469	#endif
470	char * command = benchCalloc(1, TSDB_MAX_ALLOWED_SQL_LEN, false);	×
471	FILE * fp = g_arguments->fpOfInsertResult;	×
472	SDataBase * database = benchArrayGet(g_arguments->databases, 0);	×
473	SSuperTable * stbInfo = benchArrayGet(database->superTbls, 0);	×
474	int64_t totalData = stbInfo->childTblCount * stbInfo->insertRows;	×
475	char **aggreFunc;
476	int n;
477
478	if (g_arguments->demo_mode) {	×
479	aggreFunc = g_aggreFuncDemo;	×
480	n = sizeof(g_aggreFuncDemo) / sizeof(g_aggreFuncDemo[0]);	×
481	} else {
482	aggreFunc = g_aggreFunc;	×
483	n = sizeof(g_aggreFunc) / sizeof(g_aggreFunc[0]);	×
484	}
485
486	infoPrint("totalData: %" PRId64 "\n", totalData);	×
487	if (fp) {	×
488	fprintf(fp,	×
489	"\| QFunctions \| QRecords \| QSpeed(R/s) \| "
490	"QLatency(ms) \|\n");
491	}
492
493	for (int j = 0; j < n; j++) {	×
494	double totalT = 0;	×
495	uint64_t count = 0;	×
496	for (int64_t i = 0; i < stbInfo->childTblCount; i++) {	×
497	snprintf(command,	×
498	TSDB_MAX_ALLOWED_SQL_LEN,
499	g_arguments->escape_character	×
500	? "SELECT %s FROM `%s`.`%s%" PRId64 "` WHERE ts>= %" PRIu64
501	: "SELECT %s FROM %s.%s%" PRId64 " WHERE ts>= %" PRIu64 ,
502	aggreFunc[j],	×
503	database->dbName,
504	stbInfo->childTblPrefix, i,
505	(uint64_t) DEFAULT_START_TIME);
506	double t = (double)toolsGetTimestampUs();	×
507	int32_t code = -1;	×
508	if (REST_IFACE == g_arguments->iface) {	×
509	code = postProcessSql(command, NULL, 0, REST_IFACE,	×
510	0, g_arguments->port, 0,	×
511	pThreadInfo->sockfd, NULL);
512	} else {
513	TAOS_RES *res = taos_query(taos, command);	×
514	code = taos_errno(res);	×
515	if (code != 0) {	×
516	printErrCmdCodeStr(command, code, res);	×
517	free(command);	×
518	return NULL;	×
519	}
520	while (taos_fetch_row(res) != NULL) {	×
521	count++;	×
522	}
523	taos_free_result(res);	×
524	}
525
526	t = toolsGetTimestampUs() - t;	×
527	totalT += t;	×
528	}
529	if (fp) {	×
530	fprintf(fp, "\|%10s \| %" PRId64 " \| %12.2f \| %10.2f \|\n",	×
531	(aggreFunc[j][0] == '*')	×
532	?(" * "):(aggreFunc[j]), totalData,	×
533	(double)(stbInfo->childTblCount*stbInfo->insertRows)/totalT,	×
534	totalT / 1000000);
535	}
536	infoPrint("<%s> took %.6f second(s)\n", command,	×
537	totalT / 1000000);
538	}
539	free(command);	×
540	return NULL;	×
541	}
542
543	void queryAggrFunc() {	×
544	pthread_t read_id;
545	threadInfo *pThreadInfo = benchCalloc(1, sizeof(threadInfo), false);	×
546	if (NULL == pThreadInfo) {	×
547	errorPrint("%s() failed to allocate memory\n", __func__);	×
548	return;	×
549	}
550	SDataBase * database = benchArrayGet(g_arguments->databases, 0);	×
551	if (NULL == database) {	×
552	errorPrint("%s() failed to get database\n", __func__);	×
553	free(pThreadInfo);	×
554	return;	×
555	}
556	SSuperTable * stbInfo = benchArrayGet(database->superTbls, 0);	×
557	if (NULL == stbInfo) {	×
558	errorPrint("%s() failed to get super table\n", __func__);	×
559	free(pThreadInfo);	×
560	return;	×
561	}
562
563	// REST
564	if (REST_IFACE != g_arguments->iface) {	×
565	pThreadInfo->conn = initBenchConn();	×
566	if (pThreadInfo->conn == NULL) {	×
567	errorPrint("%s() failed to init connection\n", __func__);	×
568	free(pThreadInfo);	×
569	return;	×
570	}
571	} else {
572	pThreadInfo->sockfd = createSockFd();	×
573	if (pThreadInfo->sockfd < 0) {	×
574	free(pThreadInfo);	×
575	return;	×
576	}
577	}
578	if (stbInfo->use_metric) {	×
579	pthread_create(&read_id, NULL, queryStableAggrFunc, pThreadInfo);	×
580	} else {
581	pthread_create(&read_id, NULL, queryNtableAggrFunc, pThreadInfo);	×
582	}
583	pthread_join(read_id, NULL);	×
584	// REST
585	if (REST_IFACE != g_arguments->iface) {	×
586	closeBenchConn(pThreadInfo->conn);	×
587	} else {
588	if (pThreadInfo->sockfd) {	×
589	destroySockFd(pThreadInfo->sockfd);	×
590	}
591	}
592	free(pThreadInfo);	×
593	}

taosdata / TDengine / #4582

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous