#3903

Committed 24 Apr 2025 11:36AM UTC coverage: 55.307% (+0.09%) from 55.213%

Build # #3903

Build Type

push

travis-ci

Committed by

happyguoxy

Commit Message

Sync branches at 2025-04-24 19:35

Run Details

175024 of 316459 relevant lines covered (55.31%)

1151858.11 hits per line

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

56.57

/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);

    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) {	21✔
40	Field* field = benchCalloc(1, sizeof(Field), true);	21✔
41	benchArrayPush(fields, field);	21✔
42	field = benchArrayGet(fields, index);	21✔
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);	21✔
50	if (!reti) {	21✔
51	reti = regexec(&regex, token, 3, pmatch, 0);	21✔
52	if (!reti) {	21✔
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);	21✔
64	}
65
66	// DECIMAL
67	reti = regcomp(&regex, "^DECIMAL\\s\\(\\s(-?[0-9]+)\\s,\\s(-?[0-9]+)\\s*\\)$", REG_ICASE \| REG_EXTENDED);	21✔
68	if (!reti) {	21✔
69	reti = regexec(&regex, token, 3, pmatch, 0);	21✔
70	if (!reti) {	21✔
71	char precision[DECIMAL_BUFF_LEN] = {0};	12✔
72	char scale[DECIMAL_BUFF_LEN] = {0};	12✔
73	strncpy(precision, token + pmatch[1].rm_so, pmatch[1].rm_eo - pmatch[1].rm_so);	12✔
74	precision[pmatch[1].rm_eo - pmatch[1].rm_so] = '\0';	12✔
75	strncpy(scale, token + pmatch[2].rm_so, pmatch[2].rm_eo - pmatch[2].rm_so);	12✔
76	scale[pmatch[2].rm_eo - pmatch[2].rm_so] = '\0';	12✔
77
78	int p = atoi(precision), s = atoi(scale);	12✔
79	if (p > TSDB_DECIMAL128_MAX_PRECISION \|\| p <= 0) {	12✔
80	errorPrint("Invalid precision value of decimal type in args, precision: %d, scale: %d\n", p, s);	3✔
81	exit(EXIT_FAILURE);	3✔
82	}
83	if (s < 0 \|\| s > p) {	9✔
84	errorPrint("Invalid scale value of decimal type in args, precision: %d, scale: %d\n", p, s);	2✔
85	exit(EXIT_FAILURE);	2✔
86	}
87	field->precision = p;	7✔
88	field->scale = s;	7✔
89	field->type = convertStringToDatatype("DECIMAL", 0, &field->precision);	7✔
90	field->length = convertTypeToLength(field->type);	7✔
91	regfree(&regex);	7✔
92
93	if (field->type == TSDB_DATA_TYPE_DECIMAL) {	7✔
94	getDecimal128DefaultMax(p, s, &field->decMax.dec128);	1✔
95	getDecimal128DefaultMin(p, s, &field->decMin.dec128);	1✔
96	} else {
97	getDecimal64DefaultMax(p, s, &field->decMax.dec64);	6✔
98	getDecimal64DefaultMin(p, s, &field->decMin.dec64);	6✔
99	}
100
101	goto SET_PROPS;	7✔
102	}
103	regfree(&regex);	9✔
104	}
105
106	// other
107	field->type = convertStringToDatatype(token, 0, NULL);	9✔
108	field->length = convertTypeToLength(field->type);	9✔
109
110	SET_PROPS:	16✔
111	field->min = convertDatatypeToDefaultMin(field->type);	16✔
112	field->max = convertDatatypeToDefaultMax(field->type);	16✔
113	snprintf(field->name, TSDB_COL_NAME_LEN, isTag ? "t%d" : "c%d", index);	16✔
114	}	16✔
115
116
117	void parseFieldDatatype(char* dataType, BArray* fields, bool isTag) {	7✔
118	benchArrayClear(fields);	7✔
119	if (strstr(dataType, ",") == NULL) {	7✔
120	processSingleToken(dataType, fields, 0, isTag);	×
121	} else {
122	char* dupStr = strdup(dataType);	7✔
123	char* start = dupStr;	7✔
124	char* current = start;	7✔
125	int bracketDepth = 0;	7✔
126	int index = 0;	7✔
127
128	while (*current != '\0') {	210✔
129	if (*current == '(') {	203✔
130	bracketDepth++;	12✔
131	} else if (*current == ')') {	191✔
132	if (bracketDepth > 0) bracketDepth--;	12✔
133	else {
134	errorPrint("Unbalanced parentheses in data type: %s\n", dataType);	×
135	exit(EXIT_FAILURE);	×
136	}
137	} else if (*current == ',' && bracketDepth == 0) {	179✔
138	*current = '\0';	14✔
139	processSingleToken(start, fields, index++, isTag);	14✔
140	start = current + 1;	14✔
141	}
142	current++;	203✔
143	}
144
145	if (start < current) {	7✔
146	processSingleToken(start, fields, index, isTag);	7✔
147	}
148	tmfree(dupStr);	2✔
149	}
150	}	2✔
151
152
153	static void initStable() {	93✔
154	SDataBase *database = benchArrayGet(g_arguments->databases, 0);	93✔
155	database->superTbls = benchArrayInit(1, sizeof(SSuperTable));	93✔
156	SSuperTable * stbInfo = benchCalloc(1, sizeof(SSuperTable), true);	93✔
157	benchArrayPush(database->superTbls, stbInfo);	93✔
158	stbInfo = benchArrayGet(database->superTbls, 0);	93✔
159	stbInfo->iface = TAOSC_IFACE;	93✔
160	stbInfo->stbName = "meters";	93✔
161	stbInfo->childTblPrefix = DEFAULT_TB_PREFIX;	93✔
162	stbInfo->use_metric = 1;	93✔
163	stbInfo->max_sql_len = TSDB_MAX_ALLOWED_SQL_LEN;	93✔
164	stbInfo->cols = benchArrayInit(3, sizeof(Field));	93✔
165	for (int i = 0; i < 3; ++i) {	372✔
166	Field *col = benchCalloc(1, sizeof(Field), true);	279✔
167	benchArrayPush(stbInfo->cols, col);	279✔
168	}
169	Field * c1 = benchArrayGet(stbInfo->cols, 0);	93✔
170	Field * c2 = benchArrayGet(stbInfo->cols, 1);	93✔
171	Field * c3 = benchArrayGet(stbInfo->cols, 2);	93✔
172
173	c1->type = TSDB_DATA_TYPE_FLOAT;	93✔
174	c2->type = TSDB_DATA_TYPE_INT;	93✔
175	c3->type = TSDB_DATA_TYPE_FLOAT;	93✔
176
177	c1->length = sizeof(float);	93✔
178	c2->length = sizeof(int32_t);	93✔
179	c3->length = sizeof(float);	93✔
180
181	TOOLS_STRNCPY(c1->name, "current", TSDB_COL_NAME_LEN + 1);	93✔
182	TOOLS_STRNCPY(c2->name, "voltage", TSDB_COL_NAME_LEN + 1);	93✔
183	TOOLS_STRNCPY(c3->name, "phase", TSDB_COL_NAME_LEN + 1);	93✔
184
185	c1->min = 9;	93✔
186	c1->max = 10;	93✔
187	//fun = "4sin(x)+10random(5)+10"
188	c1->funType = FUNTYPE_SIN;	93✔
189	c1->multiple = 4;	93✔
190	c1->random = 5;	93✔
191	c1->addend = 10;	93✔
192	c1->base = 10;	93✔
193
194	c2->min = 110;	93✔
195	c2->max = 119;	93✔
196	//fun = "1square(0,60,50,0)+100random(20)+120"
197	c2->funType = FUNTYPE_SQUARE;	93✔
198	c2->multiple = 1;	93✔
199	c2->random = 20;	93✔
200	c2->addend = 100;	93✔
201	c2->base = 120;	93✔
202
203	c3->min = 115;	93✔
204	c3->max = 125;	93✔
205	// fun = "1saw(0,40,40,0)+50random(10)+30"
206	c3->funType = FUNTYPE_SAW;	93✔
207	c3->multiple = 1;	93✔
208	c3->random = 10;	93✔
209	c3->addend = 50;	93✔
210	c3->base = 30;	93✔
211
212	stbInfo->tags = benchArrayInit(2, sizeof(Field));	93✔
213	for (int i = 0; i < 2; ++i) {	279✔
214	Field * tag = benchCalloc(1, sizeof(Field), true);	186✔
215	benchArrayPush(stbInfo->tags, tag);	186✔
216	}
217	Field * t1 = benchArrayGet(stbInfo->tags, 0);	93✔
218	Field * t2 = benchArrayGet(stbInfo->tags, 1);	93✔
219
220	t1->type = TSDB_DATA_TYPE_INT;	93✔
221	t2->type = TSDB_DATA_TYPE_BINARY;	93✔
222
223	t1->length = sizeof(int32_t);	93✔
224	t2->length = 24;	93✔
225
226	TOOLS_STRNCPY(t1->name, "groupid", TSDB_COL_NAME_LEN + 1);	93✔
227	TOOLS_STRNCPY(t2->name, "location", TSDB_COL_NAME_LEN + 1);	93✔
228
229	t1->min = 1;	93✔
230	t1->max = 100000;	93✔
231
232
233	stbInfo->insert_interval = 0;	93✔
234	stbInfo->timestamp_step = 1;	93✔
235	stbInfo->angle_step = 1;	93✔
236	stbInfo->interlaceRows = 0;	93✔
237	stbInfo->childTblCount = DEFAULT_CHILDTABLES;	93✔
238	stbInfo->childTblLimit = 0;	93✔
239	stbInfo->childTblOffset = 0;	93✔
240	stbInfo->autoTblCreating = false;	93✔
241	stbInfo->childTblExists = false;	93✔
242	stbInfo->random_data_source = true;	93✔
243	stbInfo->lineProtocol = TSDB_SML_LINE_PROTOCOL;	93✔
244
245	stbInfo->insertRows = DEFAULT_INSERT_ROWS;	93✔
246	stbInfo->disorderRange = DEFAULT_DISORDER_RANGE;	93✔
247	stbInfo->disorderRatio = 0;	93✔
248	stbInfo->file_factor = -1;	93✔
249	stbInfo->delay = -1;	93✔
250	stbInfo->keep_trying = 0;	93✔
251	stbInfo->trying_interval = 0;	93✔
252	}	93✔
253
254	static void initDatabase() {	93✔
255	g_arguments->databases = benchArrayInit(1, sizeof(SDataBase));	93✔
256	SDataBase *database = benchCalloc(1, sizeof(SDataBase), true);	93✔
257	benchArrayPush(g_arguments->databases, database);	93✔
258	database = benchArrayGet(g_arguments->databases, 0);	93✔
259	database->dbName = DEFAULT_DATABASE;	93✔
260	database->drop = true;	93✔
261	database->precision = TSDB_TIME_PRECISION_MILLI;	93✔
262	database->sml_precision = TSDB_SML_TIMESTAMP_MILLI_SECONDS;	93✔
263	database->cfgs = benchArrayInit(1, sizeof(SDbCfg));	93✔
264	}	93✔
265
266	void initArgument() {	93✔
267	g_arguments = benchCalloc(1, sizeof(SArguments), true);	93✔
268	if (taos_get_client_info()[0] == '3') {	93✔
269	g_arguments->taosc_version = 3;	93✔
270	} else {
271	g_arguments->taosc_version = 2;	×
272	}
273	g_arguments->test_mode = INSERT_TEST;	93✔
274	g_arguments->demo_mode = true;	93✔
275	g_arguments->host = NULL;	93✔
276	g_arguments->port = 0;	93✔
277	g_arguments->port_inputted = false;	93✔
278	g_arguments->telnet_tcp_port = TELNET_TCP_PORT;	93✔
279	g_arguments->user = NULL;	93✔
280	g_arguments->password = NULL;	93✔
281	g_arguments->answer_yes = 0;	93✔
282	g_arguments->debug_print = 0;	93✔
283	g_arguments->binwidth = DEFAULT_BINWIDTH;	93✔
284	g_arguments->performance_print = 0;	93✔
285	g_arguments->output_file = DEFAULT_OUTPUT;	93✔
286	g_arguments->nthreads = DEFAULT_NTHREADS;	93✔
287	g_arguments->table_threads = DEFAULT_NTHREADS;	93✔
288	g_arguments->prepared_rand = DEFAULT_PREPARED_RAND;	93✔
289	g_arguments->reqPerReq = DEFAULT_REQ_PER_REQ;	93✔
290	g_arguments->totalChildTables = DEFAULT_CHILDTABLES;	93✔
291	g_arguments->actualChildTables = 0;	93✔
292	g_arguments->autoCreatedChildTables = 0;	93✔
293	g_arguments->existedChildTables = 0;	93✔
294	g_arguments->chinese = false;	93✔
295	g_arguments->aggr_func = 0;	93✔
296	g_arguments->terminate = false;	93✔
297
298	g_arguments->supplementInsert = false;	93✔
299	g_arguments->startTimestamp = DEFAULT_START_TIME;	93✔
300	g_arguments->partialColNum = 0;	93✔
301
302	g_arguments->keep_trying = 0;	93✔
303	g_arguments->trying_interval = 0;	93✔
304	g_arguments->iface = TAOSC_IFACE;	93✔
305	g_arguments->rest_server_ver_major = -1;	93✔
306	g_arguments->inputted_vgroups = -1;	93✔
307
308	g_arguments->mistMode = false;	93✔
309	g_arguments->connMode = CONN_MODE_INVALID;	93✔
310
311	initDatabase();	93✔
312	initStable();	93✔
313	g_arguments->streams = benchArrayInit(1, sizeof(SSTREAM));	93✔
314	}	93✔
315
316	void modifyArgument() {	20✔
317	SDataBase * database = benchArrayGet(g_arguments->databases, 0);	20✔
318	SSuperTable *superTable = benchArrayGet(database->superTbls, 0);	20✔
319
320	superTable->startTimestamp = g_arguments->startTimestamp;	20✔
321
322	if (0 != g_arguments->partialColNum) {	20✔
323	superTable->partialColNum = g_arguments->partialColNum;	×
324	}
325
326	for (int i = 0; i < superTable->cols->size; ++i) {	80✔
327	Field * col = benchArrayGet(superTable->cols, i);	60✔
328	if (!g_arguments->demo_mode) {	60✔
329	snprintf(col->name, TSDB_COL_NAME_LEN, "c%d", i);	9✔
330	col->min = convertDatatypeToDefaultMin(col->type);	9✔
331	col->max = convertDatatypeToDefaultMax(col->type);	9✔
332	}
333	if (col->length == 0) {	60✔
334	col->length = g_arguments->binwidth;	1✔
335	}
336	}
337
338	for (int i = 0; i < superTable->tags->size; ++i) {	58✔
339	Field* tag = benchArrayGet(superTable->tags, i);	38✔
340	if (!g_arguments->demo_mode) {	38✔
341	snprintf(tag->name, TSDB_COL_NAME_LEN, "t%d", i);	4✔
342	}
343	if (tag->length == 0) {	38✔
344	tag->length = g_arguments->binwidth;	×
345	}
346	}
347
348	if (g_arguments->intColumnCount > superTable->cols->size) {	20✔
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) {	20✔
363	superTable->keep_trying = g_arguments->keep_trying;	1✔
364	superTable->trying_interval = g_arguments->trying_interval;	1✔
365	}
366	}	20✔
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 / #3903

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