taosdemo.c 203.5 KB
Newer Older
S
TD-1057  
Shengliang Guan 已提交
1
/*
H
hzcheng 已提交
2 3 4 5 6 7 8 9 10 11 12 13 14 15
 * 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 GNU Affero General Public License, version 3
 * or later ("AGPL"), 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.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

16 17 18 19 20

/*
   when in some thread query return error, thread don't exit, but return, otherwise coredump in other thread.
*/

S
slguan 已提交
21
#define _GNU_SOURCE
22 23
#define CURL_STATICLIB

S
TD-1057  
Shengliang Guan 已提交
24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45
#ifdef LINUX
  #include <argp.h>
  #include <assert.h>
  #include <inttypes.h>
  #ifndef _ALPINE
    #include <error.h>
  #endif
  #include <pthread.h>
  #include <semaphore.h>
  #include <stdbool.h>
  #include <stdio.h>
  #include <string.h>
  #include <sys/time.h>
  #include <time.h>
  #include <unistd.h>
  #include <wordexp.h>
  #include <regex.h>
#else  
  #include <assert.h>
  #include <regex.h>
  #include <stdio.h>
#endif  
H
hzcheng 已提交
46

47
#include <stdlib.h>
48 49
#include "cJSON.h"

50
#include "os.h"
H
hzcheng 已提交
51
#include "taos.h"
52
#include "taoserror.h"
H
Hui Li 已提交
53
#include "tutil.h"
S
slguan 已提交
54

55 56 57
#define REQ_EXTRA_BUF_LEN   1024
#define RESP_BUF_LEN        4096

S
slguan 已提交
58 59
extern char configDir[];

60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90
#define INSERT_JSON_NAME      "insert.json"
#define QUERY_JSON_NAME       "query.json"
#define SUBSCRIBE_JSON_NAME   "subscribe.json"

#define INSERT_MODE        0
#define QUERY_MODE         1
#define SUBSCRIBE_MODE     2

#define MAX_SQL_SIZE       65536
#define BUFFER_SIZE        (65536*2)
#define MAX_DB_NAME_SIZE   64
#define MAX_TB_NAME_SIZE   64
#define MAX_DATA_SIZE      16000
#define MAX_NUM_DATATYPE   10
#define OPT_ABORT          1 /* –abort */
#define STRING_LEN         60000
#define MAX_PREPARED_RAND  1000000
#define MAX_FILE_NAME_LEN  256

#define   MAX_SAMPLES_ONCE_FROM_FILE   10000
#define   MAX_NUM_DATATYPE 10

#define   MAX_DB_COUNT           8
#define   MAX_SUPER_TABLE_COUNT  8
#define   MAX_COLUMN_COUNT       1024
#define   MAX_TAG_COUNT          128

#define   MAX_QUERY_SQL_COUNT    10
#define   MAX_QUERY_SQL_LENGTH   256

#define   MAX_DATABASE_COUNT     256
91
#define INPUT_BUF_LEN   256
92

93 94
#define DEFAULT_TIMESTAMP_STEP  10

95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111
typedef enum CREATE_SUB_TALBE_MOD_EN {
  PRE_CREATE_SUBTBL,
  AUTO_CREATE_SUBTBL,
  NO_CREATE_SUBTBL
} CREATE_SUB_TALBE_MOD_EN;
  
typedef enum TALBE_EXISTS_EN {
  TBL_ALREADY_EXISTS,
  TBL_NO_EXISTS,
  TBL_EXISTS_BUTT
} TALBE_EXISTS_EN;

enum MODE {
  SYNC, 
  ASYNC,
  MODE_BUT
};
112

113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166
enum QUERY_TYPE {
  NO_INSERT_TYPE,
  INSERT_TYPE, 
  QUERY_TYPE_BUT
} ;  

enum _show_db_index {
  TSDB_SHOW_DB_NAME_INDEX,
  TSDB_SHOW_DB_CREATED_TIME_INDEX,
  TSDB_SHOW_DB_NTABLES_INDEX,
  TSDB_SHOW_DB_VGROUPS_INDEX,
  TSDB_SHOW_DB_REPLICA_INDEX,
  TSDB_SHOW_DB_QUORUM_INDEX,  
  TSDB_SHOW_DB_DAYS_INDEX,
  TSDB_SHOW_DB_KEEP_INDEX,
  TSDB_SHOW_DB_CACHE_INDEX,
  TSDB_SHOW_DB_BLOCKS_INDEX,
  TSDB_SHOW_DB_MINROWS_INDEX,
  TSDB_SHOW_DB_MAXROWS_INDEX,
  TSDB_SHOW_DB_WALLEVEL_INDEX,
  TSDB_SHOW_DB_FSYNC_INDEX,
  TSDB_SHOW_DB_COMP_INDEX,
  TSDB_SHOW_DB_CACHELAST_INDEX,
  TSDB_SHOW_DB_PRECISION_INDEX,  
  TSDB_SHOW_DB_UPDATE_INDEX,
  TSDB_SHOW_DB_STATUS_INDEX,
  TSDB_MAX_SHOW_DB
};

// -----------------------------------------SHOW TABLES CONFIGURE -------------------------------------
enum _show_stables_index {
  TSDB_SHOW_STABLES_NAME_INDEX,
  TSDB_SHOW_STABLES_CREATED_TIME_INDEX,
  TSDB_SHOW_STABLES_COLUMNS_INDEX,
  TSDB_SHOW_STABLES_METRIC_INDEX,  
  TSDB_SHOW_STABLES_UID_INDEX,  
  TSDB_SHOW_STABLES_TID_INDEX,
  TSDB_SHOW_STABLES_VGID_INDEX,  
  TSDB_MAX_SHOW_STABLES
};  
enum _describe_table_index {
  TSDB_DESCRIBE_METRIC_FIELD_INDEX,
  TSDB_DESCRIBE_METRIC_TYPE_INDEX,
  TSDB_DESCRIBE_METRIC_LENGTH_INDEX,
  TSDB_DESCRIBE_METRIC_NOTE_INDEX,
  TSDB_MAX_DESCRIBE_METRIC
};

typedef struct {
  char field[TSDB_COL_NAME_LEN + 1];
  char type[16];
  int length;
  char note[128];
} SColDes;
H
hzcheng 已提交
167 168

/* Used by main to communicate with parse_opt. */
169 170
typedef struct SArguments_S {
  char *   metaFile;
171
  int      test_mode;
172 173 174 175 176 177 178 179 180 181
  char *   host;
  uint16_t port;
  char *   user;
  char *   password;
  char *   database;
  int      replica;
  char *   tb_prefix;
  char *   sqlFile;
  bool     use_metric;
  bool     insert_only;
182
  bool     answer_yes;
183
  bool     debug_print;
184
  bool     verbose_print;
185 186 187 188 189 190
  char *   output_file;
  int      mode;
  char *   datatype[MAX_NUM_DATATYPE + 1];
  int      len_of_binary;
  int      num_of_CPR;
  int      num_of_threads;
191
  int      insert_interval;
192
  int      num_of_RPR;
193
  int      max_sql_len;
194 195 196
  int      num_of_tables;
  int      num_of_DPT;
  int      abort;
197 198
  int      disorderRatio;
  int      disorderRange;
199 200
  int      method_of_delete;
  char **  arg_list;
201 202 203 204 205 206 207 208 209 210
} SArguments;

typedef struct SColumn_S {
  char  field[TSDB_COL_NAME_LEN + 1];
  char  dataType[MAX_TB_NAME_SIZE];
  int   dataLen;
  char  note[128];  
} StrColumn;

typedef struct SSuperTable_S {
211
  char         sTblName[MAX_TB_NAME_SIZE+1];
212 213 214 215 216 217
  int          childTblCount;
  bool         superTblExists;    // 0: no, 1: yes
  bool         childTblExists;    // 0: no, 1: yes  
  int          batchCreateTableNum;  // 0: no batch,  > 0: batch table number in one sql
  int8_t       autoCreateTable;                  // 0: create sub table, 1: auto create sub table
  char         childTblPrefix[MAX_TB_NAME_SIZE];
218
  char         dataSource[MAX_TB_NAME_SIZE+1];  // rand_gen or sample
219
  char         insertMode[MAX_TB_NAME_SIZE];  // taosc, restful
220 221
  int           childTblLimit;
  int           childTblOffset;
222 223 224 225 226 227 228

  int          multiThreadWriteOneTbl;   // 0: no, 1: yes
  int          numberOfTblInOneSql;      // 0/1: one table, > 1: number of tbl
  int          rowsPerTbl;               // 
  int          disorderRatio;            // 0: no disorder, >0: x%
  int          disorderRange;            // ms or us by database precision
  int          maxSqlLen;                // 
229 230

  int          insertInterval;          // insert interval, will override global insert interval
231 232 233 234
  int64_t      insertRows;               // 0: no limit
  int          timeStampStep;
  char         startTimestamp[MAX_TB_NAME_SIZE];  // 
  char         sampleFormat[MAX_TB_NAME_SIZE];  // csv, json
235 236
  char         sampleFile[MAX_FILE_NAME_LEN+1];
  char         tagsFile[MAX_FILE_NAME_LEN+1];
237 238 239 240 241 242 243

  int          columnCount;
  StrColumn    columns[MAX_COLUMN_COUNT];
  int          tagCount;
  StrColumn    tags[MAX_TAG_COUNT];

  char*        childTblName;
244
  char*        colsOfCreateChildTable;
245 246 247 248 249 250 251 252 253 254 255 256 257 258
  int          lenOfOneRow;
  int          lenOfTagOfOneRow;

  char*        sampleDataBuf;
  int          sampleDataBufSize;
  //int          sampleRowCount;
  //int          sampleUsePos;

  int          tagSource;    // 0: rand, 1: tag sample
  char*        tagDataBuf;
  int          tagSampleCount;
  int          tagUsePos;

  // statistics  
259
  int64_t    totalInsertRows;
260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290
  int64_t    totalAffectedRows;
} SSuperTable;

typedef struct {
  char     name[TSDB_DB_NAME_LEN + 1];
  char     create_time[32];
  int32_t  ntables;
  int32_t  vgroups;  
  int16_t  replica;
  int16_t  quorum;
  int16_t  days;  
  char     keeplist[32];
  int32_t  cache; //MB
  int32_t  blocks;
  int32_t  minrows;
  int32_t  maxrows;
  int8_t   wallevel;
  int32_t  fsync;
  int8_t   comp;
  int8_t   cachelast;
  char     precision[8];   // time resolution
  int8_t   update;
  char     status[16];
} SDbInfo;

typedef struct SDbCfg_S { 
//  int       maxtablesPerVnode;
  int       minRows; 
  int       maxRows;
  int       comp;
  int       walLevel;
291
  int       cacheLast;
292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311
  int       fsync;  
  int       replica;
  int       update;
  int       keep;
  int       days;
  int       cache;
  int       blocks;
  int       quorum;
  char      precision[MAX_TB_NAME_SIZE];  
} SDbCfg;

typedef struct SDataBase_S {
  char         dbName[MAX_DB_NAME_SIZE];
  int          drop;  // 0: use exists, 1: if exists, drop then new create
  SDbCfg       dbCfg;
  int          superTblCount;
  SSuperTable  superTbls[MAX_SUPER_TABLE_COUNT];
} SDataBase;

typedef struct SDbs_S {
312
  char         cfgDir[MAX_FILE_NAME_LEN+1];
313 314 315 316
  char         host[MAX_DB_NAME_SIZE];
  uint16_t     port;
  char         user[MAX_DB_NAME_SIZE];
  char         password[MAX_DB_NAME_SIZE];
317
  char         resultFile[MAX_FILE_NAME_LEN+1];
318 319 320 321 322 323 324 325 326 327 328
  bool         use_metric;
  bool         insert_only;
  bool         do_aggreFunc;
  bool         queryMode;
  
  int          threadCount;
  int          threadCountByCreateTbl;
  int          dbCount;
  SDataBase    db[MAX_DB_COUNT];

  // statistics
329
  int64_t    totalInsertRows;
330
  int64_t    totalAffectedRows;
331

332 333 334 335 336 337 338 339 340 341
} SDbs;

typedef struct SuperQueryInfo_S {
  int          rate;  // 0: unlimit  > 0   loop/s
  int          concurrent;
  int          sqlCount;
  int          subscribeMode; // 0: sync, 1: async
  int          subscribeInterval; // ms
  int          subscribeRestart;
  int          subscribeKeepProgress;
342 343
  char         sql[MAX_QUERY_SQL_COUNT][MAX_QUERY_SQL_LENGTH+1];  
  char         result[MAX_QUERY_SQL_COUNT][MAX_FILE_NAME_LEN+1];
344 345 346 347
  TAOS_SUB*    tsub[MAX_QUERY_SQL_COUNT];
} SuperQueryInfo;

typedef struct SubQueryInfo_S {
348
  char         sTblName[MAX_TB_NAME_SIZE+1];
349 350 351 352 353 354 355 356 357
  int          rate;  // 0: unlimit  > 0   loop/s
  int          threadCnt;  
  int          subscribeMode; // 0: sync, 1: async
  int          subscribeInterval; // ms
  int          subscribeRestart;
  int          subscribeKeepProgress;
  int          childTblCount;
  char         childTblPrefix[MAX_TB_NAME_SIZE];
  int          sqlCount;
358 359
  char         sql[MAX_QUERY_SQL_COUNT][MAX_QUERY_SQL_LENGTH+1];  
  char         result[MAX_QUERY_SQL_COUNT][MAX_FILE_NAME_LEN+1];
360 361 362 363 364 365
  TAOS_SUB*    tsub[MAX_QUERY_SQL_COUNT];
  
  char*        childTblName;
} SubQueryInfo;

typedef struct SQueryMetaInfo_S {
366
  char         cfgDir[MAX_FILE_NAME_LEN+1];
367 368 369 370
  char         host[MAX_DB_NAME_SIZE];
  uint16_t     port;
  char         user[MAX_DB_NAME_SIZE];
  char         password[MAX_DB_NAME_SIZE];
371
  char         dbName[MAX_DB_NAME_SIZE+1];
372 373 374 375 376 377 378 379 380
  char         queryMode[MAX_TB_NAME_SIZE];  // taosc, restful

  SuperQueryInfo  superQueryInfo;
  SubQueryInfo    subQueryInfo;  
} SQueryMetaInfo;

typedef struct SThreadInfo_S {
  TAOS *taos;
  int threadID;
381
  char db_name[MAX_DB_NAME_SIZE+1];
382 383 384 385 386
  char fp[4096];
  char tb_prefix[MAX_TB_NAME_SIZE];
  int start_table_id;
  int end_table_id;
  int data_of_rate;
387
  uint64_t start_time;  
388 389 390 391 392 393 394
  char* cols;  
  bool  use_metric;  
  SSuperTable* superTblInfo;

  // for async insert
  tsem_t lock_sem;
  int64_t  counter;  
395 396
  uint64_t  st;
  uint64_t  et;
397 398 399
  int64_t  lastTs;

  // statistics
400
  int64_t totalInsertRows;
401
  int64_t totalAffectedRows;
402 403 404 405 406 407 408 409

  // insert delay statistics
  int64_t cntDelay;
  int64_t totalDelay;
  int64_t avgDelay;
  int64_t maxDelay;
  int64_t minDelay;
  
410 411
} threadInfo;

412
#ifdef WINDOWS
413 414
#define _CRT_RAND_S

415
#include <windows.h>
416 417 418 419 420
#include <winsock2.h>

typedef unsigned __int32 uint32_t;

#pragma comment ( lib, "ws2_32.lib" )
421 422
// Some old MinGW/CYGWIN distributions don't define this:
#ifndef ENABLE_VIRTUAL_TERMINAL_PROCESSING
423 424
  #define ENABLE_VIRTUAL_TERMINAL_PROCESSING  0x0004
#endif // ENABLE_VIRTUAL_TERMINAL_PROCESSING
425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459

static HANDLE g_stdoutHandle;
static DWORD g_consoleMode;

void setupForAnsiEscape(void) {
  DWORD mode = 0;
  g_stdoutHandle = GetStdHandle(STD_OUTPUT_HANDLE);

  if(g_stdoutHandle == INVALID_HANDLE_VALUE) {
    exit(GetLastError());
  }

  if(!GetConsoleMode(g_stdoutHandle, &mode)) {
    exit(GetLastError());
  }

  g_consoleMode = mode;

  // Enable ANSI escape codes
  mode |= ENABLE_VIRTUAL_TERMINAL_PROCESSING;

  if(!SetConsoleMode(g_stdoutHandle, mode)) {
    exit(GetLastError());
  }    
}

void resetAfterAnsiEscape(void) {
  // Reset colors
  printf("\x1b[0m");    

  // Reset console mode
  if(!SetConsoleMode(g_stdoutHandle, g_consoleMode)) {
    exit(GetLastError());
  }
}
460 461 462 463 464 465 466 467

int taosRandom()
{
    int number;
    rand_s(&number);

    return number;
}
468 469 470 471 472 473 474
#else
void setupForAnsiEscape(void) {}

void resetAfterAnsiEscape(void) {
  // Reset colors
  printf("\x1b[0m");
}
475 476 477 478 479 480

int taosRandom()
{
    return random();
}

481 482
#endif

483 484 485
static int createDatabases();
static void createChildTables();
static int queryDbExec(TAOS *taos, char *command, int type);
486 487 488 489 490 491 492 493 494 495

/* ************ Global variables ************  */

int32_t  randint[MAX_PREPARED_RAND];
int64_t  randbigint[MAX_PREPARED_RAND];
float    randfloat[MAX_PREPARED_RAND];
double   randdouble[MAX_PREPARED_RAND];
char *aggreFunc[] = {"*", "count(*)", "avg(col0)", "sum(col0)", 
    "max(col0)", "min(col0)", "first(col0)", "last(col0)"};

496
SArguments g_args = {
497 498
                     NULL,            // metaFile
                     0,               // test_mode
499 500 501 502 503 504 505 506 507 508 509 510 511 512 513
                     "127.0.0.1",     // host
                     6030,            // port
                     "root",          // user
                     #ifdef _TD_POWER_ 
                     "powerdb",      // password
                     #else
                     "taosdata",      // password
                     #endif
                     "test",          // database
                     1,               // replica
                     "t",             // tb_prefix
                     NULL,            // sqlFile
                     false,           // use_metric
                     false,           // insert_only
                     false,           // debug_print
514
                     false,           // verbose_print
515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534
                     false,           // answer_yes;
                     "./output.txt",  // output_file
                     0,               // mode : sync or async
                     {
                     "TINYINT",           // datatype
                     "SMALLINT",
                     "INT",
                     "BIGINT",
                     "FLOAT",
                     "DOUBLE",
                     "BINARY",
                     "NCHAR",
                     "BOOL",
                     "TIMESTAMP"
                     },
                     16,              // len_of_binary
                     10,              // num_of_CPR
                     10,              // num_of_connections/thread
                     0,               // insert_interval
                     100,             // num_of_RPR
535
                     TSDB_PAYLOAD_SIZE,  // max_sql_len
536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552
                     10000,           // num_of_tables
                     10000,           // num_of_DPT
                     0,               // abort
                     0,               // disorderRatio
                     1000,            // disorderRange
                     1,               // method_of_delete
                     NULL             // arg_list
};



static SDbs            g_Dbs;
static int             g_totalChildTables = 0;
static SQueryMetaInfo  g_queryInfo;
static FILE *          g_fpOfInsertResult = NULL;

#define debugPrint(fmt, ...) \
553
    do { if (g_args.debug_print || g_args.verbose_print) \
554
      fprintf(stderr, "DEBG: "fmt, __VA_ARGS__); } while(0)
555
#define verbosePrint(fmt, ...) \
556 557
    do { if (g_args.verbose_print) fprintf(stderr, "VERB: "fmt, __VA_ARGS__); } while(0)

558
///////////////////////////////////////////////////
H
hzcheng 已提交
559

560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614
void printHelp() {
  char indent[10] = "        ";
  printf("%s%s%s%s\n", indent, "-f", indent, 
          "The meta file to the execution procedure. Default is './meta.json'.");
  printf("%s%s%s%s\n", indent, "-u", indent, 
          "The TDengine user name to use when connecting to the server. Default is 'root'.");
#ifdef _TD_POWER_
  printf("%s%s%s%s\n", indent, "-P", indent, 
          "The password to use when connecting to the server. Default is 'powerdb'.");
  printf("%s%s%s%s\n", indent, "-c", indent, 
          "Configuration directory. Default is '/etc/power/'.");
#else
  printf("%s%s%s%s\n", indent, "-P", indent, 
          "The password to use when connecting to the server. Default is 'taosdata'.");
  printf("%s%s%s%s\n", indent, "-c", indent, 
          "Configuration directory. Default is '/etc/taos/'.");
#endif  
  printf("%s%s%s%s\n", indent, "-h", indent, 
          "The host to connect to TDengine. Default is localhost.");
  printf("%s%s%s%s\n", indent, "-p", indent, 
          "The TCP/IP port number to use for the connection. Default is 0.");
  printf("%s%s%s%s\n", indent, "-d", indent, 
          "Destination database. Default is 'test'.");
  printf("%s%s%s%s\n", indent, "-a", indent, 
          "Set the replica parameters of the database, Default 1, min: 1, max: 3.");
  printf("%s%s%s%s\n", indent, "-m", indent, 
          "Table prefix name. Default is 't'.");
  printf("%s%s%s%s\n", indent, "-s", indent, "The select sql file.");
  printf("%s%s%s%s\n", indent, "-M", indent, "Use metric flag.");
  printf("%s%s%s%s\n", indent, "-o", indent, 
          "Direct output to the named file. Default is './output.txt'.");
  printf("%s%s%s%s\n", indent, "-q", indent, 
          "Query mode--0: SYNC, 1: ASYNC. Default is SYNC.");
  printf("%s%s%s%s\n", indent, "-b", indent, 
          "The data_type of columns, default: TINYINT,SMALLINT,INT,BIGINT,FLOAT,DOUBLE,BINARY,NCHAR,BOOL,TIMESTAMP.");
  printf("%s%s%s%s\n", indent, "-w", indent, 
          "The length of data_type 'BINARY' or 'NCHAR'. Default is 16");
  printf("%s%s%s%s\n", indent, "-l", indent, 
          "The number of columns per record. Default is 10.");
  printf("%s%s%s%s\n", indent, "-T", indent, 
          "The number of threads. Default is 10.");
  printf("%s%s%s%s\n", indent, "-i", indent, 
          "The sleep time (ms) between insertion. Default is 0.");
  printf("%s%s%s%s\n", indent, "-r", indent, 
          "The number of records per request. Default is 100.");
  printf("%s%s%s%s\n", indent, "-t", indent, 
          "The number of tables. Default is 10000.");
  printf("%s%s%s%s\n", indent, "-n", indent, 
          "The number of records per table. Default is 10000.");
  printf("%s%s%s%s\n", indent, "-x", indent, "Not insert only flag.");
  printf("%s%s%s%s\n", indent, "-y", indent, "Default input yes for prompt.");
  printf("%s%s%s%s\n", indent, "-O", indent, 
          "Insert mode--0: In order, > 0: disorder ratio. Default is in order.");
  printf("%s%s%s%s\n", indent, "-R", indent, 
          "Out of order data's range, ms, default is 1000.");
615 616
  printf("%s%s%s%s\n", indent, "-g", indent, 
          "Print debug info.");
617 618 619 620 621 622
/*    printf("%s%s%s%s\n", indent, "-D", indent, 
          "if elete database if exists. 0: no, 1: yes, default is 1");
          */
}

void parse_args(int argc, char *argv[], SArguments *arguments) {
623
  char **sptr;
624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665
  wordexp_t full_path;

  for (int i = 1; i < argc; i++) {
    if (strcmp(argv[i], "-f") == 0) {
      arguments->metaFile = argv[++i];
    } else if (strcmp(argv[i], "-c") == 0) {
        char *configPath = argv[++i];
      if (wordexp(configPath, &full_path, 0) != 0) {
          fprintf(stderr, "Invalid path %s\n", configPath);
          return;
      }
      taos_options(TSDB_OPTION_CONFIGDIR, full_path.we_wordv[0]);
      wordfree(&full_path);
    } else if (strcmp(argv[i], "-h") == 0) {
      arguments->host = argv[++i];
    } else if (strcmp(argv[i], "-p") == 0) {
      arguments->port = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-u") == 0) {
      arguments->user = argv[++i];
    } else if (strcmp(argv[i], "-P") == 0) {
      arguments->password = argv[++i];
    } else if (strcmp(argv[i], "-o") == 0) {
      arguments->output_file = argv[++i];
    } else if (strcmp(argv[i], "-s") == 0) {
      arguments->sqlFile = argv[++i];
    } else if (strcmp(argv[i], "-q") == 0) {
      arguments->mode = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-T") == 0) {
      arguments->num_of_threads = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-i") == 0) {
      arguments->insert_interval = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-r") == 0) {
      arguments->num_of_RPR = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-t") == 0) {
      arguments->num_of_tables = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-n") == 0) {
      arguments->num_of_DPT = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-d") == 0) {
      arguments->database = argv[++i];
    } else if (strcmp(argv[i], "-l") == 0) {
      arguments->num_of_CPR = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-b") == 0) {
666
      sptr = arguments->datatype;
667 668
      ++i;
      if (strstr(argv[i], ",") == NULL) {
669
        // only one col
670 671 672 673 674 675 676 677 678 679 680 681
        if (strcasecmp(argv[i], "INT")
                && strcasecmp(argv[i], "FLOAT")
                && strcasecmp(argv[i], "TINYINT")
                && strcasecmp(argv[i], "BOOL")
                && strcasecmp(argv[i], "SMALLINT")
                && strcasecmp(argv[i], "BIGINT")
                && strcasecmp(argv[i], "DOUBLE")
                && strcasecmp(argv[i], "BINARY") 
                && strcasecmp(argv[i], "NCHAR")) {
          fprintf(stderr, "Invalid data_type!\n");
          printHelp();
          exit(EXIT_FAILURE);
682
        }
683
        sptr[0] = argv[i];
684
      } else {
685
        // more than one col
686
        int index = 0;
687
        char *dupstr = strdup(argv[i]);
688 689 690
        char *running = dupstr;
        char *token = strsep(&running, ",");
        while (token != NULL) {
691 692 693 694 695 696 697 698 699 700 701 702
          if (strcasecmp(token, "INT")
                  && strcasecmp(token, "FLOAT")
                  && strcasecmp(token, "TINYINT")
                  && strcasecmp(token, "BOOL")
                  && strcasecmp(token, "SMALLINT")
                  && strcasecmp(token, "BIGINT")
                  && strcasecmp(token, "DOUBLE")
                  && strcasecmp(token, "BINARY")
                  && strcasecmp(token, "NCHAR")) {
            fprintf(stderr, "Invalid data_type!\n");
            printHelp();
            exit(EXIT_FAILURE);
H
hzcheng 已提交
703
          }
704 705 706
          sptr[index++] = token;
          token = strsep(&running, ",");
          if (index >= MAX_NUM_DATATYPE) break;
S
TD-1057  
Shengliang Guan 已提交
707
        }
708
        sptr[index] = NULL;
709
      }
710 711 712 713 714
    } else if (strcmp(argv[i], "-w") == 0) {
      arguments->len_of_binary = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-m") == 0) {
      arguments->tb_prefix = argv[++i];
    } else if (strcmp(argv[i], "-M") == 0) {
715
      arguments->use_metric = true;
716
    } else if (strcmp(argv[i], "-x") == 0) {
717
      arguments->insert_only = true;
718
    } else if (strcmp(argv[i], "-y") == 0) {
719
      arguments->answer_yes = true;
720 721
    } else if (strcmp(argv[i], "-g") == 0) {
      arguments->debug_print = true;
722 723
    } else if (strcmp(argv[i], "-gg") == 0) {
      arguments->verbose_print = true;
724 725 726 727 728 729 730 731 732
    } else if (strcmp(argv[i], "-c") == 0) {
      strcpy(configDir, argv[++i]);
    } else if (strcmp(argv[i], "-O") == 0) {
      arguments->disorderRatio = atoi(argv[++i]);
      if (arguments->disorderRatio > 1 
              || arguments->disorderRatio < 0) {
        arguments->disorderRatio = 0;
      } else if (arguments->disorderRatio == 1) {
        arguments->disorderRange = 10;
733
      }
734 735 736 737 738 739
    } else if (strcmp(argv[i], "-R") == 0) {
      arguments->disorderRange = atoi(argv[++i]);
      if (arguments->disorderRange == 1 
              && (arguments->disorderRange > 50 
              || arguments->disorderRange <= 0)) {
        arguments->disorderRange = 10;
740
      }
741 742 743 744
    } else if (strcmp(argv[i], "-a") == 0) {
      arguments->replica = atoi(argv[++i]);
      if (arguments->replica > 3 || arguments->replica < 1) {
          arguments->replica = 1;
745
      }
746 747 748 749 750
    } else if (strcmp(argv[i], "-D") == 0) {
      arguments->method_of_delete = atoi(argv[++i]);
      if (arguments->method_of_delete < 0
              || arguments->method_of_delete > 3) {
        arguments->method_of_delete = 0;
S
TD-1057  
Shengliang Guan 已提交
751
      }
752 753 754 755 756 757 758
    } else if (strcmp(argv[i], "--help") == 0) {
      printHelp();
      exit(0);
    } else {
      fprintf(stderr, "wrong options\n");
      printHelp();
      exit(EXIT_FAILURE);
S
TD-1057  
Shengliang Guan 已提交
759 760
    }
  }
761

762 763
  if (((arguments->debug_print) && (arguments->metaFile == NULL))
          || arguments->verbose_print) {
764
    printf("###################################################################\n");
765
    printf("# meta file:                         %s\n", arguments->metaFile);
766 767 768 769 770 771
    printf("# Server IP:                         %s:%hu\n", 
            arguments->host == NULL ? "localhost" : arguments->host,
            arguments->port );
    printf("# User:                              %s\n", arguments->user);
    printf("# Password:                          %s\n", arguments->password);
    printf("# Use metric:                        %s\n", arguments->use_metric ? "true" : "false");
772 773 774 775 776 777 778 779 780
    if (*(arguments->datatype)) {
        printf("# Specified data type:               ");
        for (int i = 0; i < MAX_NUM_DATATYPE; i++)
            if (arguments->datatype[i])
               printf("%s,", arguments->datatype[i]);
            else
                break;
        printf("\n");
    }
781
    printf("# Insertion interval:                %d\n", arguments->insert_interval);
782
    printf("# Number of records per req:         %d\n", arguments->num_of_RPR);
783
    printf("# Max SQL length:                    %d\n", arguments->max_sql_len);
784
    printf("# Length of Binary:                  %d\n", arguments->len_of_binary);
785 786 787 788 789 790 791 792 793 794 795 796 797
    printf("# Number of Threads:                 %d\n", arguments->num_of_threads);
    printf("# Number of Tables:                  %d\n", arguments->num_of_tables);
    printf("# Number of Data per Table:          %d\n", arguments->num_of_DPT);
    printf("# Database name:                     %s\n", arguments->database);
    printf("# Table prefix:                      %s\n", arguments->tb_prefix);
    if (arguments->disorderRatio) {
      printf("# Data order:                        %d\n", arguments->disorderRatio);
      printf("# Data out of order rate:            %d\n", arguments->disorderRange);
  
    }
    printf("# Delete method:                     %d\n", arguments->method_of_delete);
    printf("# Answer yes when prompt:            %d\n", arguments->answer_yes);
    printf("# Print debug info:                  %d\n", arguments->debug_print);
798
    printf("# Print verbose info:                %d\n", arguments->verbose_print);
799 800 801 802 803 804
    printf("###################################################################\n");
    if (!arguments->answer_yes) {
        printf("Press enter key to continue\n\n");
        (void) getchar();
    }
  }
805
}
H
hzcheng 已提交
806

807 808 809 810 811 812 813 814 815
static bool getInfoFromJsonFile(char* file);
//static int generateOneRowDataForStb(SSuperTable* stbInfo);
//static int getDataIntoMemForStb(SSuperTable* stbInfo);
static void init_rand_data();
void tmfclose(FILE *fp) {
  if (NULL != fp) {
    fclose(fp);
  }
}
816

817 818 819
void tmfree(char *buf) {
  if (NULL != buf) {
    free(buf);
H
Hui Li 已提交
820
  }
821
}
H
Hui Li 已提交
822

823
static int queryDbExec(TAOS *taos, char *command, int type) {
824 825 826
  int i;
  TAOS_RES *res = NULL;
  int32_t   code = -1;
827

828 829 830 831
  for (i = 0; i < 5; i++) {
    if (NULL != res) {
      taos_free_result(res);
      res = NULL;
H
hzcheng 已提交
832
    }
833

834 835 836 837 838
    res = taos_query(taos, command);
    code = taos_errno(res);
    if (0 == code) {
      break;
    }    
H
hzcheng 已提交
839 840
  }

841
  if (code != 0) {
842
    debugPrint("%s() LN%d - command: %s\n", __func__, __LINE__, command);
843 844 845 846
    fprintf(stderr, "Failed to run %s, reason: %s\n", command, taos_errstr(res));
    taos_free_result(res);
    //taos_close(taos);
    return -1;
847
  }
H
hzcheng 已提交
848

849 850 851 852
  if (INSERT_TYPE == type) {
    int affectedRows = taos_affected_rows(res);
    taos_free_result(res);
    return affectedRows;
H
hzcheng 已提交
853
  }
854
  
855
  taos_free_result(res);
856 857
  return 0;
}
H
hzcheng 已提交
858

859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875
static void getResult(TAOS_RES *res, char* resultFileName) {  
  TAOS_ROW    row = NULL;
  int         num_rows = 0;
  int         num_fields = taos_field_count(res);
  TAOS_FIELD *fields     = taos_fetch_fields(res);

  FILE *fp = NULL;
  if (resultFileName[0] != 0) {
    fp = fopen(resultFileName, "at");
    if (fp == NULL) {
      fprintf(stderr, "failed to open result file: %s, result will not save to file\n", resultFileName);
    }
  }
  
  char* databuf = (char*) calloc(1, 100*1024*1024);
  if (databuf == NULL) {
    fprintf(stderr, "failed to malloc, warning: save result to file slowly!\n");
876 877
    if (fp)
        fclose(fp);
878 879
    return ;
  }
H
hzcheng 已提交
880

881 882
  int   totalLen = 0;
  char  temp[16000];
H
hzcheng 已提交
883

884 885 886 887 888 889
  // fetch the records row by row
  while ((row = taos_fetch_row(res))) {
    if (totalLen >= 100*1024*1024 - 32000) {
      if (fp) fprintf(fp, "%s", databuf);
      totalLen = 0;
      memset(databuf, 0, 100*1024*1024);
H
hzcheng 已提交
890
    }
891 892 893 894 895 896
    num_rows++;
    int len = taos_print_row(temp, row, fields, num_fields);
    len += sprintf(temp + len, "\n");
    //printf("query result:%s\n", temp);
    memcpy(databuf + totalLen, temp, len);
    totalLen += len;
H
hzcheng 已提交
897 898
  }

899 900 901 902
  if (fp) fprintf(fp, "%s", databuf);
  tmfclose(fp);
  free(databuf);
}
H
hzcheng 已提交
903

904 905 906 907 908 909
static void selectAndGetResult(TAOS *taos, char *command, char* resultFileName) {
  TAOS_RES *res = taos_query(taos, command);
  if (res == NULL || taos_errno(res) != 0) {
    printf("failed to sql:%s, reason:%s\n", command, taos_errstr(res));
    taos_free_result(res);
    return;
H
hzcheng 已提交
910
  }
H
Hui Li 已提交
911
  
912 913 914
  getResult(res, resultFileName);
  taos_free_result(res);
}
H
hzcheng 已提交
915

916 917 918 919 920
double getCurrentTime() {
  struct timeval tv;
  if (gettimeofday(&tv, NULL) != 0) {
    perror("Failed to get current time in ms");
    return 0.0;
H
hzcheng 已提交
921
  }
S
Shuduo Sang 已提交
922

923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992
  return tv.tv_sec + tv.tv_usec / 1E6;
}

static int32_t rand_bool(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randint[cursor] % 2;
}

static int32_t rand_tinyint(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randint[cursor] % 128;
}

static int32_t rand_smallint(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randint[cursor] % 32767;
}

static int32_t rand_int(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randint[cursor];
}

static int64_t rand_bigint(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randbigint[cursor];
  
}

static float rand_float(){
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randfloat[cursor];    
}

static const char charset[] = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890";
void rand_string(char *str, int size) {
  str[0] = 0;
  if (size > 0) {
    //--size;
    int n;
    for (n = 0; n < size; n++) {
      int key = rand_tinyint() % (int)(sizeof(charset) - 1);
      str[n] = charset[key];
    }
    str[n] = 0;
  }
}

static double rand_double() {
  static int cursor;
  cursor++;
  cursor = cursor % MAX_PREPARED_RAND;
  return randdouble[cursor];

}

static void init_rand_data() {
  for (int i = 0; i < MAX_PREPARED_RAND; i++){
993 994 995 996
    randint[i] = (int)(taosRandom() % 65535);
    randbigint[i] = (int64_t)(taosRandom() % 2147483648);
    randfloat[i] = (float)(taosRandom() / 1000.0);
    randdouble[i] = (double)(taosRandom() / 1000000.0);
997 998 999
  }
}

1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019
#define SHOW_PARSE_RESULT_START()   \
    do { if (g_args.metaFile)  \
        printf("\033[1m\033[40;32m================ %s parse result START ================\033[0m\n", \
                g_args.metaFile); } while(0)

#define SHOW_PARSE_RESULT_END() \
    do { if (g_args.metaFile)   \
        printf("\033[1m\033[40;32m================ %s parse result END================\033[0m\n", \
                g_args.metaFile); } while(0)

#define SHOW_PARSE_RESULT_START_TO_FILE(fp)   \
    do { if (g_args.metaFile)  \
        fprintf(fp, "\033[1m\033[40;32m================ %s parse result START ================\033[0m\n", \
                g_args.metaFile); } while(0)

#define SHOW_PARSE_RESULT_END_TO_FILE(fp) \
    do { if (g_args.metaFile)   \
        fprintf(fp, "\033[1m\033[40;32m================ %s parse result END================\033[0m\n", \
                g_args.metaFile); } while(0)

1020
static int printfInsertMeta() {
1021 1022
    SHOW_PARSE_RESULT_START();

1023 1024 1025 1026 1027 1028
  printf("host:                       \033[33m%s:%u\033[0m\n", g_Dbs.host, g_Dbs.port);
  printf("user:                       \033[33m%s\033[0m\n", g_Dbs.user);
  printf("password:                   \033[33m%s\033[0m\n", g_Dbs.password);
  printf("resultFile:                 \033[33m%s\033[0m\n", g_Dbs.resultFile);
  printf("thread num of insert data:  \033[33m%d\033[0m\n", g_Dbs.threadCount);
  printf("thread num of create table: \033[33m%d\033[0m\n", g_Dbs.threadCountByCreateTbl);
1029 1030
  printf("insert interval:            \033[33m%d\033[0m\n", g_args.insert_interval);
  printf("number of records per req:  \033[33m%d\033[0m\n", g_args.num_of_RPR);
1031
  printf("max sql length:             \033[33m%d\033[0m\n", g_args.max_sql_len);
1032 1033 1034 1035

  printf("database count:             \033[33m%d\033[0m\n", g_Dbs.dbCount);
  for (int i = 0; i < g_Dbs.dbCount; i++) {
    printf("database[\033[33m%d\033[0m]:\n", i);
1036
    printf("  database[%d] name:      \033[33m%s\033[0m\n", i, g_Dbs.db[i].dbName);
1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079
    if (0 == g_Dbs.db[i].drop) {
      printf("  drop:                  \033[33mno\033[0m\n");     
    }else {
      printf("  drop:                  \033[33myes\033[0m\n"); 
    }

    if (g_Dbs.db[i].dbCfg.blocks > 0) {
      printf("  blocks:                \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.blocks);
    }
    if (g_Dbs.db[i].dbCfg.cache > 0) {
      printf("  cache:                 \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.cache);
    }
    if (g_Dbs.db[i].dbCfg.days > 0) {
      printf("  days:                  \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.days);
    }
    if (g_Dbs.db[i].dbCfg.keep > 0) {
      printf("  keep:                  \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.keep);
    }
    if (g_Dbs.db[i].dbCfg.replica > 0) {
      printf("  replica:               \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.replica);
    }
    if (g_Dbs.db[i].dbCfg.update > 0) {
      printf("  update:                \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.update);
    }
    if (g_Dbs.db[i].dbCfg.minRows > 0) {
      printf("  minRows:               \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.minRows);
    }
    if (g_Dbs.db[i].dbCfg.maxRows > 0) {
      printf("  maxRows:               \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.maxRows);
    }
    if (g_Dbs.db[i].dbCfg.comp > 0) {
      printf("  comp:                  \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.comp);
    }
    if (g_Dbs.db[i].dbCfg.walLevel > 0) {
      printf("  walLevel:              \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.walLevel);
    }
    if (g_Dbs.db[i].dbCfg.fsync > 0) {
      printf("  fsync:                 \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.fsync);
    }
    if (g_Dbs.db[i].dbCfg.quorum > 0) {
      printf("  quorum:                \033[33m%d\033[0m\n", g_Dbs.db[i].dbCfg.quorum);
    }
    if (g_Dbs.db[i].dbCfg.precision[0] != 0) {
1080 1081
      if ((0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ms", 2))
              || (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "us", 2))) {
1082 1083
        printf("  precision:             \033[33m%s\033[0m\n", g_Dbs.db[i].dbCfg.precision);
      } else {
1084 1085 1086
        printf("\033[1m\033[40;31m  precision error:       %s\033[0m\n",
                g_Dbs.db[i].dbCfg.precision);
        return -1;
1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110
      }
    }

    printf("  super table count:     \033[33m%d\033[0m\n", g_Dbs.db[i].superTblCount);
    for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) {
      printf("  super table[\033[33m%d\033[0m]:\n", j);
    
      printf("      stbName:           \033[33m%s\033[0m\n",  g_Dbs.db[i].superTbls[j].sTblName);   

      if (PRE_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable) {
        printf("      autoCreateTable:   \033[33m%s\033[0m\n",  "no");
      } else if (AUTO_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable) {
        printf("      autoCreateTable:   \033[33m%s\033[0m\n",  "yes");
      } else {
        printf("      autoCreateTable:   \033[33m%s\033[0m\n",  "error");
      }
      
      if (TBL_NO_EXISTS == g_Dbs.db[i].superTbls[j].childTblExists) {
        printf("      childTblExists:    \033[33m%s\033[0m\n",  "no");
      } else if (TBL_ALREADY_EXISTS == g_Dbs.db[i].superTbls[j].childTblExists) {
        printf("      childTblExists:    \033[33m%s\033[0m\n",  "yes");
      } else {
        printf("      childTblExists:    \033[33m%s\033[0m\n",  "error");
      }
1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122

      printf("      childTblCount:     \033[33m%d\033[0m\n",  g_Dbs.db[i].superTbls[j].childTblCount);
      printf("      childTblPrefix:    \033[33m%s\033[0m\n",  g_Dbs.db[i].superTbls[j].childTblPrefix);
      printf("      dataSource:        \033[33m%s\033[0m\n",  g_Dbs.db[i].superTbls[j].dataSource);
      printf("      insertMode:        \033[33m%s\033[0m\n",  g_Dbs.db[i].superTbls[j].insertMode);
      if (g_Dbs.db[i].superTbls[j].childTblLimit > 0) {
        printf("      childTblLimit:     \033[33m%d\033[0m\n",  g_Dbs.db[i].superTbls[j].childTblLimit);
      }
      if (g_Dbs.db[i].superTbls[j].childTblOffset > 0) {
        printf("      childTblOffset:    \033[33m%d\033[0m\n",  g_Dbs.db[i].superTbls[j].childTblOffset);
      }
      printf("      insertRows:        \033[33m%"PRId64"\033[0m\n", g_Dbs.db[i].superTbls[j].insertRows);
1123 1124

      if (0 == g_Dbs.db[i].superTbls[j].multiThreadWriteOneTbl) {
1125
        printf("      multiThreadWriteOneTbl:  \033[33mno\033[0m\n");
1126
      }else {
1127
        printf("      multiThreadWriteOneTbl:  \033[33myes\033[0m\n");
1128
      }
1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150
      printf("      numberOfTblInOneSql:     \033[33m%d\033[0m\n",
              g_Dbs.db[i].superTbls[j].numberOfTblInOneSql);
      printf("      rowsPerTbl:        \033[33m%d\033[0m\n",
              g_Dbs.db[i].superTbls[j].rowsPerTbl);
      printf("      disorderRange:     \033[33m%d\033[0m\n",
              g_Dbs.db[i].superTbls[j].disorderRange);
      printf("      disorderRatio:     \033[33m%d\033[0m\n",
              g_Dbs.db[i].superTbls[j].disorderRatio);
      printf("      maxSqlLen:         \033[33m%d\033[0m\n",
              g_Dbs.db[i].superTbls[j].maxSqlLen);
      printf("      timeStampStep:     \033[33m%d\033[0m\n",
              g_Dbs.db[i].superTbls[j].timeStampStep);
      printf("      startTimestamp:    \033[33m%s\033[0m\n",
              g_Dbs.db[i].superTbls[j].startTimestamp);
      printf("      sampleFormat:      \033[33m%s\033[0m\n",
              g_Dbs.db[i].superTbls[j].sampleFormat);
      printf("      sampleFile:        \033[33m%s\033[0m\n",
              g_Dbs.db[i].superTbls[j].sampleFile);
      printf("      tagsFile:          \033[33m%s\033[0m\n",
              g_Dbs.db[i].superTbls[j].tagsFile);
      printf("      columnCount:       \033[33m%d\033[0m\n        ",
              g_Dbs.db[i].superTbls[j].columnCount);
1151 1152
      for (int k = 0; k < g_Dbs.db[i].superTbls[j].columnCount; k++) {
        //printf("dataType:%s, dataLen:%d\t", g_Dbs.db[i].superTbls[j].columns[k].dataType, g_Dbs.db[i].superTbls[j].columns[k].dataLen);
1153 1154 1155 1156
        if ((0 == strncasecmp(g_Dbs.db[i].superTbls[j].columns[k].dataType,
                       "binary", 6))
                || (0 == strncasecmp(g_Dbs.db[i].superTbls[j].columns[k].dataType,
                       "nchar", 5))) {
1157
          printf("column[\033[33m%d\033[0m]:\033[33m%s(%d)\033[0m ", k,
1158 1159
                  g_Dbs.db[i].superTbls[j].columns[k].dataType,
                  g_Dbs.db[i].superTbls[j].columns[k].dataLen);
1160
        } else {
1161 1162
          printf("column[%d]:\033[33m%s\033[0m ", k,
                  g_Dbs.db[i].superTbls[j].columns[k].dataType);
1163 1164 1165
        }
      }
      printf("\n");
1166 1167 1168

      printf("      tagCount:            \033[33m%d\033[0m\n        ",
              g_Dbs.db[i].superTbls[j].tagCount);
1169 1170
      for (int k = 0; k < g_Dbs.db[i].superTbls[j].tagCount; k++) {
        //printf("dataType:%s, dataLen:%d\t", g_Dbs.db[i].superTbls[j].tags[k].dataType, g_Dbs.db[i].superTbls[j].tags[k].dataLen);
1171 1172 1173
        if ((0 == strncasecmp(g_Dbs.db[i].superTbls[j].tags[k].dataType, "binary", 6))
                || (0 == strncasecmp(g_Dbs.db[i].superTbls[j].tags[k].dataType, "nchar", 5))) {
          printf("tag[%d]:\033[33m%s(%d)\033[0m ", k,
1174 1175
                  g_Dbs.db[i].superTbls[j].tags[k].dataType,
                  g_Dbs.db[i].superTbls[j].tags[k].dataLen);
1176
        } else {
1177 1178
          printf("tag[%d]:\033[33m%s\033[0m ", k,
                  g_Dbs.db[i].superTbls[j].tags[k].dataType);
1179
        }
1180 1181 1182 1183 1184
      }
      printf("\n");
    }
    printf("\n");
  }
1185 1186

  SHOW_PARSE_RESULT_END();
1187 1188

  return 0;
1189 1190 1191
}

static void printfInsertMetaToFile(FILE* fp) {
1192 1193
    SHOW_PARSE_RESULT_START_TO_FILE(fp);

1194 1195 1196 1197 1198 1199 1200 1201 1202 1203
  fprintf(fp, "host:                       %s:%u\n", g_Dbs.host, g_Dbs.port);
  fprintf(fp, "user:                       %s\n", g_Dbs.user);
  fprintf(fp, "password:                   %s\n", g_Dbs.password);
  fprintf(fp, "resultFile:                 %s\n", g_Dbs.resultFile);
  fprintf(fp, "thread num of insert data:  %d\n", g_Dbs.threadCount);
  fprintf(fp, "thread num of create table: %d\n", g_Dbs.threadCountByCreateTbl);

  fprintf(fp, "database count:          %d\n", g_Dbs.dbCount);
  for (int i = 0; i < g_Dbs.dbCount; i++) {
    fprintf(fp, "database[%d]:\n", i);
1204
    fprintf(fp, "  database[%d] name:       %s\n", i, g_Dbs.db[i].dbName);
1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247
    if (0 == g_Dbs.db[i].drop) {
      fprintf(fp, "  drop:                  no\n");     
    }else {
      fprintf(fp, "  drop:                  yes\n"); 
    }

    if (g_Dbs.db[i].dbCfg.blocks > 0) {
      fprintf(fp, "  blocks:                %d\n", g_Dbs.db[i].dbCfg.blocks);
    }
    if (g_Dbs.db[i].dbCfg.cache > 0) {
      fprintf(fp, "  cache:                 %d\n", g_Dbs.db[i].dbCfg.cache);
    }
    if (g_Dbs.db[i].dbCfg.days > 0) {
      fprintf(fp, "  days:                  %d\n", g_Dbs.db[i].dbCfg.days);
    }
    if (g_Dbs.db[i].dbCfg.keep > 0) {
      fprintf(fp, "  keep:                  %d\n", g_Dbs.db[i].dbCfg.keep);
    }
    if (g_Dbs.db[i].dbCfg.replica > 0) {
      fprintf(fp, "  replica:               %d\n", g_Dbs.db[i].dbCfg.replica);
    }
    if (g_Dbs.db[i].dbCfg.update > 0) {
      fprintf(fp, "  update:                %d\n", g_Dbs.db[i].dbCfg.update);
    }
    if (g_Dbs.db[i].dbCfg.minRows > 0) {
      fprintf(fp, "  minRows:               %d\n", g_Dbs.db[i].dbCfg.minRows);
    }
    if (g_Dbs.db[i].dbCfg.maxRows > 0) {
      fprintf(fp, "  maxRows:               %d\n", g_Dbs.db[i].dbCfg.maxRows);
    }
    if (g_Dbs.db[i].dbCfg.comp > 0) {
      fprintf(fp, "  comp:                  %d\n", g_Dbs.db[i].dbCfg.comp);
    }
    if (g_Dbs.db[i].dbCfg.walLevel > 0) {
      fprintf(fp, "  walLevel:              %d\n", g_Dbs.db[i].dbCfg.walLevel);
    }
    if (g_Dbs.db[i].dbCfg.fsync > 0) {
      fprintf(fp, "  fsync:                 %d\n", g_Dbs.db[i].dbCfg.fsync);
    }
    if (g_Dbs.db[i].dbCfg.quorum > 0) {
      fprintf(fp, "  quorum:                %d\n", g_Dbs.db[i].dbCfg.quorum);
    }
    if (g_Dbs.db[i].dbCfg.precision[0] != 0) {
1248 1249
      if ((0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ms", 2))
              || (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "us", 2))) {
1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281 1282
        fprintf(fp, "  precision:             %s\n", g_Dbs.db[i].dbCfg.precision);
      } else {
        fprintf(fp, "  precision error:       %s\n", g_Dbs.db[i].dbCfg.precision);
      }
    }

    fprintf(fp, "  super table count:     %d\n", g_Dbs.db[i].superTblCount);
    for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) {
      fprintf(fp, "  super table[%d]:\n", j);
    
      fprintf(fp, "      stbName:           %s\n",  g_Dbs.db[i].superTbls[j].sTblName);   

      if (PRE_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable) {
        fprintf(fp, "      autoCreateTable:   %s\n",  "no");
      } else if (AUTO_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable) {
        fprintf(fp, "      autoCreateTable:   %s\n",  "yes");
      } else {
        fprintf(fp, "      autoCreateTable:   %s\n",  "error");
      }
      
      if (TBL_NO_EXISTS == g_Dbs.db[i].superTbls[j].childTblExists) {
        fprintf(fp, "      childTblExists:    %s\n",  "no");
      } else if (TBL_ALREADY_EXISTS == g_Dbs.db[i].superTbls[j].childTblExists) {
        fprintf(fp, "      childTblExists:    %s\n",  "yes");
      } else {
        fprintf(fp, "      childTblExists:    %s\n",  "error");
      }
      
      fprintf(fp, "      childTblCount:     %d\n",  g_Dbs.db[i].superTbls[j].childTblCount);      
      fprintf(fp, "      childTblPrefix:    %s\n",  g_Dbs.db[i].superTbls[j].childTblPrefix);      
      fprintf(fp, "      dataSource:        %s\n",  g_Dbs.db[i].superTbls[j].dataSource);      
      fprintf(fp, "      insertMode:        %s\n",  g_Dbs.db[i].superTbls[j].insertMode);      
      fprintf(fp, "      insertRows:        %"PRId64"\n", g_Dbs.db[i].superTbls[j].insertRows); 
1283
      fprintf(fp, "      insert interval:   %d\n", g_Dbs.db[i].superTbls[j].insertInterval);
1284 1285 1286 1287 1288 1289 1290 1291 1292 1293 1294 1295 1296 1297 1298 1299 1300 1301 1302 1303 1304 1305 1306 1307 1308 1309 1310 1311 1312 1313 1314 1315 1316 1317 1318 1319 1320 1321 1322 1323 1324 1325

      if (0 == g_Dbs.db[i].superTbls[j].multiThreadWriteOneTbl) {
        fprintf(fp, "      multiThreadWriteOneTbl:  no\n");     
      }else {
        fprintf(fp, "      multiThreadWriteOneTbl:  yes\n"); 
      }
      fprintf(fp, "      numberOfTblInOneSql:     %d\n",  g_Dbs.db[i].superTbls[j].numberOfTblInOneSql);     
      fprintf(fp, "      rowsPerTbl:        %d\n",  g_Dbs.db[i].superTbls[j].rowsPerTbl);     
      fprintf(fp, "      disorderRange:     %d\n",  g_Dbs.db[i].superTbls[j].disorderRange);     
      fprintf(fp, "      disorderRatio:     %d\n",  g_Dbs.db[i].superTbls[j].disorderRatio);
      fprintf(fp, "      maxSqlLen:         %d\n",  g_Dbs.db[i].superTbls[j].maxSqlLen);     
      
      fprintf(fp, "      timeStampStep:     %d\n",  g_Dbs.db[i].superTbls[j].timeStampStep);      
      fprintf(fp, "      startTimestamp:    %s\n",  g_Dbs.db[i].superTbls[j].startTimestamp);             
      fprintf(fp, "      sampleFormat:      %s\n",  g_Dbs.db[i].superTbls[j].sampleFormat);
      fprintf(fp, "      sampleFile:        %s\n",  g_Dbs.db[i].superTbls[j].sampleFile); 
      fprintf(fp, "      tagsFile:          %s\n",  g_Dbs.db[i].superTbls[j].tagsFile);   
    
      fprintf(fp, "      columnCount:       %d\n        ",  g_Dbs.db[i].superTbls[j].columnCount);
      for (int k = 0; k < g_Dbs.db[i].superTbls[j].columnCount; k++) {
        //printf("dataType:%s, dataLen:%d\t", g_Dbs.db[i].superTbls[j].columns[k].dataType, g_Dbs.db[i].superTbls[j].columns[k].dataLen);
        if ((0 == strncasecmp(g_Dbs.db[i].superTbls[j].columns[k].dataType, "binary", 6)) || (0 == strncasecmp(g_Dbs.db[i].superTbls[j].columns[k].dataType, "nchar", 5))) {
          fprintf(fp, "column[%d]:%s(%d) ", k, g_Dbs.db[i].superTbls[j].columns[k].dataType, g_Dbs.db[i].superTbls[j].columns[k].dataLen);
        } else {
          fprintf(fp, "column[%d]:%s ", k, g_Dbs.db[i].superTbls[j].columns[k].dataType);
        }
      }
      fprintf(fp, "\n");
      
      fprintf(fp, "      tagCount:            %d\n        ",  g_Dbs.db[i].superTbls[j].tagCount);
      for (int k = 0; k < g_Dbs.db[i].superTbls[j].tagCount; k++) {
        //printf("dataType:%s, dataLen:%d\t", g_Dbs.db[i].superTbls[j].tags[k].dataType, g_Dbs.db[i].superTbls[j].tags[k].dataLen);
        if ((0 == strncasecmp(g_Dbs.db[i].superTbls[j].tags[k].dataType, "binary", 6)) || (0 == strncasecmp(g_Dbs.db[i].superTbls[j].tags[k].dataType, "nchar", 5))) {
          fprintf(fp, "tag[%d]:%s(%d) ", k, g_Dbs.db[i].superTbls[j].tags[k].dataType, g_Dbs.db[i].superTbls[j].tags[k].dataLen);
        } else {
          fprintf(fp, "tag[%d]:%s ", k, g_Dbs.db[i].superTbls[j].tags[k].dataType);
        }     
      }
      fprintf(fp, "\n");
    }
    fprintf(fp, "\n");
  }
1326
    SHOW_PARSE_RESULT_END_TO_FILE(fp);
1327 1328 1329
}

static void printfQueryMeta() {
1330
    SHOW_PARSE_RESULT_START();
1331 1332 1333 1334 1335 1336 1337 1338 1339 1340 1341
  printf("host:                    \033[33m%s:%u\033[0m\n", g_queryInfo.host, g_queryInfo.port);
  printf("user:                    \033[33m%s\033[0m\n", g_queryInfo.user);
  printf("password:                \033[33m%s\033[0m\n", g_queryInfo.password);
  printf("database name:           \033[33m%s\033[0m\n", g_queryInfo.dbName);

  printf("\n");
  printf("specified table query info:                   \n");  
  printf("query interval: \033[33m%d\033[0m\n", g_queryInfo.superQueryInfo.rate);
  printf("concurrent:     \033[33m%d\033[0m\n", g_queryInfo.superQueryInfo.concurrent);
  printf("sqlCount:       \033[33m%d\033[0m\n", g_queryInfo.superQueryInfo.sqlCount); 

1342
  if (SUBSCRIBE_MODE == g_args.test_mode) {
1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358
    printf("mod:            \033[33m%d\033[0m\n", g_queryInfo.superQueryInfo.subscribeMode);
    printf("interval:       \033[33m%d\033[0m\n", g_queryInfo.superQueryInfo.subscribeInterval);
    printf("restart:        \033[33m%d\033[0m\n", g_queryInfo.superQueryInfo.subscribeRestart);
    printf("keepProgress:   \033[33m%d\033[0m\n", g_queryInfo.superQueryInfo.subscribeKeepProgress);
  }

  for (int i = 0; i < g_queryInfo.superQueryInfo.sqlCount; i++) {
    printf("  sql[%d]: \033[33m%s\033[0m\n", i, g_queryInfo.superQueryInfo.sql[i]);
  }
  printf("\n");
  printf("super table query info:                   \n");  
  printf("query interval: \033[33m%d\033[0m\n", g_queryInfo.subQueryInfo.rate);
  printf("threadCnt:      \033[33m%d\033[0m\n", g_queryInfo.subQueryInfo.threadCnt);
  printf("childTblCount:  \033[33m%d\033[0m\n", g_queryInfo.subQueryInfo.childTblCount);
  printf("stable name:    \033[33m%s\033[0m\n", g_queryInfo.subQueryInfo.sTblName);

1359
  if (SUBSCRIBE_MODE == g_args.test_mode) {
1360 1361 1362 1363 1364 1365 1366 1367 1368 1369 1370
    printf("mod:            \033[33m%d\033[0m\n", g_queryInfo.subQueryInfo.subscribeMode);
    printf("interval:       \033[33m%d\033[0m\n", g_queryInfo.subQueryInfo.subscribeInterval);
    printf("restart:        \033[33m%d\033[0m\n", g_queryInfo.subQueryInfo.subscribeRestart);
    printf("keepProgress:   \033[33m%d\033[0m\n", g_queryInfo.subQueryInfo.subscribeKeepProgress);
  }
  
  printf("sqlCount:       \033[33m%d\033[0m\n", g_queryInfo.subQueryInfo.sqlCount);  
  for (int i = 0; i < g_queryInfo.subQueryInfo.sqlCount; i++) {
    printf("  sql[%d]: \033[33m%s\033[0m\n", i, g_queryInfo.subQueryInfo.sql[i]);
  }  
  printf("\n");
1371 1372

    SHOW_PARSE_RESULT_END();
1373 1374 1375 1376 1377 1378 1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413 1414 1415 1416 1417 1418 1419 1420 1421 1422 1423 1424 1425 1426 1427 1428 1429 1430 1431 1432 1433 1434 1435 1436 1437 1438 1439 1440 1441 1442 1443 1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456 1457 1458 1459 1460 1461 1462 1463 1464 1465 1466 1467 1468 1469 1470 1471 1472 1473 1474 1475 1476 1477 1478 1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490 1491 1492 1493 1494 1495 1496 1497 1498 1499 1500 1501 1502 1503 1504 1505 1506 1507 1508 1509 1510 1511 1512 1513 1514 1515 1516 1517 1518 1519 1520
}


static char* xFormatTimestamp(char* buf, int64_t val, int precision) {
  time_t tt;
  if (precision == TSDB_TIME_PRECISION_MICRO) {
    tt = (time_t)(val / 1000000);
  } else {
    tt = (time_t)(val / 1000);
  }

/* comment out as it make testcases like select_with_tags.sim fail.
  but in windows, this may cause the call to localtime crash if tt < 0,
  need to find a better solution.
  if (tt < 0) {
    tt = 0;
  }
  */

#ifdef WINDOWS
  if (tt < 0) tt = 0;
#endif

  struct tm* ptm = localtime(&tt);
  size_t pos = strftime(buf, 32, "%Y-%m-%d %H:%M:%S", ptm);

  if (precision == TSDB_TIME_PRECISION_MICRO) {
    sprintf(buf + pos, ".%06d", (int)(val % 1000000));
  } else {
    sprintf(buf + pos, ".%03d", (int)(val % 1000));
  }

  return buf;
}

static void xDumpFieldToFile(FILE* fp, const char* val, TAOS_FIELD* field, int32_t length, int precision) {
  if (val == NULL) {
    fprintf(fp, "%s", TSDB_DATA_NULL_STR);
    return;
  }

  char buf[TSDB_MAX_BYTES_PER_ROW];
  switch (field->type) {
    case TSDB_DATA_TYPE_BOOL:
      fprintf(fp, "%d", ((((int32_t)(*((char *)val))) == 1) ? 1 : 0));
      break;
    case TSDB_DATA_TYPE_TINYINT:
      fprintf(fp, "%d", *((int8_t *)val));
      break;
    case TSDB_DATA_TYPE_SMALLINT:
      fprintf(fp, "%d", *((int16_t *)val));
      break;
    case TSDB_DATA_TYPE_INT:
      fprintf(fp, "%d", *((int32_t *)val));
      break;
    case TSDB_DATA_TYPE_BIGINT:
      fprintf(fp, "%" PRId64, *((int64_t *)val));
      break;
    case TSDB_DATA_TYPE_FLOAT:
      fprintf(fp, "%.5f", GET_FLOAT_VAL(val));
      break;
    case TSDB_DATA_TYPE_DOUBLE:
      fprintf(fp, "%.9f", GET_DOUBLE_VAL(val));
      break;
    case TSDB_DATA_TYPE_BINARY:
    case TSDB_DATA_TYPE_NCHAR:
      memcpy(buf, val, length);
      buf[length] = 0;
      fprintf(fp, "\'%s\'", buf);
      break;
    case TSDB_DATA_TYPE_TIMESTAMP:
      xFormatTimestamp(buf, *(int64_t*)val, precision);
      fprintf(fp, "'%s'", buf);
      break;
    default:
      break;
  }
}

static int xDumpResultToFile(const char* fname, TAOS_RES* tres) {
  TAOS_ROW row = taos_fetch_row(tres);
  if (row == NULL) {
    return 0;
  }

  FILE* fp = fopen(fname, "at");
  if (fp == NULL) {
    fprintf(stderr, "ERROR: failed to open file: %s\n", fname);
    return -1;
  }

  int num_fields = taos_num_fields(tres);
  TAOS_FIELD *fields = taos_fetch_fields(tres);
  int precision = taos_result_precision(tres);

  for (int col = 0; col < num_fields; col++) {
    if (col > 0) {
      fprintf(fp, ",");
    }
    fprintf(fp, "%s", fields[col].name);
  }
  fputc('\n', fp);
  
  int numOfRows = 0;
  do {
    int32_t* length = taos_fetch_lengths(tres);
    for (int i = 0; i < num_fields; i++) {
      if (i > 0) {
        fputc(',', fp);
      }
      xDumpFieldToFile(fp, (const char*)row[i], fields +i, length[i], precision);
    }
    fputc('\n', fp);

    numOfRows++;
    row = taos_fetch_row(tres);
  } while( row != NULL);

  fclose(fp);

  return numOfRows;
}

static int getDbFromServer(TAOS * taos, SDbInfo** dbInfos) {  
  TAOS_RES * res;  
  TAOS_ROW row = NULL;
  int count = 0;
  
  res = taos_query(taos, "show databases;");  
  int32_t code = taos_errno(res);
  
  if (code != 0) {
    fprintf(stderr, "failed to run <show databases>, reason: %s\n", taos_errstr(res));
    return -1;
  }

  TAOS_FIELD *fields = taos_fetch_fields(res);

  while ((row = taos_fetch_row(res)) != NULL) {
    // sys database name : 'log'
    if (strncasecmp(row[TSDB_SHOW_DB_NAME_INDEX], "log", fields[TSDB_SHOW_DB_NAME_INDEX].bytes) == 0) continue;

    dbInfos[count] = (SDbInfo *)calloc(1, sizeof(SDbInfo));
    if (dbInfos[count] == NULL) {
      fprintf(stderr, "failed to allocate memory for some dbInfo[%d]\n", count);
      return -1;
    }

1521
    tstrncpy(dbInfos[count]->name, (char *)row[TSDB_SHOW_DB_NAME_INDEX], fields[TSDB_SHOW_DB_NAME_INDEX].bytes);
1522 1523 1524 1525 1526 1527 1528
    xFormatTimestamp(dbInfos[count]->create_time, *(int64_t*)row[TSDB_SHOW_DB_CREATED_TIME_INDEX], TSDB_TIME_PRECISION_MILLI);
    dbInfos[count]->ntables = *((int32_t *)row[TSDB_SHOW_DB_NTABLES_INDEX]);
    dbInfos[count]->vgroups = *((int32_t *)row[TSDB_SHOW_DB_VGROUPS_INDEX]);  
    dbInfos[count]->replica = *((int16_t *)row[TSDB_SHOW_DB_REPLICA_INDEX]);
    dbInfos[count]->quorum = *((int16_t *)row[TSDB_SHOW_DB_QUORUM_INDEX]);
    dbInfos[count]->days = *((int16_t *)row[TSDB_SHOW_DB_DAYS_INDEX]);  

1529
    tstrncpy(dbInfos[count]->keeplist, (char *)row[TSDB_SHOW_DB_KEEP_INDEX], fields[TSDB_SHOW_DB_KEEP_INDEX].bytes);      
1530 1531 1532 1533 1534 1535 1536 1537 1538
    dbInfos[count]->cache = *((int32_t *)row[TSDB_SHOW_DB_CACHE_INDEX]);
    dbInfos[count]->blocks = *((int32_t *)row[TSDB_SHOW_DB_BLOCKS_INDEX]);
    dbInfos[count]->minrows = *((int32_t *)row[TSDB_SHOW_DB_MINROWS_INDEX]);
    dbInfos[count]->maxrows = *((int32_t *)row[TSDB_SHOW_DB_MAXROWS_INDEX]);
    dbInfos[count]->wallevel = *((int8_t *)row[TSDB_SHOW_DB_WALLEVEL_INDEX]);
    dbInfos[count]->fsync = *((int32_t *)row[TSDB_SHOW_DB_FSYNC_INDEX]);
    dbInfos[count]->comp = (int8_t)(*((int8_t *)row[TSDB_SHOW_DB_COMP_INDEX]));
    dbInfos[count]->cachelast = (int8_t)(*((int8_t *)row[TSDB_SHOW_DB_CACHELAST_INDEX]));

1539 1540 1541
    tstrncpy(dbInfos[count]->precision, 
            (char *)row[TSDB_SHOW_DB_PRECISION_INDEX],
            fields[TSDB_SHOW_DB_PRECISION_INDEX].bytes);  
1542
    dbInfos[count]->update = *((int8_t *)row[TSDB_SHOW_DB_UPDATE_INDEX]);
1543
    tstrncpy(dbInfos[count]->status, (char *)row[TSDB_SHOW_DB_STATUS_INDEX], fields[TSDB_SHOW_DB_STATUS_INDEX].bytes); 
1544 1545 1546 1547 1548 1549 1550 1551 1552 1553 1554 1555
    
    count++;
    if (count > MAX_DATABASE_COUNT) {
      fprintf(stderr, "The database count overflow than %d\n", MAX_DATABASE_COUNT);      
      break;
    }
  }

  return count;
}

static void printfDbInfoForQueryToFile(char* filename, SDbInfo* dbInfos, int index) {
1556
  if (filename[0] == 0)
1557
      return;
1558 1559 1560 1561 1562

  FILE *fp = fopen(filename, "at");
  if (fp == NULL) {
    fprintf(stderr, "failed to open file: %s\n", filename);
    return;
1563 1564 1565 1566 1567 1568 1569 1570 1571 1572 1573 1574 1575 1576 1577 1578 1579 1580 1581 1582 1583 1584 1585 1586 1587 1588 1589 1590 1591 1592 1593 1594 1595 1596 1597 1598 1599 1600 1601 1602 1603 1604 1605 1606 1607 1608 1609 1610 1611 1612 1613 1614 1615 1616 1617 1618
  }

  fprintf(fp, "================ database[%d] ================\n", index);
  fprintf(fp, "name: %s\n", dbInfos->name);
  fprintf(fp, "created_time: %s\n", dbInfos->create_time);
  fprintf(fp, "ntables: %d\n", dbInfos->ntables);
  fprintf(fp, "vgroups: %d\n", dbInfos->vgroups);  
  fprintf(fp, "replica: %d\n", dbInfos->replica);
  fprintf(fp, "quorum: %d\n", dbInfos->quorum);
  fprintf(fp, "days: %d\n", dbInfos->days);    
  fprintf(fp, "keep0,keep1,keep(D): %s\n", dbInfos->keeplist);      
  fprintf(fp, "cache(MB): %d\n", dbInfos->cache);
  fprintf(fp, "blocks: %d\n", dbInfos->blocks);
  fprintf(fp, "minrows: %d\n", dbInfos->minrows);
  fprintf(fp, "maxrows: %d\n", dbInfos->maxrows);
  fprintf(fp, "wallevel: %d\n", dbInfos->wallevel);
  fprintf(fp, "fsync: %d\n", dbInfos->fsync);
  fprintf(fp, "comp: %d\n", dbInfos->comp);
  fprintf(fp, "cachelast: %d\n", dbInfos->cachelast);  
  fprintf(fp, "precision: %s\n", dbInfos->precision);  
  fprintf(fp, "update: %d\n", dbInfos->update);
  fprintf(fp, "status: %s\n", dbInfos->status); 
  fprintf(fp, "\n");

  fclose(fp);
}

static void printfQuerySystemInfo(TAOS * taos) {
  char filename[MAX_QUERY_SQL_LENGTH+1] = {0};
  char buffer[MAX_QUERY_SQL_LENGTH+1] = {0};
  TAOS_RES* res;

  time_t t;
  struct tm* lt;
  time(&t);
  lt = localtime(&t);
  snprintf(filename, MAX_QUERY_SQL_LENGTH, "querySystemInfo-%d-%d-%d %d:%d:%d", lt->tm_year+1900, lt->tm_mon, lt->tm_mday, lt->tm_hour, lt->tm_min, lt->tm_sec);
  
  // show variables
  res = taos_query(taos, "show variables;");
  //getResult(res, filename);
  xDumpResultToFile(filename, res);

  // show dnodes
  res = taos_query(taos, "show dnodes;");
  xDumpResultToFile(filename, res);
  //getResult(res, filename);
  
  // show databases
  res = taos_query(taos, "show databases;");
  SDbInfo** dbInfos = (SDbInfo **)calloc(MAX_DATABASE_COUNT, sizeof(SDbInfo *));
  if (dbInfos == NULL) {
    fprintf(stderr, "failed to allocate memory\n");
    return;
  }
  int dbCount = getDbFromServer(taos, dbInfos);
1619 1620 1621 1622
  if (dbCount <= 0) {
      free(dbInfos);
      return;
  }
1623 1624 1625 1626 1627 1628 1629 1630 1631 1632 1633 1634 1635 1636 1637 1638 1639 1640 1641 1642 1643 1644

  for (int i = 0; i < dbCount; i++) {
    // printf database info 
    printfDbInfoForQueryToFile(filename, dbInfos[i], i);
    
    // show db.vgroups
    snprintf(buffer, MAX_QUERY_SQL_LENGTH, "show %s.vgroups;", dbInfos[i]->name);
    res = taos_query(taos, buffer);
    xDumpResultToFile(filename, res);
  
    // show db.stables
    snprintf(buffer, MAX_QUERY_SQL_LENGTH, "show %s.stables;", dbInfos[i]->name);
    res = taos_query(taos, buffer);
    xDumpResultToFile(filename, res);

    free(dbInfos[i]);
  }

  free(dbInfos);
  
}

1645
void ERROR_EXIT(const char *msg) { perror(msg); exit(-1); }
1646 1647 1648

int postProceSql(char* host, uint16_t port, char* sqlstr)
{
1649
    char *req_fmt = "POST %s HTTP/1.1\r\nHost: %s:%d\r\nAccept: */*\r\nAuthorization: Basic %s\r\nContent-Length: %d\r\nContent-Type: application/x-www-form-urlencoded\r\n\r\n%s";
1650 1651 1652 1653 1654

    char *url = "/rest/sql";

    struct hostent *server;
    struct sockaddr_in serv_addr;
1655
    int bytes, sent, received, req_str_len, resp_len;
1656 1657 1658
    char *request_buf;
    char response_buf[RESP_BUF_LEN];
    uint16_t rest_port = port + TSDB_PORT_HTTP;
1659

1660 1661 1662 1663
    int req_buf_len = strlen(sqlstr) + REQ_EXTRA_BUF_LEN;

    request_buf = malloc(req_buf_len);
    if (NULL == request_buf)
1664
        ERROR_EXIT("ERROR, cannot allocate memory.");
1665

1666 1667 1668 1669 1670 1671 1672 1673 1674 1675 1676
    char userpass_buf[INPUT_BUF_LEN];
    int mod_table[] = {0, 2, 1};

    static char base64[] = {'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H',
      'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P',
      'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X',
      'Y', 'Z', 'a', 'b', 'c', 'd', 'e', 'f',
      'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n',
      'o', 'p', 'q', 'r', 's', 't', 'u', 'v',
      'w', 'x', 'y', 'z', '0', '1', '2', '3',
      '4', '5', '6', '7', '8', '9', '+', '/'};
1677

1678 1679 1680 1681 1682 1683
    snprintf(userpass_buf, INPUT_BUF_LEN, "%s:%s",
        g_Dbs.user, g_Dbs.password);
    size_t userpass_buf_len = strlen(userpass_buf);
    size_t encoded_len = 4 * ((userpass_buf_len +2) / 3);

    char base64_buf[INPUT_BUF_LEN];
1684
#ifdef WINDOWS
1685
    WSADATA wsaData;
1686
    WSAStartup(MAKEWORD(2, 2), &wsaData);
1687 1688 1689 1690
    SOCKET sockfd;
#else
    int sockfd;
#endif
1691
    sockfd = socket(AF_INET, SOCK_STREAM, 0);
1692
    if (sockfd < 0) {
1693 1694 1695 1696
#ifdef WINDOWS
        fprintf(stderr, "Could not create socket : %d" , WSAGetLastError());
#endif
        debugPrint("%s() LN%d sockfd=%d\n", __func__, __LINE__, sockfd);
1697
        free(request_buf);
1698
        ERROR_EXIT("ERROR opening socket");
1699
    }
1700 1701

    server = gethostbyname(host);
1702 1703
    if (server == NULL) {
        free(request_buf);
1704
        ERROR_EXIT("ERROR, no such host");
1705
    }
1706

1707
    debugPrint("h_name: %s\nh_addretype: %s\nh_length: %d\n",
1708 1709 1710 1711
            server->h_name,
            (server->h_addrtype == AF_INET)?"ipv4":"ipv6",
            server->h_length);

1712
    memset(&serv_addr, 0, sizeof(serv_addr));
1713
    serv_addr.sin_family = AF_INET;
1714
    serv_addr.sin_port = htons(rest_port);
1715 1716 1717
#ifdef WINDOWS
    serv_addr.sin_addr.s_addr = inet_addr(host);
#else
1718
    memcpy(&serv_addr.sin_addr.s_addr,server->h_addr,server->h_length);
1719
#endif
1720

1721 1722 1723
    int retConn = connect(sockfd,(struct sockaddr *)&serv_addr,sizeof(serv_addr));
    debugPrint("%s() LN%d connect() return %d\n", __func__, __LINE__, retConn);
    if (retConn < 0) {
1724
        free(request_buf);
1725
        ERROR_EXIT("ERROR connecting");
1726
    }
1727

1728 1729 1730
    memset(base64_buf, 0, INPUT_BUF_LEN);

    for (int n = 0, m = 0; n < userpass_buf_len;) {
1731
      uint32_t oct_a = n < userpass_buf_len ?
1732
        (unsigned char) userpass_buf[n++]:0;
1733
      uint32_t oct_b = n < userpass_buf_len ?
1734
        (unsigned char) userpass_buf[n++]:0;
1735
      uint32_t oct_c = n < userpass_buf_len ?
1736 1737 1738 1739 1740 1741 1742 1743 1744 1745 1746 1747 1748 1749 1750
        (unsigned char) userpass_buf[n++]:0;
      uint32_t triple = (oct_a << 0x10) + (oct_b << 0x08) + oct_c;

      base64_buf[m++] = base64[(triple >> 3* 6) & 0x3f];
      base64_buf[m++] = base64[(triple >> 2* 6) & 0x3f];
      base64_buf[m++] = base64[(triple >> 1* 6) & 0x3f];
      base64_buf[m++] = base64[(triple >> 0* 6) & 0x3f];
    }

    for (int l = 0; l < mod_table[userpass_buf_len % 3]; l++)
      base64_buf[encoded_len - 1 - l] = '=';

    debugPrint("%s() LN%d: auth string base64 encoded: %s\n", __func__, __LINE__, base64_buf);
    char *auth = base64_buf;

1751 1752 1753
    int r = snprintf(request_buf,
            req_buf_len,
            req_fmt, url, host, rest_port,
1754 1755 1756 1757 1758
            auth, strlen(sqlstr), sqlstr);
    if (r >= req_buf_len) {
        free(request_buf);
        ERROR_EXIT("ERROR too long request");
    }
1759
    verbosePrint("%s() LN%d: Request:\n%s\n", __func__, __LINE__, request_buf);
1760

1761
    req_str_len = strlen(request_buf);
1762 1763
    sent = 0;
    do {
1764 1765 1766
#ifdef WINDOWS
        bytes = send(sockfd, request_buf + sent, req_str_len - sent, 0);
#else
1767
        bytes = write(sockfd, request_buf + sent, req_str_len - sent);
1768
#endif
1769
        if (bytes < 0)
1770
            ERROR_EXIT("ERROR writing message to socket");
1771 1772 1773
        if (bytes == 0)
            break;
        sent+=bytes;
1774
    } while (sent < req_str_len);
1775

1776 1777
    memset(response_buf, 0, RESP_BUF_LEN);
    resp_len = sizeof(response_buf) - 1;
1778 1779
    received = 0;
    do {
1780 1781 1782
#ifdef WINDOWS
        bytes = recv(sockfd, response_buf + received, resp_len - received, 0);
#else
1783
        bytes = read(sockfd, response_buf + received, resp_len - received);
1784
#endif
1785 1786
        if (bytes < 0) {
            free(request_buf);
1787
            ERROR_EXIT("ERROR reading response from socket");
1788
        }
1789 1790
        if (bytes == 0)
            break;
1791 1792
        received += bytes;
    } while (received < resp_len);
1793

1794 1795
    if (received == resp_len) {
        free(request_buf);
1796
        ERROR_EXIT("ERROR storing complete response from socket");
1797
    }
1798

1799
    response_buf[RESP_BUF_LEN - 1] = '\0';
1800 1801 1802
    printf("Response:\n%s\n", response_buf);

    free(request_buf);
1803 1804 1805 1806
#ifdef WINDOWS
    closesocket(sockfd);
    WSACleanup();
#else
1807
    close(sockfd);
1808
#endif
1809 1810 1811 1812

    return 0;
}

1813
static char* getTagValueFromTagSample(SSuperTable* stbInfo, int tagUsePos) {
1814 1815 1816 1817 1818
  char*  dataBuf = (char*)calloc(TSDB_MAX_SQL_LEN+1, 1);
  if (NULL == dataBuf) {
    printf("calloc failed! size:%d\n", TSDB_MAX_SQL_LEN+1);
    return NULL;
  }
1819

1820
  int    dataLen = 0;
1821 1822 1823
  dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
          "(%s)", stbInfo->tagDataBuf + stbInfo->lenOfTagOfOneRow * tagUsePos);

1824 1825 1826
  return dataBuf;
}

1827
static char* generateTagVaulesForStb(SSuperTable* stbInfo) {
1828 1829 1830 1831 1832 1833 1834 1835 1836
  char*  dataBuf = (char*)calloc(TSDB_MAX_SQL_LEN+1, 1);
  if (NULL == dataBuf) {
    printf("calloc failed! size:%d\n", TSDB_MAX_SQL_LEN+1);
    return NULL;
  }

  int    dataLen = 0;
  dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen, "(");
  for (int i = 0; i < stbInfo->tagCount; i++) {
1837 1838
    if ((0 == strncasecmp(stbInfo->tags[i].dataType, "binary", strlen("binary")))
            || (0 == strncasecmp(stbInfo->tags[i].dataType, "nchar", strlen("nchar")))) {
1839
      if (stbInfo->tags[i].dataLen > TSDB_MAX_BINARY_LEN) {
1840 1841
        printf("binary or nchar length overflow, max size:%u\n",
                (uint32_t)TSDB_MAX_BINARY_LEN);
1842 1843 1844
        tmfree(dataBuf);
        return NULL;
      }
1845

1846 1847 1848 1849 1850 1851 1852
      char* buf = (char*)calloc(stbInfo->tags[i].dataLen+1, 1);
      if (NULL == buf) {
        printf("calloc failed! size:%d\n", stbInfo->tags[i].dataLen);
        tmfree(dataBuf);
        return NULL;
      }
      rand_string(buf, stbInfo->tags[i].dataLen);
1853 1854
      dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
              "\'%s\', ", buf);
1855
      tmfree(buf);
1856 1857 1858 1859 1860 1861 1862 1863 1864 1865 1866 1867 1868 1869 1870 1871 1872 1873 1874 1875 1876 1877 1878 1879 1880 1881 1882 1883 1884 1885 1886 1887
    } else if (0 == strncasecmp(stbInfo->tags[i].dataType,
                "int", strlen("int"))) {
      dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
              "%d, ", rand_int());
    } else if (0 == strncasecmp(stbInfo->tags[i].dataType,
                "bigint", strlen("bigint"))) {
      dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
              "%"PRId64", ", rand_bigint());
    }  else if (0 == strncasecmp(stbInfo->tags[i].dataType,
                "float", strlen("float"))) {
      dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
              "%f, ", rand_float());
    }  else if (0 == strncasecmp(stbInfo->tags[i].dataType,
                "double", strlen("double"))) {
      dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
              "%f, ", rand_double());
    }  else if (0 == strncasecmp(stbInfo->tags[i].dataType,
                "smallint", strlen("smallint"))) {
      dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
              "%d, ", rand_smallint());
    }  else if (0 == strncasecmp(stbInfo->tags[i].dataType,
                "tinyint", strlen("tinyint"))) {
      dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
              "%d, ", rand_tinyint());
    }  else if (0 == strncasecmp(stbInfo->tags[i].dataType,
                "bool", strlen("bool"))) {
      dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
              "%d, ", rand_bool());
    }  else if (0 == strncasecmp(stbInfo->tags[i].dataType,
                "timestamp", strlen("timestamp"))) {
      dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen,
              "%"PRId64", ", rand_bigint());
1888 1889 1890 1891 1892 1893
    }  else {
      printf("No support data type: %s\n", stbInfo->tags[i].dataType);
      tmfree(dataBuf);
      return NULL;
    }
  }
1894

1895 1896 1897 1898 1899 1900 1901 1902 1903 1904 1905 1906 1907 1908 1909 1910 1911 1912 1913 1914 1915 1916 1917 1918 1919 1920 1921 1922 1923 1924 1925 1926 1927 1928 1929 1930 1931 1932 1933 1934 1935 1936 1937 1938
  dataLen -= 2;
  dataLen += snprintf(dataBuf + dataLen, TSDB_MAX_SQL_LEN - dataLen, ")");  
  return dataBuf;
}

static int calcRowLen(SSuperTable*  superTbls) {  
  int colIndex;
  int  lenOfOneRow = 0;
  
  for (colIndex = 0; colIndex < superTbls->columnCount; colIndex++) {
    char* dataType = superTbls->columns[colIndex].dataType;
    
    if (strcasecmp(dataType, "BINARY") == 0) {
      lenOfOneRow += superTbls->columns[colIndex].dataLen + 3;
    } else if (strcasecmp(dataType, "NCHAR") == 0) {
      lenOfOneRow += superTbls->columns[colIndex].dataLen + 3;
    } else if (strcasecmp(dataType, "INT") == 0)  {
      lenOfOneRow += 11;
    } else if (strcasecmp(dataType, "BIGINT") == 0)  {
      lenOfOneRow += 21;
    } else if (strcasecmp(dataType, "SMALLINT") == 0)  {
      lenOfOneRow += 6;
    } else if (strcasecmp(dataType, "TINYINT") == 0)  {
      lenOfOneRow += 4;
    } else if (strcasecmp(dataType, "BOOL") == 0)  {
      lenOfOneRow += 6;
    } else if (strcasecmp(dataType, "FLOAT") == 0) {
      lenOfOneRow += 22;
    } else if (strcasecmp(dataType, "DOUBLE") == 0) { 
      lenOfOneRow += 42;
    }  else if (strcasecmp(dataType, "TIMESTAMP") == 0) { 
      lenOfOneRow += 21;
    } else {
      printf("get error data type : %s\n", dataType);
      exit(-1);
    }
  }

  superTbls->lenOfOneRow = lenOfOneRow + 20; // timestamp

  int tagIndex;
  int lenOfTagOfOneRow = 0;
  for (tagIndex = 0; tagIndex < superTbls->tagCount; tagIndex++) {
    char* dataType = superTbls->tags[tagIndex].dataType;
1939

1940 1941 1942 1943 1944 1945 1946 1947 1948 1949 1950 1951 1952 1953 1954 1955 1956 1957 1958 1959 1960 1961 1962 1963 1964 1965 1966 1967 1968 1969
    if (strcasecmp(dataType, "BINARY") == 0) {
      lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 3;
    } else if (strcasecmp(dataType, "NCHAR") == 0) {
      lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 3;
    } else if (strcasecmp(dataType, "INT") == 0)  {
      lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 11;
    } else if (strcasecmp(dataType, "BIGINT") == 0)  {
      lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 21;
    } else if (strcasecmp(dataType, "SMALLINT") == 0)  {
      lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 6;
    } else if (strcasecmp(dataType, "TINYINT") == 0)  {
      lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 4;
    } else if (strcasecmp(dataType, "BOOL") == 0)  {
      lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 6;
    } else if (strcasecmp(dataType, "FLOAT") == 0) {
      lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 22;
    } else if (strcasecmp(dataType, "DOUBLE") == 0) { 
      lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 42;
    } else {
      printf("get error tag type : %s\n", dataType);
      exit(-1);
    }
  }

  superTbls->lenOfTagOfOneRow = lenOfTagOfOneRow;
    
  return 0;
}


1970 1971 1972 1973
static int getChildNameOfSuperTableWithLimitAndOffset(TAOS * taos,
        char* dbName, char* sTblName, char** childTblNameOfSuperTbl,
        int* childTblCountOfSuperTbl, int limit, int offset) {

1974
  char command[BUFFER_SIZE] = "\0";
1975 1976
  char limitBuf[100] = "\0";

1977 1978 1979 1980
  TAOS_RES * res;  
  TAOS_ROW row = NULL;

  char* childTblName = *childTblNameOfSuperTbl;
1981 1982 1983 1984 1985

  if (offset >= 0) {
    snprintf(limitBuf, 100, " limit %d offset %d", limit, offset);
  }

1986
  //get all child table name use cmd: select tbname from superTblName;  
1987 1988
  snprintf(command, BUFFER_SIZE, "select tbname from %s.%s %s", dbName, sTblName, limitBuf);

1989 1990 1991 1992 1993 1994 1995 1996 1997
  res = taos_query(taos, command);  
  int32_t code = taos_errno(res);
  if (code != 0) {
    printf("failed to run command %s\n", command);
    taos_free_result(res);
    taos_close(taos);
    exit(-1);
  }

1998
  int childTblCount = (limit < 0)?10000:limit;
1999
  int count = 0;
2000
//  childTblName = (char*)calloc(1, childTblCount * TSDB_TABLE_NAME_LEN);
2001 2002 2003
  char* pTblName = childTblName;
  while ((row = taos_fetch_row(res)) != NULL) {
    int32_t* len = taos_fetch_lengths(res);
2004
    tstrncpy(pTblName, (char *)row[0], len[0]+1);
2005 2006 2007 2008 2009 2010 2011
    //printf("==== sub table name: %s\n", pTblName);
    count++;
    if (count >= childTblCount - 1) {
      char *tmp = realloc(childTblName, (size_t)childTblCount*1.5*TSDB_TABLE_NAME_LEN+1);
      if (tmp != NULL) {
        childTblName = tmp;
        childTblCount = (int)(childTblCount*1.5);
2012 2013
        memset(childTblName + count*TSDB_TABLE_NAME_LEN, 0,
                (size_t)((childTblCount-count)*TSDB_TABLE_NAME_LEN));
2014 2015 2016 2017 2018 2019 2020 2021 2022 2023 2024
      } else {
        // exit, if allocate more memory failed
        printf("realloc fail for save child table name of %s.%s\n", dbName, sTblName);
        tmfree(childTblName);
        taos_free_result(res);
        taos_close(taos);
        exit(-1);
      }
    }
    pTblName = childTblName + count * TSDB_TABLE_NAME_LEN;    
  }
2025

2026 2027 2028 2029 2030 2031 2032
  *childTblCountOfSuperTbl = count;
  *childTblNameOfSuperTbl  = childTblName;

  taos_free_result(res);
  return 0;
}

2033 2034
static int getAllChildNameOfSuperTable(TAOS * taos, char* dbName,
        char* sTblName, char** childTblNameOfSuperTbl,
2035
        int* childTblCountOfSuperTbl) {
2036 2037 2038 2039 2040 2041

    return getChildNameOfSuperTableWithLimitAndOffset(taos, dbName, sTblName,
            childTblNameOfSuperTbl, childTblCountOfSuperTbl,
            -1, -1);
}

2042 2043
static int getSuperTableFromServer(TAOS * taos, char* dbName,
        SSuperTable*  superTbls) {
2044 2045 2046 2047
  char command[BUFFER_SIZE] = "\0";
  TAOS_RES * res;  
  TAOS_ROW row = NULL;
  int count = 0;
2048

2049
  //get schema use cmd: describe superTblName;
2050
  snprintf(command, BUFFER_SIZE, "describe %s.%s", dbName, superTbls->sTblName);
2051
  res = taos_query(taos, command);
2052 2053 2054
  int32_t code = taos_errno(res);
  if (code != 0) {
    printf("failed to run command %s\n", command);
2055
    taos_free_result(res);
2056 2057 2058 2059 2060 2061 2062 2063 2064 2065
    return -1;
  }

  int tagIndex = 0;
  int columnIndex = 0;
  TAOS_FIELD *fields = taos_fetch_fields(res);
  while ((row = taos_fetch_row(res)) != NULL) {
    if (0 == count) {
      count++;
      continue;
2066
    }
2067 2068

    if (strcmp((char *)row[TSDB_DESCRIBE_METRIC_NOTE_INDEX], "TAG") == 0) {
2069 2070 2071 2072 2073 2074 2075 2076 2077 2078 2079
      tstrncpy(superTbls->tags[tagIndex].field,
              (char *)row[TSDB_DESCRIBE_METRIC_FIELD_INDEX],
              fields[TSDB_DESCRIBE_METRIC_FIELD_INDEX].bytes);
      tstrncpy(superTbls->tags[tagIndex].dataType,
              (char *)row[TSDB_DESCRIBE_METRIC_TYPE_INDEX],
              fields[TSDB_DESCRIBE_METRIC_TYPE_INDEX].bytes);
      superTbls->tags[tagIndex].dataLen =
          *((int *)row[TSDB_DESCRIBE_METRIC_LENGTH_INDEX]);
      tstrncpy(superTbls->tags[tagIndex].note,
              (char *)row[TSDB_DESCRIBE_METRIC_NOTE_INDEX],
              fields[TSDB_DESCRIBE_METRIC_NOTE_INDEX].bytes);
2080
      tagIndex++;
2081
    } else {
2082 2083 2084 2085 2086 2087 2088 2089 2090 2091 2092
      tstrncpy(superTbls->columns[columnIndex].field,
              (char *)row[TSDB_DESCRIBE_METRIC_FIELD_INDEX],
              fields[TSDB_DESCRIBE_METRIC_FIELD_INDEX].bytes);
      tstrncpy(superTbls->columns[columnIndex].dataType,
              (char *)row[TSDB_DESCRIBE_METRIC_TYPE_INDEX],
              fields[TSDB_DESCRIBE_METRIC_TYPE_INDEX].bytes);
      superTbls->columns[columnIndex].dataLen =
          *((int *)row[TSDB_DESCRIBE_METRIC_LENGTH_INDEX]);
      tstrncpy(superTbls->columns[columnIndex].note,
              (char *)row[TSDB_DESCRIBE_METRIC_NOTE_INDEX],
              fields[TSDB_DESCRIBE_METRIC_NOTE_INDEX].bytes);
2093 2094 2095 2096 2097 2098 2099 2100 2101 2102 2103 2104 2105
      columnIndex++;
    }
    count++;
  }

  superTbls->columnCount = columnIndex;
  superTbls->tagCount    = tagIndex;
  taos_free_result(res);

  calcRowLen(superTbls);

  if (TBL_ALREADY_EXISTS == superTbls->childTblExists) {
    //get all child table name use cmd: select tbname from superTblName;  
2106
    getAllChildNameOfSuperTable(taos, dbName,
2107 2108 2109
            superTbls->sTblName,
            &superTbls->childTblName,
            &superTbls->childTblCount);
2110 2111 2112 2113
  }
  return 0;
}

2114
static int createSuperTable(TAOS * taos, char* dbName, SSuperTable*  superTbls, bool use_metric) {
2115
  char command[BUFFER_SIZE] = "\0";
2116

2117 2118 2119 2120 2121 2122 2123
  char cols[STRING_LEN] = "\0";
  int colIndex;
  int len = 0;

  int  lenOfOneRow = 0;
  for (colIndex = 0; colIndex < superTbls->columnCount; colIndex++) {
    char* dataType = superTbls->columns[colIndex].dataType;
2124
 
2125
    if (strcasecmp(dataType, "BINARY") == 0) {
2126 2127 2128
      len += snprintf(cols + len, STRING_LEN - len,
          ", col%d %s(%d)", colIndex, "BINARY",
          superTbls->columns[colIndex].dataLen);
2129 2130
      lenOfOneRow += superTbls->columns[colIndex].dataLen + 3;
    } else if (strcasecmp(dataType, "NCHAR") == 0) {
2131 2132 2133
      len += snprintf(cols + len, STRING_LEN - len,
          ", col%d %s(%d)", colIndex, "NCHAR",
          superTbls->columns[colIndex].dataLen);
2134 2135 2136 2137 2138 2139 2140 2141 2142 2143 2144 2145 2146 2147 2148 2149 2150 2151 2152
      lenOfOneRow += superTbls->columns[colIndex].dataLen + 3;
    } else if (strcasecmp(dataType, "INT") == 0)  {
      len += snprintf(cols + len, STRING_LEN - len, ", col%d %s", colIndex, "INT");
      lenOfOneRow += 11;
    } else if (strcasecmp(dataType, "BIGINT") == 0)  {
      len += snprintf(cols + len, STRING_LEN - len, ", col%d %s", colIndex, "BIGINT");
      lenOfOneRow += 21;
    } else if (strcasecmp(dataType, "SMALLINT") == 0)  {
      len += snprintf(cols + len, STRING_LEN - len, ", col%d %s", colIndex, "SMALLINT");
      lenOfOneRow += 6;
    } else if (strcasecmp(dataType, "TINYINT") == 0)  {
      len += snprintf(cols + len, STRING_LEN - len, ", col%d %s", colIndex, "TINYINT");
      lenOfOneRow += 4;
    } else if (strcasecmp(dataType, "BOOL") == 0)  {
      len += snprintf(cols + len, STRING_LEN - len, ", col%d %s", colIndex, "BOOL");
      lenOfOneRow += 6;
    } else if (strcasecmp(dataType, "FLOAT") == 0) {
      len += snprintf(cols + len, STRING_LEN - len, ", col%d %s", colIndex, "FLOAT");
      lenOfOneRow += 22;
2153
    } else if (strcasecmp(dataType, "DOUBLE") == 0) {
2154 2155
      len += snprintf(cols + len, STRING_LEN - len, ", col%d %s", colIndex, "DOUBLE");
      lenOfOneRow += 42;
2156
    }  else if (strcasecmp(dataType, "TIMESTAMP") == 0) {
2157 2158 2159 2160 2161 2162 2163 2164 2165 2166 2167 2168 2169
      len += snprintf(cols + len, STRING_LEN - len, ", col%d %s", colIndex, "TIMESTAMP");
      lenOfOneRow += 21;
    } else {
      taos_close(taos);
      printf("config error data type : %s\n", dataType);
      exit(-1);
    }
  }

  superTbls->lenOfOneRow = lenOfOneRow + 20; // timestamp
  //printf("%s.%s column count:%d, column length:%d\n\n", g_Dbs.db[i].dbName, g_Dbs.db[i].superTbls[j].sTblName, g_Dbs.db[i].superTbls[j].columnCount, lenOfOneRow);

  // save for creating child table
2170 2171
  superTbls->colsOfCreateChildTable = (char*)calloc(len+20, 1);
  if (NULL == superTbls->colsOfCreateChildTable) {
2172 2173 2174 2175
    printf("Failed when calloc, size:%d", len+1);
    taos_close(taos);
    exit(-1);
  }
2176
  snprintf(superTbls->colsOfCreateChildTable, len+20, "(ts timestamp%s)", cols);
2177
  verbosePrint("%s() LN%d: %s\n", __func__, __LINE__, superTbls->colsOfCreateChildTable);
2178 2179 2180 2181 2182 2183 2184 2185 2186 2187

  if (use_metric) {
    char tags[STRING_LEN] = "\0";
    int tagIndex;
    len = 0;

    int lenOfTagOfOneRow = 0;
    len += snprintf(tags + len, STRING_LEN - len, "(");
    for (tagIndex = 0; tagIndex < superTbls->tagCount; tagIndex++) {
      char* dataType = superTbls->tags[tagIndex].dataType;
2188

2189
      if (strcasecmp(dataType, "BINARY") == 0) {
2190 2191
        len += snprintf(tags + len, STRING_LEN - len, "t%d %s(%d), ", tagIndex,
                "BINARY", superTbls->tags[tagIndex].dataLen);
2192 2193
        lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 3;
      } else if (strcasecmp(dataType, "NCHAR") == 0) {
2194 2195
        len += snprintf(tags + len, STRING_LEN - len, "t%d %s(%d), ", tagIndex,
                "NCHAR", superTbls->tags[tagIndex].dataLen);
2196 2197
        lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 3;
      } else if (strcasecmp(dataType, "INT") == 0)  {
2198 2199
        len += snprintf(tags + len, STRING_LEN - len, "t%d %s, ", tagIndex,
                "INT");
2200 2201
        lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 11;
      } else if (strcasecmp(dataType, "BIGINT") == 0)  {
2202 2203
        len += snprintf(tags + len, STRING_LEN - len, "t%d %s, ", tagIndex,
                "BIGINT");
2204 2205
        lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 21;
      } else if (strcasecmp(dataType, "SMALLINT") == 0)  {
2206 2207
        len += snprintf(tags + len, STRING_LEN - len, "t%d %s, ", tagIndex,
                "SMALLINT");
2208 2209
        lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 6;
      } else if (strcasecmp(dataType, "TINYINT") == 0)  {
2210 2211
        len += snprintf(tags + len, STRING_LEN - len, "t%d %s, ", tagIndex,
                "TINYINT");
2212 2213
        lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 4;
      } else if (strcasecmp(dataType, "BOOL") == 0)  {
2214 2215
        len += snprintf(tags + len, STRING_LEN - len, "t%d %s, ", tagIndex,
                "BOOL");
2216 2217
        lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 6;
      } else if (strcasecmp(dataType, "FLOAT") == 0) {
2218 2219
        len += snprintf(tags + len, STRING_LEN - len, "t%d %s, ", tagIndex,
                "FLOAT");
2220 2221
        lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 22;
      } else if (strcasecmp(dataType, "DOUBLE") == 0) { 
2222 2223
        len += snprintf(tags + len, STRING_LEN - len, "t%d %s, ", tagIndex,
                "DOUBLE");
2224 2225 2226 2227 2228 2229 2230 2231 2232 2233 2234
        lenOfTagOfOneRow += superTbls->tags[tagIndex].dataLen + 42;
      } else {
        taos_close(taos);
        printf("config error tag type : %s\n", dataType);
        exit(-1);
      }
    }
    len -= 2;
    len += snprintf(tags + len, STRING_LEN - len, ")");

    superTbls->lenOfTagOfOneRow = lenOfTagOfOneRow;
2235 2236 2237 2238

    snprintf(command, BUFFER_SIZE,
            "create table if not exists %s.%s (ts timestamp%s) tags %s",
            dbName, superTbls->sTblName, cols, tags);
2239
    verbosePrint("%s() LN%d: %s\n", __func__, __LINE__, command);
2240

2241
    if (0 != queryDbExec(taos, command, NO_INSERT_TYPE)) {
2242 2243
        fprintf(stderr, "create supertable %s failed!\n\n",
                superTbls->sTblName);
2244
        return -1;
2245
    }
2246
    debugPrint("create supertable %s success!\n\n", superTbls->sTblName);
2247 2248 2249 2250
  }
  return 0;
}

2251
static int createDatabases() {
2252 2253 2254 2255 2256
  TAOS * taos = NULL;
  int    ret = 0;
  taos = taos_connect(g_Dbs.host, g_Dbs.user, g_Dbs.password, NULL, g_Dbs.port);
  if (taos == NULL) {
    fprintf(stderr, "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL));
2257
    return -1;
2258 2259 2260 2261 2262 2263
  }
  char command[BUFFER_SIZE] = "\0";

  for (int i = 0; i < g_Dbs.dbCount; i++) {   
    if (g_Dbs.db[i].drop) {
      sprintf(command, "drop database if exists %s;", g_Dbs.db[i].dbName);
2264
      verbosePrint("%s() %d command: %s\n", __func__, __LINE__, command);
2265 2266 2267 2268 2269
      if (0 != queryDbExec(taos, command, NO_INSERT_TYPE)) {
        taos_close(taos);
        return -1;
      }
    }
2270

2271
    int dataLen = 0;
2272
    dataLen += snprintf(command + dataLen, 
2273
        BUFFER_SIZE - dataLen, "create database if not exists %s", g_Dbs.db[i].dbName);
2274 2275

    if (g_Dbs.db[i].dbCfg.blocks > 0) {
2276
      dataLen += snprintf(command + dataLen,
2277
          BUFFER_SIZE - dataLen, " blocks %d", g_Dbs.db[i].dbCfg.blocks);
2278 2279
    }
    if (g_Dbs.db[i].dbCfg.cache > 0) {
2280
      dataLen += snprintf(command + dataLen,
2281
          BUFFER_SIZE - dataLen, " cache %d", g_Dbs.db[i].dbCfg.cache);
2282 2283
    }
    if (g_Dbs.db[i].dbCfg.days > 0) {
2284
      dataLen += snprintf(command + dataLen,
2285
          BUFFER_SIZE - dataLen, " days %d", g_Dbs.db[i].dbCfg.days);
2286 2287
    }
    if (g_Dbs.db[i].dbCfg.keep > 0) {
2288
      dataLen += snprintf(command + dataLen,
2289
          BUFFER_SIZE - dataLen, " keep %d", g_Dbs.db[i].dbCfg.keep);
2290
    }
2291 2292
    if (g_Dbs.db[i].dbCfg.quorum > 1) {
      dataLen += snprintf(command + dataLen,
2293
          BUFFER_SIZE - dataLen, " quorum %d", g_Dbs.db[i].dbCfg.quorum);
2294 2295
    }
    if (g_Dbs.db[i].dbCfg.replica > 0) {
2296
      dataLen += snprintf(command + dataLen,
2297
          BUFFER_SIZE - dataLen, " replica %d", g_Dbs.db[i].dbCfg.replica);
2298 2299
    }
    if (g_Dbs.db[i].dbCfg.update > 0) {
2300
      dataLen += snprintf(command + dataLen,
2301
          BUFFER_SIZE - dataLen, " update %d", g_Dbs.db[i].dbCfg.update);
2302 2303
    }
    //if (g_Dbs.db[i].dbCfg.maxtablesPerVnode > 0) {
2304 2305
    //  dataLen += snprintf(command + dataLen,
    //  BUFFER_SIZE - dataLen, "tables %d ", g_Dbs.db[i].dbCfg.maxtablesPerVnode);
2306 2307
    //}
    if (g_Dbs.db[i].dbCfg.minRows > 0) {
2308
      dataLen += snprintf(command + dataLen,
2309
          BUFFER_SIZE - dataLen, " minrows %d", g_Dbs.db[i].dbCfg.minRows);
2310 2311
    }
    if (g_Dbs.db[i].dbCfg.maxRows > 0) {
2312
      dataLen += snprintf(command + dataLen,
2313
          BUFFER_SIZE - dataLen, " maxrows %d", g_Dbs.db[i].dbCfg.maxRows);
2314 2315
    }
    if (g_Dbs.db[i].dbCfg.comp > 0) {
2316
      dataLen += snprintf(command + dataLen,
2317
          BUFFER_SIZE - dataLen, " comp %d", g_Dbs.db[i].dbCfg.comp);
2318 2319
    }
    if (g_Dbs.db[i].dbCfg.walLevel > 0) {
2320
      dataLen += snprintf(command + dataLen,
2321
          BUFFER_SIZE - dataLen, " wal %d", g_Dbs.db[i].dbCfg.walLevel);
2322
    }
2323
    if (g_Dbs.db[i].dbCfg.cacheLast > 0) {
2324
      dataLen += snprintf(command + dataLen,
2325
          BUFFER_SIZE - dataLen, " cachelast %d", g_Dbs.db[i].dbCfg.cacheLast);
2326
    }
2327
    if (g_Dbs.db[i].dbCfg.fsync > 0) {
2328 2329
      dataLen += snprintf(command + dataLen, BUFFER_SIZE - dataLen,
              " fsync %d", g_Dbs.db[i].dbCfg.fsync);
2330
    }
2331 2332
    if ((0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "ms", strlen("ms")))
            || (0 == strncasecmp(g_Dbs.db[i].dbCfg.precision, "us", strlen("us")))) {
2333
      dataLen += snprintf(command + dataLen, BUFFER_SIZE - dataLen,
2334
              " precision \'%s\';", g_Dbs.db[i].dbCfg.precision);
2335
    }
2336

2337
    debugPrint("%s() %d command: %s\n", __func__, __LINE__, command);
2338 2339
    if (0 != queryDbExec(taos, command, NO_INSERT_TYPE)) {
      taos_close(taos);
2340
      fprintf(stderr, "\ncreate database %s failed!\n\n", g_Dbs.db[i].dbName);
2341 2342 2343 2344
      return -1;
    }
    printf("\ncreate database %s success!\n\n", g_Dbs.db[i].dbName);

2345
    debugPrint("%s() %d supertbl count:%d\n", __func__, __LINE__, g_Dbs.db[i].superTblCount);
2346 2347 2348
    for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) {
      // describe super table, if exists
      sprintf(command, "describe %s.%s;", g_Dbs.db[i].dbName, g_Dbs.db[i].superTbls[j].sTblName);
2349
      verbosePrint("%s() %d command: %s\n", __func__, __LINE__, command);
2350 2351 2352 2353 2354 2355 2356 2357 2358
      if (0 != queryDbExec(taos, command, NO_INSERT_TYPE)) {
        g_Dbs.db[i].superTbls[j].superTblExists = TBL_NO_EXISTS;
        ret = createSuperTable(taos, g_Dbs.db[i].dbName, &g_Dbs.db[i].superTbls[j], g_Dbs.use_metric);
      } else {      
        g_Dbs.db[i].superTbls[j].superTblExists = TBL_ALREADY_EXISTS;
        ret = getSuperTableFromServer(taos, g_Dbs.db[i].dbName, &g_Dbs.db[i].superTbls[j]);
      }

      if (0 != ret) {
2359
        printf("\ncreate super table %d failed!\n\n", j);
2360 2361 2362 2363 2364 2365 2366 2367 2368 2369
        taos_close(taos);
        return -1;
      }
    }    
  }

  taos_close(taos);
  return 0;
}

2370
static void* createTable(void *sarg) 
2371 2372 2373 2374 2375 2376
{  
  threadInfo *winfo = (threadInfo *)sarg; 
  SSuperTable* superTblInfo = winfo->superTblInfo;

  int64_t  lastPrintTime = taosGetTimestampMs();

2377 2378 2379 2380 2381 2382
  int buff_len;
  if (superTblInfo)
    buff_len = superTblInfo->maxSqlLen;
  else
    buff_len = BUFFER_SIZE;

2383
  char *buffer = calloc(buff_len, 1);
2384 2385 2386 2387
  if (buffer == NULL) {
    fprintf(stderr, "Memory allocated failed!");
    exit(-1);
  }
2388 2389 2390 2391 2392 2393

  int len = 0;
  int batchNum = 0;
  //printf("Creating table from %d to %d\n", winfo->start_table_id, winfo->end_table_id);
  for (int i = winfo->start_table_id; i <= winfo->end_table_id; i++) {
    if (0 == g_Dbs.use_metric) {
2394
      snprintf(buffer, buff_len, 
2395 2396
              "create table if not exists %s.%s%d %s;",
              winfo->db_name,
2397 2398
              g_args.tb_prefix, i,
              winfo->cols);
2399 2400 2401
    } else {
      if (0 == len) {  
        batchNum = 0;
2402
        memset(buffer, 0, buff_len);
2403
        len += snprintf(buffer + len,
2404
                buff_len - len, "create table ");
2405
      }
2406

2407 2408 2409 2410
      char* tagsValBuf = NULL;
      if (0 == superTblInfo->tagSource) {
        tagsValBuf = generateTagVaulesForStb(superTblInfo);
      } else {
2411 2412 2413
        tagsValBuf = getTagValueFromTagSample(
                superTblInfo,
                i % superTblInfo->tagSampleCount);
2414 2415 2416 2417 2418 2419
      }
      if (NULL == tagsValBuf) {
        free(buffer);
        return NULL;
      }
      
2420 2421 2422 2423 2424 2425
      len += snprintf(buffer + len,
              superTblInfo->maxSqlLen - len,
              "if not exists %s.%s%d using %s.%s tags %s ",
              winfo->db_name, superTblInfo->childTblPrefix,
              i, winfo->db_name,
              superTblInfo->sTblName, tagsValBuf);
2426 2427 2428
      free(tagsValBuf);
      batchNum++;

2429 2430 2431
      if ((batchNum < superTblInfo->batchCreateTableNum)
              && ((superTblInfo->maxSqlLen - len) 
                  >= (superTblInfo->lenOfTagOfOneRow + 256))) {
2432 2433 2434 2435 2436
        continue;
      }
    }

    len = 0;
2437
    verbosePrint("%s() LN%d %s\n", __func__, __LINE__, buffer);
2438
    if (0 != queryDbExec(winfo->taos, buffer, NO_INSERT_TYPE)){
2439
      fprintf(stderr, "queryDbExec() failed. buffer:\n%s\n", buffer);
2440 2441 2442 2443 2444 2445
      free(buffer);
      return NULL;
    }

    int64_t  currentPrintTime = taosGetTimestampMs();
    if (currentPrintTime - lastPrintTime > 30*1000) {
2446 2447
      printf("thread[%d] already create %d - %d tables\n",
              winfo->threadID, winfo->start_table_id, i);
2448 2449 2450 2451 2452
      lastPrintTime = currentPrintTime;
    }
  }
  
  if (0 != len) {
2453
    verbosePrint("%s() %d buffer: %s\n", __func__, __LINE__, buffer);
2454 2455 2456
    if (0 != queryDbExec(winfo->taos, buffer, NO_INSERT_TYPE)) {
      fprintf(stderr, "queryDbExec() failed. buffer:\n%s\n", buffer);
    }
2457
  }
2458

2459 2460 2461 2462
  free(buffer);
  return NULL;
}

2463
int startMultiThreadCreateChildTable(
2464
        char* cols, int threads, int startFrom, int ntables,
2465
        char* db_name, SSuperTable* superTblInfo) {
2466 2467 2468 2469 2470 2471 2472 2473 2474 2475 2476 2477 2478 2479 2480 2481 2482 2483 2484 2485
  pthread_t *pids = malloc(threads * sizeof(pthread_t));
  threadInfo *infos = malloc(threads * sizeof(threadInfo));

  if ((NULL == pids) || (NULL == infos)) {
    printf("malloc failed\n");
    exit(-1);
  }

  if (threads < 1) {
    threads = 1;
  }

  int a = ntables / threads;
  if (a < 1) {
    threads = ntables;
    a = 1;
  }

  int b = 0;
  b = ntables % threads;
2486
 
2487
  int last = startFrom;
2488 2489 2490
  for (int i = 0; i < threads; i++) {
    threadInfo *t_info = infos + i;
    t_info->threadID = i;
H
Hui Li 已提交
2491
    tstrncpy(t_info->db_name, db_name, MAX_DB_NAME_SIZE);
2492
    t_info->superTblInfo = superTblInfo;
2493
    verbosePrint("%s() %d db_name: %s\n", __func__, __LINE__, db_name);
2494 2495 2496 2497 2498 2499
    t_info->taos = taos_connect(
            g_Dbs.host,
            g_Dbs.user,
            g_Dbs.password,
            db_name,
            g_Dbs.port);
2500 2501
    if (t_info->taos == NULL) {
      fprintf(stderr, "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL));
2502 2503
      free(pids);
      free(infos);  
2504 2505
      return -1;
    }
H
hzcheng 已提交
2506 2507 2508
    t_info->start_table_id = last;
    t_info->end_table_id = i < b ? last + a : last + a - 1;
    last = t_info->end_table_id + 1;
2509 2510
    t_info->use_metric = 1;
    t_info->cols = cols;
2511
    t_info->minDelay = INT16_MAX;
2512 2513 2514 2515 2516 2517 2518 2519 2520 2521 2522 2523 2524 2525
    pthread_create(pids + i, NULL, createTable, t_info);
  }
  
  for (int i = 0; i < threads; i++) {
    pthread_join(pids[i], NULL);
  }

  for (int i = 0; i < threads; i++) {
    threadInfo *t_info = infos + i;
    taos_close(t_info->taos);
  }

  free(pids);
  free(infos);  
2526 2527

  return 0;
2528 2529 2530
}


2531 2532 2533 2534
static void createChildTables() {
    char tblColsBuf[MAX_SQL_SIZE];
    int len;

2535
  for (int i = 0; i < g_Dbs.dbCount; i++) {
2536
    if (g_Dbs.db[i].superTblCount > 0) {
2537
        // with super table
2538 2539
      for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) {
        if ((AUTO_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable)
2540
              || (TBL_ALREADY_EXISTS == g_Dbs.db[i].superTbls[j].childTblExists)) {
2541 2542 2543
          continue;
        }

2544
        verbosePrint("%s() LN%d: %s\n", __func__, __LINE__,
2545
                g_Dbs.db[i].superTbls[j].colsOfCreateChildTable);
2546 2547
        int startFrom = 0;
        g_totalChildTables += g_Dbs.db[i].superTbls[j].childTblCount;
2548 2549 2550

        verbosePrint("%s() LN%d: create %d child tables from %d\n", __func__, __LINE__,
              g_totalChildTables, startFrom);  
2551
        startMultiThreadCreateChildTable(
2552
              g_Dbs.db[i].superTbls[j].colsOfCreateChildTable,
2553
              g_Dbs.threadCountByCreateTbl,
2554 2555
              startFrom,
              g_totalChildTables,
2556
              g_Dbs.db[i].dbName, &(g_Dbs.db[i].superTbls[j]));
2557 2558
      }
    } else {
2559 2560
        // normal table
        len = snprintf(tblColsBuf, MAX_SQL_SIZE, "(TS TIMESTAMP");
2561 2562 2563
        int j = 0;
        while (g_args.datatype[j]) {
            if ((strncasecmp(g_args.datatype[j], "BINARY", strlen("BINARY")) == 0)
2564 2565 2566 2567
                    || (strncasecmp(g_args.datatype[j],
                        "NCHAR", strlen("NCHAR")) == 0)) {
                len = snprintf(tblColsBuf + len, MAX_SQL_SIZE,
                        ", COL%d %s(60)", j, g_args.datatype[j]);
2568
            } else {
2569 2570
                len = snprintf(tblColsBuf + len, MAX_SQL_SIZE,
                        ", COL%d %s", j, g_args.datatype[j]);
2571 2572
            }
            len = strlen(tblColsBuf);
2573
            j++;
2574 2575
        }

2576
        len = snprintf(tblColsBuf + len, MAX_SQL_SIZE - len, ")");
2577

2578 2579
        verbosePrint("%s() LN%d: dbName: %s num of tb: %d schema: %s\n",
                __func__, __LINE__,
2580
                g_Dbs.db[i].dbName, g_args.num_of_tables, tblColsBuf);
2581
        startMultiThreadCreateChildTable(
2582
              tblColsBuf,
2583
              g_Dbs.threadCountByCreateTbl,
2584
              0,
2585
              g_args.num_of_tables,
2586 2587 2588
              g_Dbs.db[i].dbName,
              NULL);
    }
2589 2590 2591 2592 2593 2594 2595 2596 2597 2598
  }
}

/*
  Read 10000 lines at most. If more than 10000 lines, continue to read after using
*/
int readTagFromCsvFileToMem(SSuperTable  * superTblInfo) {
  size_t  n = 0;
  ssize_t readLen = 0;
  char *  line = NULL;
2599

2600 2601
  FILE *fp = fopen(superTblInfo->tagsFile, "r");
  if (fp == NULL) {
2602 2603
    printf("Failed to open tags file: %s, reason:%s\n",
            superTblInfo->tagsFile, strerror(errno));
2604 2605 2606 2607 2608 2609 2610
    return -1;
  }

  if (superTblInfo->tagDataBuf) {
    free(superTblInfo->tagDataBuf);
    superTblInfo->tagDataBuf = NULL;
  }
2611

2612 2613 2614 2615 2616 2617 2618 2619 2620
  int tagCount = 10000;
  int count = 0;
  char* tagDataBuf = calloc(1, superTblInfo->lenOfTagOfOneRow * tagCount);
  if (tagDataBuf == NULL) {
    printf("Failed to calloc, reason:%s\n", strerror(errno));
    fclose(fp);
    return -1;
  }

2621
  while ((readLen = tgetline(&line, &n, fp)) != -1) {
2622 2623 2624 2625 2626 2627 2628 2629 2630 2631 2632 2633
    if (('\r' == line[readLen - 1]) || ('\n' == line[readLen - 1])) {
      line[--readLen] = 0;
    }

    if (readLen == 0) {
      continue;
    }

    memcpy(tagDataBuf + count * superTblInfo->lenOfTagOfOneRow, line, readLen);
    count++;

    if (count >= tagCount - 1) {
2634 2635
      char *tmp = realloc(tagDataBuf,
              (size_t)tagCount*1.5*superTblInfo->lenOfTagOfOneRow);
2636 2637 2638
      if (tmp != NULL) {
        tagDataBuf = tmp;
        tagCount = (int)(tagCount*1.5);
2639 2640
        memset(tagDataBuf + count*superTblInfo->lenOfTagOfOneRow,
                0, (size_t)((tagCount-count)*superTblInfo->lenOfTagOfOneRow));
2641 2642 2643 2644 2645 2646 2647 2648 2649 2650 2651 2652 2653 2654 2655 2656 2657 2658 2659 2660 2661 2662 2663 2664 2665 2666 2667 2668
      } else {
        // exit, if allocate more memory failed
        printf("realloc fail for save tag val from %s\n", superTblInfo->tagsFile);
        tmfree(tagDataBuf);
        free(line);
        fclose(fp);
        return -1;
      }
    }
  }

  superTblInfo->tagDataBuf = tagDataBuf;
  superTblInfo->tagSampleCount = count;

  free(line);
  fclose(fp);
  return 0;
}

int readSampleFromJsonFileToMem(SSuperTable  * superTblInfo) {
  // TODO
  return 0;
}


/*
  Read 10000 lines at most. If more than 10000 lines, continue to read after using
*/
2669 2670
static int readSampleFromCsvFileToMem(
        SSuperTable* superTblInfo) {
2671 2672 2673 2674
  size_t  n = 0;
  ssize_t readLen = 0;
  char *  line = NULL;
  int getRows = 0;
2675 2676 2677
 
  FILE*  fp = fopen(superTblInfo->sampleFile, "r");
  if (fp == NULL) {
2678
      fprintf(stderr, "Failed to open sample file: %s, reason:%s\n",
2679 2680 2681
              superTblInfo->sampleFile, strerror(errno));
      return -1;
  }
2682

2683 2684
  memset(superTblInfo->sampleDataBuf, 0,
          MAX_SAMPLES_ONCE_FROM_FILE * superTblInfo->lenOfOneRow);
2685
  while (1) {
2686
    readLen = tgetline(&line, &n, fp);
2687 2688
    if (-1 == readLen) {
      if(0 != fseek(fp, 0, SEEK_SET)) {
2689
        fprintf(stderr, "Failed to fseek file: %s, reason:%s\n",
2690
                superTblInfo->sampleFile, strerror(errno));
2691
        fclose(fp);
2692 2693 2694 2695 2696 2697 2698 2699 2700 2701 2702 2703 2704 2705
        return -1;
      }
      continue;
    }
  
    if (('\r' == line[readLen - 1]) || ('\n' == line[readLen - 1])) {
      line[--readLen] = 0;
    }

    if (readLen == 0) {
      continue;
    }

    if (readLen > superTblInfo->lenOfOneRow) {
2706 2707
      printf("sample row len[%d] overflow define schema len[%d], so discard this row\n",
              (int32_t)readLen, superTblInfo->lenOfOneRow);
2708 2709 2710
      continue;
    }

2711 2712
    memcpy(superTblInfo->sampleDataBuf + getRows * superTblInfo->lenOfOneRow,
          line, readLen);
2713 2714 2715 2716 2717 2718 2719
    getRows++;

    if (getRows == MAX_SAMPLES_ONCE_FROM_FILE) {
      break;
    }
  }

2720
  fclose(fp);
2721 2722 2723 2724 2725 2726 2727 2728 2729 2730 2731 2732 2733 2734 2735 2736 2737 2738 2739 2740 2741 2742 2743 2744
  tmfree(line);
  return 0;
}

/*
void readSampleFromFileToMem(SSuperTable  * supterTblInfo) {
  int ret;
  if (0 == strncasecmp(supterTblInfo->sampleFormat, "csv", 3)) {
    ret = readSampleFromCsvFileToMem(supterTblInfo);
  } else if (0 == strncasecmp(supterTblInfo->sampleFormat, "json", 4)) {
    ret = readSampleFromJsonFileToMem(supterTblInfo);
  }

  if (0 != ret) {
    exit(-1);
  }
}
*/
static bool getColumnAndTagTypeFromInsertJsonFile(cJSON* stbInfo, SSuperTable* superTbls) {  
  bool  ret = false;
  
  // columns 
  cJSON *columns = cJSON_GetObjectItem(stbInfo, "columns");
  if (columns && columns->type != cJSON_Array) {
2745
    printf("ERROR: failed to read json, columns not found\n");
2746 2747 2748 2749 2750 2751 2752 2753 2754
    goto PARSE_OVER;
  } else if (NULL == columns) {
    superTbls->columnCount = 0;
    superTbls->tagCount    = 0;
    return true;
  }
  
  int columnSize = cJSON_GetArraySize(columns);
  if (columnSize > MAX_COLUMN_COUNT) {
2755
    printf("ERROR: failed to read json, column size overflow, max column size is %d\n",
2756
            MAX_COLUMN_COUNT);
2757 2758 2759 2760 2761 2762
    goto PARSE_OVER;
  }

  int count = 1;
  int index = 0;
  StrColumn    columnCase;
2763

2764 2765 2766 2767 2768 2769 2770 2771 2772 2773
  //superTbls->columnCount = columnSize;  
  for (int k = 0; k < columnSize; ++k) {
    cJSON* column = cJSON_GetArrayItem(columns, k);
    if (column == NULL) continue;

    count = 1;
    cJSON* countObj = cJSON_GetObjectItem(column, "count");
    if (countObj && countObj->type == cJSON_Number) {
      count = countObj->valueint;    
    } else if (countObj && countObj->type != cJSON_Number) {
2774
      printf("ERROR: failed to read json, column count not found\n");
2775 2776 2777 2778 2779 2780 2781 2782 2783
      goto PARSE_OVER;
    } else {
      count = 1;
    }

    // column info 
    memset(&columnCase, 0, sizeof(StrColumn));
    cJSON *dataType = cJSON_GetObjectItem(column, "type");
    if (!dataType || dataType->type != cJSON_String || dataType->valuestring == NULL) {
2784
      printf("ERROR: failed to read json, column type not found\n");
2785 2786
      goto PARSE_OVER;
    }
2787 2788
    //tstrncpy(superTbls->columns[k].dataType, dataType->valuestring, MAX_TB_NAME_SIZE);
    tstrncpy(columnCase.dataType, dataType->valuestring, MAX_TB_NAME_SIZE);
2789 2790 2791 2792 2793
            
    cJSON* dataLen = cJSON_GetObjectItem(column, "len");
    if (dataLen && dataLen->type == cJSON_Number) {
      columnCase.dataLen = dataLen->valueint;    
    } else if (dataLen && dataLen->type != cJSON_Number) {
2794
      printf("ERROR: failed to read json, column len not found\n");
2795 2796 2797 2798 2799 2800
      goto PARSE_OVER;
    } else {
      columnCase.dataLen = 8;
    }
    
    for (int n = 0; n < count; ++n) {
2801
      tstrncpy(superTbls->columns[index].dataType, columnCase.dataType, MAX_TB_NAME_SIZE);
2802 2803 2804 2805 2806 2807 2808 2809 2810 2811 2812
      superTbls->columns[index].dataLen = columnCase.dataLen; 
      index++;
    }
  }  
  superTbls->columnCount = index;  
  
  count = 1;
  index = 0;
  // tags 
  cJSON *tags = cJSON_GetObjectItem(stbInfo, "tags");
  if (!tags || tags->type != cJSON_Array) {
2813
    printf("ERROR: failed to read json, tags not found\n");
2814 2815 2816 2817 2818
    goto PARSE_OVER;
  }
  
  int tagSize = cJSON_GetArraySize(tags);
  if (tagSize > MAX_TAG_COUNT) {
2819
    printf("ERROR: failed to read json, tags size overflow, max tag size is %d\n", MAX_TAG_COUNT);
2820 2821 2822 2823 2824 2825 2826 2827 2828 2829 2830 2831 2832
    goto PARSE_OVER;
  }
  
  //superTbls->tagCount = tagSize;  
  for (int k = 0; k < tagSize; ++k) {
    cJSON* tag = cJSON_GetArrayItem(tags, k);
    if (tag == NULL) continue;
    
    count = 1;
    cJSON* countObj = cJSON_GetObjectItem(tag, "count");
    if (countObj && countObj->type == cJSON_Number) {
      count = countObj->valueint;    
    } else if (countObj && countObj->type != cJSON_Number) {
2833
      printf("ERROR: failed to read json, column count not found\n");
2834 2835 2836 2837 2838 2839 2840 2841 2842
      goto PARSE_OVER;
    } else {
      count = 1;
    }

    // column info 
    memset(&columnCase, 0, sizeof(StrColumn));
    cJSON *dataType = cJSON_GetObjectItem(tag, "type");
    if (!dataType || dataType->type != cJSON_String || dataType->valuestring == NULL) {
2843
      printf("ERROR: failed to read json, tag type not found\n");
2844 2845
      goto PARSE_OVER;
    }
2846
    tstrncpy(columnCase.dataType, dataType->valuestring, MAX_TB_NAME_SIZE);
2847 2848 2849 2850 2851
            
    cJSON* dataLen = cJSON_GetObjectItem(tag, "len");
    if (dataLen && dataLen->type == cJSON_Number) {
      columnCase.dataLen = dataLen->valueint;    
    } else if (dataLen && dataLen->type != cJSON_Number) {
2852
      printf("ERROR: failed to read json, column len not found\n");
2853 2854 2855 2856 2857 2858
      goto PARSE_OVER;
    } else {
      columnCase.dataLen = 0;
    }  
    
    for (int n = 0; n < count; ++n) {
2859
      tstrncpy(superTbls->tags[index].dataType, columnCase.dataType, MAX_TB_NAME_SIZE);
2860 2861 2862 2863 2864 2865 2866 2867 2868 2869 2870 2871 2872 2873 2874 2875 2876 2877 2878 2879
      superTbls->tags[index].dataLen = columnCase.dataLen; 
      index++;
    }
  }      
  superTbls->tagCount = index;

  ret = true;

PARSE_OVER:
  //free(content);
  //cJSON_Delete(root);
  //fclose(fp);
  return ret;
}

static bool getMetaFromInsertJsonFile(cJSON* root) {
  bool  ret = false;

  cJSON* cfgdir = cJSON_GetObjectItem(root, "cfgdir");
  if (cfgdir && cfgdir->type == cJSON_String && cfgdir->valuestring != NULL) {
2880
    tstrncpy(g_Dbs.cfgDir, cfgdir->valuestring, MAX_FILE_NAME_LEN);
2881 2882 2883 2884
  }

  cJSON* host = cJSON_GetObjectItem(root, "host");
  if (host && host->type == cJSON_String && host->valuestring != NULL) {
2885
    tstrncpy(g_Dbs.host, host->valuestring, MAX_DB_NAME_SIZE);
2886
  } else if (!host) {
2887
    tstrncpy(g_Dbs.host, "127.0.0.1", MAX_DB_NAME_SIZE);
2888
  } else {
2889
    printf("ERROR: failed to read json, host not found\n");
2890 2891 2892 2893 2894 2895 2896 2897 2898 2899 2900 2901
    goto PARSE_OVER;
  }

  cJSON* port = cJSON_GetObjectItem(root, "port");
  if (port && port->type == cJSON_Number) {
    g_Dbs.port = port->valueint;
  } else if (!port) {
    g_Dbs.port = 6030;
  }

  cJSON* user = cJSON_GetObjectItem(root, "user");
  if (user && user->type == cJSON_String && user->valuestring != NULL) {
2902
    tstrncpy(g_Dbs.user, user->valuestring, MAX_DB_NAME_SIZE);   
2903
  } else if (!user) {
2904
    tstrncpy(g_Dbs.user, "root", MAX_DB_NAME_SIZE);
2905 2906 2907 2908
  }

  cJSON* password = cJSON_GetObjectItem(root, "password");
  if (password && password->type == cJSON_String && password->valuestring != NULL) {
2909
    tstrncpy(g_Dbs.password, password->valuestring, MAX_DB_NAME_SIZE);
2910
  } else if (!password) {
2911
    tstrncpy(g_Dbs.password, "taosdata", MAX_DB_NAME_SIZE);
2912 2913 2914 2915
  }

  cJSON* resultfile = cJSON_GetObjectItem(root, "result_file");
  if (resultfile && resultfile->type == cJSON_String && resultfile->valuestring != NULL) {
2916
    tstrncpy(g_Dbs.resultFile, resultfile->valuestring, MAX_FILE_NAME_LEN);
2917
  } else if (!resultfile) {
2918
    tstrncpy(g_Dbs.resultFile, "./insert_res.txt", MAX_FILE_NAME_LEN);
2919 2920 2921 2922 2923 2924 2925 2926
  }

  cJSON* threads = cJSON_GetObjectItem(root, "thread_count");
  if (threads && threads->type == cJSON_Number) {
    g_Dbs.threadCount = threads->valueint;
  } else if (!threads) {
    g_Dbs.threadCount = 1;
  } else {
2927
    printf("ERROR: failed to read json, threads not found\n");
2928 2929
    goto PARSE_OVER;
  }  
2930
 
2931 2932 2933 2934 2935 2936
  cJSON* threads2 = cJSON_GetObjectItem(root, "thread_count_create_tbl");
  if (threads2 && threads2->type == cJSON_Number) {
    g_Dbs.threadCountByCreateTbl = threads2->valueint;
  } else if (!threads2) {
    g_Dbs.threadCountByCreateTbl = 1;
  } else {
2937
    printf("ERROR: failed to read json, threads2 not found\n");
2938 2939 2940
    goto PARSE_OVER;
  } 

2941 2942 2943 2944
      cJSON* gInsertInterval = cJSON_GetObjectItem(root, "insert_interval");
      if (gInsertInterval && gInsertInterval->type == cJSON_Number) {
        g_args.insert_interval = gInsertInterval->valueint;
      } else if (!gInsertInterval) {
2945
        g_args.insert_interval = 0;
2946
      } else {
2947
        fprintf(stderr, "ERROR: failed to read json, insert_interval input mistake\n");
2948 2949 2950
        goto PARSE_OVER;
      }
 
2951 2952 2953 2954 2955 2956 2957 2958 2959 2960 2961
      cJSON* maxSqlLen = cJSON_GetObjectItem(root, "max_sql_len");
      if (maxSqlLen && maxSqlLen->type == cJSON_Number) {
        g_args.max_sql_len = maxSqlLen->valueint;
      } else if (!maxSqlLen) {
        g_args.max_sql_len = TSDB_PAYLOAD_SIZE;
      } else {
        fprintf(stderr, "ERROR: failed to read json, max_sql_len input mistake\n");
        goto PARSE_OVER;
      }
 

2962 2963
      cJSON* numRecPerReq = cJSON_GetObjectItem(root, "num_of_records_per_req");
      if (numRecPerReq && numRecPerReq->type == cJSON_Number) {
2964
        g_args.num_of_RPR = numRecPerReq->valueint;
2965
      } else if (!numRecPerReq) {
2966
        g_args.num_of_RPR = 100;
2967
      } else {
2968
        printf("ERROR: failed to read json, num_of_records_per_req not found\n");
2969 2970 2971
        goto PARSE_OVER;
      }
 
2972
  cJSON *answerPrompt = cJSON_GetObjectItem(root, "confirm_parameter_prompt"); // yes, no,
2973 2974 2975
  if (answerPrompt 
          && answerPrompt->type == cJSON_String
          && answerPrompt->valuestring != NULL) {
2976 2977 2978 2979 2980 2981 2982 2983 2984 2985
    if (0 == strncasecmp(answerPrompt->valuestring, "yes", 3)) {
      g_args.answer_yes = false;
    } else if (0 == strncasecmp(answerPrompt->valuestring, "no", 2)) {
      g_args.answer_yes = true;
    } else {
      g_args.answer_yes = false;
    }
  } else if (!answerPrompt) {
    g_args.answer_yes = false;
  } else {
2986
    printf("ERROR: failed to read json, confirm_parameter_prompt not found\n");
2987 2988 2989 2990 2991
    goto PARSE_OVER;
  }  

  cJSON* dbs = cJSON_GetObjectItem(root, "databases");
  if (!dbs || dbs->type != cJSON_Array) {
2992
    printf("ERROR: failed to read json, databases not found\n");
2993 2994 2995 2996 2997
    goto PARSE_OVER;
  }

  int dbSize = cJSON_GetArraySize(dbs);
  if (dbSize > MAX_DB_COUNT) {
2998
    printf("ERROR: failed to read json, databases size overflow, max database is %d\n", MAX_DB_COUNT);
2999 3000 3001 3002 3003 3004 3005 3006 3007 3008 3009
    goto PARSE_OVER;
  }

  g_Dbs.dbCount = dbSize;
  for (int i = 0; i < dbSize; ++i) {
    cJSON* dbinfos = cJSON_GetArrayItem(dbs, i);
    if (dbinfos == NULL) continue;

    // dbinfo 
    cJSON *dbinfo = cJSON_GetObjectItem(dbinfos, "dbinfo");
    if (!dbinfo || dbinfo->type != cJSON_Object) {
3010
      printf("ERROR: failed to read json, dbinfo not found\n");
3011 3012 3013 3014 3015
      goto PARSE_OVER;
    }
    
    cJSON *dbName = cJSON_GetObjectItem(dbinfo, "name");
    if (!dbName || dbName->type != cJSON_String || dbName->valuestring == NULL) {
3016
      printf("ERROR: failed to read json, db name not found\n");
3017 3018
      goto PARSE_OVER;
    }
3019
    tstrncpy(g_Dbs.db[i].dbName, dbName->valuestring, MAX_DB_NAME_SIZE);
3020 3021 3022 3023 3024 3025 3026 3027 3028 3029 3030

    cJSON *drop = cJSON_GetObjectItem(dbinfo, "drop");
    if (drop && drop->type == cJSON_String && drop->valuestring != NULL) {
      if (0 == strncasecmp(drop->valuestring, "yes", 3)) {
        g_Dbs.db[i].drop = 1;
      } else {
        g_Dbs.db[i].drop = 0;
      }        
    } else if (!drop) {
      g_Dbs.db[i].drop = 0;
    } else {
3031
      printf("ERROR: failed to read json, drop not found\n");
3032 3033 3034 3035 3036
      goto PARSE_OVER;
    }

    cJSON *precision = cJSON_GetObjectItem(dbinfo, "precision");
    if (precision && precision->type == cJSON_String && precision->valuestring != NULL) {
3037
      tstrncpy(g_Dbs.db[i].dbCfg.precision, precision->valuestring, MAX_DB_NAME_SIZE);
3038
    } else if (!precision) {
3039
      //tstrncpy(g_Dbs.db[i].dbCfg.precision, "ms", MAX_DB_NAME_SIZE);
3040 3041
      memset(g_Dbs.db[i].dbCfg.precision, 0, MAX_DB_NAME_SIZE);
    } else {
3042
      printf("ERROR: failed to read json, precision not found\n");
3043 3044 3045 3046 3047 3048 3049 3050 3051
      goto PARSE_OVER;
    }

    cJSON* update = cJSON_GetObjectItem(dbinfo, "update");
    if (update && update->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.update = update->valueint;
    } else if (!update) {
      g_Dbs.db[i].dbCfg.update = -1;
    } else {
3052
      printf("ERROR: failed to read json, update not found\n");
3053 3054 3055 3056 3057 3058 3059 3060 3061
      goto PARSE_OVER;
    }

    cJSON* replica = cJSON_GetObjectItem(dbinfo, "replica");
    if (replica && replica->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.replica = replica->valueint;
    } else if (!replica) {
      g_Dbs.db[i].dbCfg.replica = -1;
    } else {
3062
      printf("ERROR: failed to read json, replica not found\n");
3063 3064 3065 3066 3067 3068 3069 3070 3071
      goto PARSE_OVER;
    }

    cJSON* keep = cJSON_GetObjectItem(dbinfo, "keep");
    if (keep && keep->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.keep = keep->valueint;
    } else if (!keep) {
      g_Dbs.db[i].dbCfg.keep = -1;
    } else {
3072
     printf("ERROR: failed to read json, keep not found\n");
3073 3074 3075 3076 3077 3078 3079 3080 3081
     goto PARSE_OVER;
    }
    
    cJSON* days = cJSON_GetObjectItem(dbinfo, "days");
    if (days && days->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.days = days->valueint;
    } else if (!days) {
      g_Dbs.db[i].dbCfg.days = -1;
    } else {
3082
     printf("ERROR: failed to read json, days not found\n");
3083 3084 3085 3086 3087 3088 3089 3090 3091
     goto PARSE_OVER;
    }
    
    cJSON* cache = cJSON_GetObjectItem(dbinfo, "cache");
    if (cache && cache->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.cache = cache->valueint;
    } else if (!cache) {
      g_Dbs.db[i].dbCfg.cache = -1;
    } else {
3092
     printf("ERROR: failed to read json, cache not found\n");
3093 3094 3095 3096 3097 3098 3099 3100 3101
     goto PARSE_OVER;
    }
        
    cJSON* blocks= cJSON_GetObjectItem(dbinfo, "blocks");
    if (blocks && blocks->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.blocks = blocks->valueint;
    } else if (!blocks) {
      g_Dbs.db[i].dbCfg.blocks = -1;
    } else {
3102
     printf("ERROR: failed to read json, block not found\n");
3103 3104 3105 3106 3107 3108 3109 3110 3111 3112 3113 3114 3115 3116 3117 3118 3119 3120 3121
     goto PARSE_OVER;
    }

    //cJSON* maxtablesPerVnode= cJSON_GetObjectItem(dbinfo, "maxtablesPerVnode");
    //if (maxtablesPerVnode && maxtablesPerVnode->type == cJSON_Number) {
    //  g_Dbs.db[i].dbCfg.maxtablesPerVnode = maxtablesPerVnode->valueint;
    //} else if (!maxtablesPerVnode) {
    //  g_Dbs.db[i].dbCfg.maxtablesPerVnode = TSDB_DEFAULT_TABLES;
    //} else {
    // printf("failed to read json, maxtablesPerVnode not found");
    // goto PARSE_OVER;
    //}

    cJSON* minRows= cJSON_GetObjectItem(dbinfo, "minRows");
    if (minRows && minRows->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.minRows = minRows->valueint;
    } else if (!minRows) {
      g_Dbs.db[i].dbCfg.minRows = -1;
    } else {
3122
     printf("ERROR: failed to read json, minRows not found\n");
3123 3124 3125 3126 3127 3128 3129 3130 3131
     goto PARSE_OVER;
    }

    cJSON* maxRows= cJSON_GetObjectItem(dbinfo, "maxRows");
    if (maxRows && maxRows->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.maxRows = maxRows->valueint;
    } else if (!maxRows) {
      g_Dbs.db[i].dbCfg.maxRows = -1;
    } else {
3132
     printf("ERROR: failed to read json, maxRows not found\n");
3133 3134 3135 3136 3137 3138 3139 3140 3141
     goto PARSE_OVER;
    }

    cJSON* comp= cJSON_GetObjectItem(dbinfo, "comp");
    if (comp && comp->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.comp = comp->valueint;
    } else if (!comp) {
      g_Dbs.db[i].dbCfg.comp = -1;
    } else {
3142
     printf("ERROR: failed to read json, comp not found\n");
3143 3144 3145 3146 3147 3148 3149 3150 3151
     goto PARSE_OVER;
    }

    cJSON* walLevel= cJSON_GetObjectItem(dbinfo, "walLevel");
    if (walLevel && walLevel->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.walLevel = walLevel->valueint;
    } else if (!walLevel) {
      g_Dbs.db[i].dbCfg.walLevel = -1;
    } else {
3152
     printf("ERROR: failed to read json, walLevel not found\n");
3153 3154 3155
     goto PARSE_OVER;
    }

3156 3157 3158 3159 3160 3161
    cJSON* cacheLast= cJSON_GetObjectItem(dbinfo, "cachelast");
    if (cacheLast && cacheLast->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.cacheLast = cacheLast->valueint;
    } else if (!cacheLast) {
      g_Dbs.db[i].dbCfg.cacheLast = -1;
    } else {
3162
     printf("ERROR: failed to read json, cacheLast not found\n");
3163 3164 3165
     goto PARSE_OVER;
    }

3166 3167 3168 3169
    cJSON* quorum= cJSON_GetObjectItem(dbinfo, "quorum");
    if (quorum && quorum->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.quorum = quorum->valueint;
    } else if (!quorum) {
3170
      g_Dbs.db[i].dbCfg.quorum = 1;
3171
    } else {
3172
     printf("failed to read json, quorum input mistake");
3173 3174 3175 3176 3177 3178 3179 3180 3181
     goto PARSE_OVER;
    }

    cJSON* fsync= cJSON_GetObjectItem(dbinfo, "fsync");
    if (fsync && fsync->type == cJSON_Number) {
      g_Dbs.db[i].dbCfg.fsync = fsync->valueint;
    } else if (!fsync) {
      g_Dbs.db[i].dbCfg.fsync = -1;
    } else {
3182
     printf("ERROR: failed to read json, fsync not found\n");
3183 3184 3185 3186 3187 3188
     goto PARSE_OVER;
    }    

    // super_talbes 
    cJSON *stables = cJSON_GetObjectItem(dbinfos, "super_tables");
    if (!stables || stables->type != cJSON_Array) {
3189
      printf("ERROR: failed to read json, super_tables not found\n");
3190 3191 3192 3193 3194
      goto PARSE_OVER;
    }    
    
    int stbSize = cJSON_GetArraySize(stables);
    if (stbSize > MAX_SUPER_TABLE_COUNT) {
3195
      printf("ERROR: failed to read json, databases size overflow, max database is %d\n", MAX_SUPER_TABLE_COUNT);
3196 3197 3198 3199 3200 3201 3202 3203 3204 3205 3206
      goto PARSE_OVER;
    }

    g_Dbs.db[i].superTblCount = stbSize;
    for (int j = 0; j < stbSize; ++j) {
      cJSON* stbInfo = cJSON_GetArrayItem(stables, j);
      if (stbInfo == NULL) continue;
    
      // dbinfo 
      cJSON *stbName = cJSON_GetObjectItem(stbInfo, "name");
      if (!stbName || stbName->type != cJSON_String || stbName->valuestring == NULL) {
3207
        printf("ERROR: failed to read json, stb name not found\n");
3208 3209
        goto PARSE_OVER;
      }
3210
      tstrncpy(g_Dbs.db[i].superTbls[j].sTblName, stbName->valuestring, MAX_TB_NAME_SIZE);
3211 3212 3213
    
      cJSON *prefix = cJSON_GetObjectItem(stbInfo, "childtable_prefix");
      if (!prefix || prefix->type != cJSON_String || prefix->valuestring == NULL) {
3214
        printf("ERROR: failed to read json, childtable_prefix not found\n");
3215 3216
        goto PARSE_OVER;
      }
3217
      tstrncpy(g_Dbs.db[i].superTbls[j].childTblPrefix, prefix->valuestring, MAX_DB_NAME_SIZE);
3218 3219

      cJSON *autoCreateTbl = cJSON_GetObjectItem(stbInfo, "auto_create_table"); // yes, no, null
3220 3221 3222
      if (autoCreateTbl
              && autoCreateTbl->type == cJSON_String
              && autoCreateTbl->valuestring != NULL) {
3223 3224 3225 3226 3227 3228 3229 3230 3231 3232
        if (0 == strncasecmp(autoCreateTbl->valuestring, "yes", 3)) {
          g_Dbs.db[i].superTbls[j].autoCreateTable = AUTO_CREATE_SUBTBL;
        } else if (0 == strncasecmp(autoCreateTbl->valuestring, "no", 2)) {
          g_Dbs.db[i].superTbls[j].autoCreateTable = PRE_CREATE_SUBTBL;
        } else {
          g_Dbs.db[i].superTbls[j].autoCreateTable = PRE_CREATE_SUBTBL;
        }
      } else if (!autoCreateTbl) {
        g_Dbs.db[i].superTbls[j].autoCreateTable = PRE_CREATE_SUBTBL;
      } else {
3233
        printf("ERROR: failed to read json, auto_create_table not found\n");
3234 3235 3236 3237 3238 3239 3240
        goto PARSE_OVER;
      }
      
      cJSON* batchCreateTbl = cJSON_GetObjectItem(stbInfo, "batch_create_tbl_num");
      if (batchCreateTbl && batchCreateTbl->type == cJSON_Number) {
        g_Dbs.db[i].superTbls[j].batchCreateTableNum = batchCreateTbl->valueint;
      } else if (!batchCreateTbl) {
3241
        g_Dbs.db[i].superTbls[j].batchCreateTableNum = 1000;
3242
      } else {
3243
        printf("ERROR: failed to read json, batch_create_tbl_num not found\n");
3244 3245 3246 3247
        goto PARSE_OVER;
      }      

      cJSON *childTblExists = cJSON_GetObjectItem(stbInfo, "child_table_exists"); // yes, no
3248 3249 3250
      if (childTblExists
              && childTblExists->type == cJSON_String
              && childTblExists->valuestring != NULL) {
3251 3252 3253 3254 3255 3256 3257 3258 3259 3260
        if (0 == strncasecmp(childTblExists->valuestring, "yes", 3)) {
          g_Dbs.db[i].superTbls[j].childTblExists = TBL_ALREADY_EXISTS;
        } else if (0 == strncasecmp(childTblExists->valuestring, "no", 2)) {
          g_Dbs.db[i].superTbls[j].childTblExists = TBL_NO_EXISTS;
        } else {
          g_Dbs.db[i].superTbls[j].childTblExists = TBL_NO_EXISTS;
        }
      } else if (!childTblExists) {
        g_Dbs.db[i].superTbls[j].childTblExists = TBL_NO_EXISTS;
      } else {
3261
        printf("ERROR: failed to read json, child_table_exists not found\n");
3262 3263 3264 3265 3266
        goto PARSE_OVER;
      }
      
      cJSON* count = cJSON_GetObjectItem(stbInfo, "childtable_count");
      if (!count || count->type != cJSON_Number || 0 >= count->valueint) {
3267
        printf("ERROR: failed to read json, childtable_count not found\n");
3268 3269 3270 3271 3272
        goto PARSE_OVER;
      }
      g_Dbs.db[i].superTbls[j].childTblCount = count->valueint;

      cJSON *dataSource = cJSON_GetObjectItem(stbInfo, "data_source");
3273 3274 3275 3276
      if (dataSource && dataSource->type == cJSON_String
              && dataSource->valuestring != NULL) {
        tstrncpy(g_Dbs.db[i].superTbls[j].dataSource,
                dataSource->valuestring, MAX_DB_NAME_SIZE);
3277
      } else if (!dataSource) {
3278
        tstrncpy(g_Dbs.db[i].superTbls[j].dataSource, "rand", MAX_DB_NAME_SIZE);
3279
      } else {
3280
        printf("ERROR: failed to read json, data_source not found\n");
3281 3282 3283 3284
        goto PARSE_OVER;
      }

      cJSON *insertMode = cJSON_GetObjectItem(stbInfo, "insert_mode"); // taosc , restful
3285 3286 3287 3288
      if (insertMode && insertMode->type == cJSON_String
              && insertMode->valuestring != NULL) {
        tstrncpy(g_Dbs.db[i].superTbls[j].insertMode,
                insertMode->valuestring, MAX_DB_NAME_SIZE);
3289
      } else if (!insertMode) {
3290
        tstrncpy(g_Dbs.db[i].superTbls[j].insertMode, "taosc", MAX_DB_NAME_SIZE);
3291
      } else {
3292
        printf("ERROR: failed to read json, insert_mode not found\n");
3293 3294 3295
        goto PARSE_OVER;
      }

3296 3297 3298 3299 3300 3301 3302 3303 3304 3305 3306 3307 3308 3309 3310 3311 3312 3313 3314 3315 3316 3317
      cJSON* childTbl_limit = cJSON_GetObjectItem(stbInfo, "childtable_limit");
      if (childTbl_limit) {
        if (childTbl_limit->type != cJSON_Number) {
            printf("ERROR: failed to read json, childtable_limit\n");
            goto PARSE_OVER;
        }
        g_Dbs.db[i].superTbls[j].childTblLimit = childTbl_limit->valueint;
      } else {
        g_Dbs.db[i].superTbls[j].childTblLimit = -1;    // select ... limit -1 means all query result
      }

      cJSON* childTbl_offset = cJSON_GetObjectItem(stbInfo, "childtable_offset");
      if (childTbl_offset) {
        if (childTbl_offset->type != cJSON_Number || 0 > childTbl_offset->valueint) {
            printf("ERROR: failed to read json, childtable_offset\n");
            goto PARSE_OVER;
        }
        g_Dbs.db[i].superTbls[j].childTblOffset = childTbl_offset->valueint;
      } else {
        g_Dbs.db[i].superTbls[j].childTblOffset = 0;
      }

3318 3319
      cJSON *ts = cJSON_GetObjectItem(stbInfo, "start_timestamp");
      if (ts && ts->type == cJSON_String && ts->valuestring != NULL) {
3320
        tstrncpy(g_Dbs.db[i].superTbls[j].startTimestamp, ts->valuestring, MAX_DB_NAME_SIZE);
3321
      } else if (!ts) {
3322
        tstrncpy(g_Dbs.db[i].superTbls[j].startTimestamp, "now", MAX_DB_NAME_SIZE);
3323
      } else {
3324
        printf("ERROR: failed to read json, start_timestamp not found\n");
3325 3326
        goto PARSE_OVER;
      }
3327

3328 3329 3330 3331
      cJSON* timestampStep = cJSON_GetObjectItem(stbInfo, "timestamp_step");
      if (timestampStep && timestampStep->type == cJSON_Number) {
        g_Dbs.db[i].superTbls[j].timeStampStep = timestampStep->valueint;
      } else if (!timestampStep) {
3332
        g_Dbs.db[i].superTbls[j].timeStampStep = DEFAULT_TIMESTAMP_STEP;
3333
      } else {
3334
        printf("ERROR: failed to read json, timestamp_step not found\n");
3335 3336 3337 3338 3339 3340 3341 3342 3343 3344 3345 3346
        goto PARSE_OVER;
      }

      cJSON* sampleDataBufSize = cJSON_GetObjectItem(stbInfo, "sample_buf_size");
      if (sampleDataBufSize && sampleDataBufSize->type == cJSON_Number) {
        g_Dbs.db[i].superTbls[j].sampleDataBufSize = sampleDataBufSize->valueint;
        if (g_Dbs.db[i].superTbls[j].sampleDataBufSize < 1024*1024) {
          g_Dbs.db[i].superTbls[j].sampleDataBufSize = 1024*1024 + 1024;
        }
      } else if (!sampleDataBufSize) {
        g_Dbs.db[i].superTbls[j].sampleDataBufSize = 1024*1024 + 1024;
      } else {
3347
        printf("ERROR: failed to read json, sample_buf_size not found\n");
3348 3349
        goto PARSE_OVER;
      }      
3350

3351 3352
      cJSON *sampleFormat = cJSON_GetObjectItem(stbInfo, "sample_format");
      if (sampleFormat && sampleFormat->type == cJSON_String && sampleFormat->valuestring != NULL) {
3353 3354
        tstrncpy(g_Dbs.db[i].superTbls[j].sampleFormat,
                sampleFormat->valuestring, MAX_DB_NAME_SIZE);
3355
      } else if (!sampleFormat) {
3356
        tstrncpy(g_Dbs.db[i].superTbls[j].sampleFormat, "csv", MAX_DB_NAME_SIZE);
3357
      } else {
3358
        printf("ERROR: failed to read json, sample_format not found\n");
3359 3360
        goto PARSE_OVER;
      }      
3361

3362 3363
      cJSON *sampleFile = cJSON_GetObjectItem(stbInfo, "sample_file");
      if (sampleFile && sampleFile->type == cJSON_String && sampleFile->valuestring != NULL) {
3364 3365
        tstrncpy(g_Dbs.db[i].superTbls[j].sampleFile,
                sampleFile->valuestring, MAX_FILE_NAME_LEN);
3366 3367 3368
      } else if (!sampleFile) {
        memset(g_Dbs.db[i].superTbls[j].sampleFile, 0, MAX_FILE_NAME_LEN);
      } else {
3369
        printf("ERROR: failed to read json, sample_file not found\n");
3370 3371
        goto PARSE_OVER;
      }          
3372

3373 3374
      cJSON *tagsFile = cJSON_GetObjectItem(stbInfo, "tags_file");
      if (tagsFile && tagsFile->type == cJSON_String && tagsFile->valuestring != NULL) {
3375 3376
        tstrncpy(g_Dbs.db[i].superTbls[j].tagsFile,
                tagsFile->valuestring, MAX_FILE_NAME_LEN);
3377 3378 3379 3380 3381 3382 3383 3384 3385
        if (0 == g_Dbs.db[i].superTbls[j].tagsFile[0]) {
          g_Dbs.db[i].superTbls[j].tagSource = 0;
        } else {
          g_Dbs.db[i].superTbls[j].tagSource = 1;
        }
      } else if (!tagsFile) {
        memset(g_Dbs.db[i].superTbls[j].tagsFile, 0, MAX_FILE_NAME_LEN);
        g_Dbs.db[i].superTbls[j].tagSource = 0;
      } else {
3386
        printf("ERROR: failed to read json, tags_file not found\n");
3387 3388
        goto PARSE_OVER;
      }
3389

3390 3391 3392 3393 3394 3395 3396 3397 3398 3399 3400 3401
      cJSON* maxSqlLen = cJSON_GetObjectItem(stbInfo, "max_sql_len");
      if (maxSqlLen && maxSqlLen->type == cJSON_Number) {
        int32_t len = maxSqlLen->valueint;
        if (len > TSDB_MAX_ALLOWED_SQL_LEN) {
          len = TSDB_MAX_ALLOWED_SQL_LEN;
        } else if (len < TSDB_MAX_SQL_LEN) {
          len = TSDB_MAX_SQL_LEN;
        }       
        g_Dbs.db[i].superTbls[j].maxSqlLen = len;
      } else if (!maxSqlLen) {
        g_Dbs.db[i].superTbls[j].maxSqlLen = TSDB_MAX_SQL_LEN;
      } else {
3402
        printf("ERROR: failed to read json, maxSqlLen not found\n");
3403 3404 3405
        goto PARSE_OVER;
      }      

3406 3407 3408 3409 3410
      cJSON *multiThreadWriteOneTbl =
          cJSON_GetObjectItem(stbInfo, "multi_thread_write_one_tbl"); // no , yes
      if (multiThreadWriteOneTbl
              && multiThreadWriteOneTbl->type == cJSON_String
              && multiThreadWriteOneTbl->valuestring != NULL) {
3411 3412 3413 3414 3415 3416 3417 3418
        if (0 == strncasecmp(multiThreadWriteOneTbl->valuestring, "yes", 3)) {
          g_Dbs.db[i].superTbls[j].multiThreadWriteOneTbl = 1;
        } else {
          g_Dbs.db[i].superTbls[j].multiThreadWriteOneTbl = 0;
        }        
      } else if (!multiThreadWriteOneTbl) {
        g_Dbs.db[i].superTbls[j].multiThreadWriteOneTbl = 0;
      } else {
3419
        printf("ERROR: failed to read json, multiThreadWriteOneTbl not found\n");
3420 3421 3422 3423 3424 3425 3426 3427 3428
        goto PARSE_OVER;
      }

      cJSON* numberOfTblInOneSql = cJSON_GetObjectItem(stbInfo, "number_of_tbl_in_one_sql");
      if (numberOfTblInOneSql && numberOfTblInOneSql->type == cJSON_Number) {
        g_Dbs.db[i].superTbls[j].numberOfTblInOneSql = numberOfTblInOneSql->valueint;
      } else if (!numberOfTblInOneSql) {
        g_Dbs.db[i].superTbls[j].numberOfTblInOneSql = 0;
      } else {
3429
        printf("ERROR: failed to read json, numberOfTblInOneSql not found\n");
3430 3431 3432 3433 3434 3435 3436 3437 3438
        goto PARSE_OVER;
      }      

      cJSON* rowsPerTbl = cJSON_GetObjectItem(stbInfo, "rows_per_tbl");
      if (rowsPerTbl && rowsPerTbl->type == cJSON_Number) {
        g_Dbs.db[i].superTbls[j].rowsPerTbl = rowsPerTbl->valueint;
      } else if (!rowsPerTbl) {
        g_Dbs.db[i].superTbls[j].rowsPerTbl = 1;
      } else {
3439
        printf("ERROR: failed to read json, rowsPerTbl not found\n");
3440 3441 3442 3443 3444 3445 3446 3447 3448
        goto PARSE_OVER;
      }      

      cJSON* disorderRatio = cJSON_GetObjectItem(stbInfo, "disorder_ratio");
      if (disorderRatio && disorderRatio->type == cJSON_Number) {
        g_Dbs.db[i].superTbls[j].disorderRatio = disorderRatio->valueint;
      } else if (!disorderRatio) {
        g_Dbs.db[i].superTbls[j].disorderRatio = 0;
      } else {
3449
        printf("ERROR: failed to read json, disorderRatio not found\n");
3450 3451 3452 3453 3454 3455 3456 3457 3458
        goto PARSE_OVER;
      }      

      cJSON* disorderRange = cJSON_GetObjectItem(stbInfo, "disorder_range");
      if (disorderRange && disorderRange->type == cJSON_Number) {
        g_Dbs.db[i].superTbls[j].disorderRange = disorderRange->valueint;
      } else if (!disorderRange) {
        g_Dbs.db[i].superTbls[j].disorderRange = 1000;
      } else {
3459
        printf("ERROR: failed to read json, disorderRange not found\n");
3460 3461
        goto PARSE_OVER;
      }
3462

3463 3464 3465 3466 3467 3468
      cJSON* insertRows = cJSON_GetObjectItem(stbInfo, "insert_rows");
      if (insertRows && insertRows->type == cJSON_Number) {
        g_Dbs.db[i].superTbls[j].insertRows = insertRows->valueint;
      } else if (!insertRows) {
        g_Dbs.db[i].superTbls[j].insertRows = 0x7FFFFFFFFFFFFFFF;
      } else {
3469
        fprintf(stderr, "failed to read json, insert_rows input mistake");
3470 3471 3472 3473 3474 3475 3476 3477 3478 3479 3480
        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;
      } else if (!insertInterval) {
        debugPrint("%s() LN%d: stable insert interval be overrided by global %d.\n",
                __func__, __LINE__, g_args.insert_interval);
        g_Dbs.db[i].superTbls[j].insertInterval = g_args.insert_interval;
      } else {
3481
        fprintf(stderr, "failed to read json, insert_interval input mistake");
3482 3483 3484
        goto PARSE_OVER;
      }

3485 3486
      if (NO_CREATE_SUBTBL == g_Dbs.db[i].superTbls[j].autoCreateTable
              || (TBL_ALREADY_EXISTS == g_Dbs.db[i].superTbls[j].childTblExists)) {
3487 3488 3489 3490 3491 3492 3493 3494 3495 3496 3497 3498 3499 3500 3501 3502 3503 3504 3505 3506 3507 3508 3509 3510
        continue;
      }

      int retVal = getColumnAndTagTypeFromInsertJsonFile(stbInfo, &g_Dbs.db[i].superTbls[j]);
      if (false == retVal) {
        goto PARSE_OVER;
      }      
    }    
  }

  ret = true;

PARSE_OVER:
  //free(content);
  //cJSON_Delete(root);
  //fclose(fp);
  return ret;
}

static bool getMetaFromQueryJsonFile(cJSON* root) {
  bool  ret = false;

  cJSON* cfgdir = cJSON_GetObjectItem(root, "cfgdir");
  if (cfgdir && cfgdir->type == cJSON_String && cfgdir->valuestring != NULL) {
3511
    tstrncpy(g_queryInfo.cfgDir, cfgdir->valuestring, MAX_FILE_NAME_LEN);
3512 3513 3514 3515
  }

  cJSON* host = cJSON_GetObjectItem(root, "host");
  if (host && host->type == cJSON_String && host->valuestring != NULL) {
3516
    tstrncpy(g_queryInfo.host, host->valuestring, MAX_DB_NAME_SIZE);
3517
  } else if (!host) {
3518
    tstrncpy(g_queryInfo.host, "127.0.0.1", MAX_DB_NAME_SIZE);
3519
  } else {
3520
    printf("ERROR: failed to read json, host not found\n");
3521 3522 3523 3524 3525 3526 3527 3528 3529 3530 3531 3532
    goto PARSE_OVER;
  }

  cJSON* port = cJSON_GetObjectItem(root, "port");
  if (port && port->type == cJSON_Number) {
    g_queryInfo.port = port->valueint;
  } else if (!port) {
    g_queryInfo.port = 6030;
  }

  cJSON* user = cJSON_GetObjectItem(root, "user");
  if (user && user->type == cJSON_String && user->valuestring != NULL) {
3533
    tstrncpy(g_queryInfo.user, user->valuestring, MAX_DB_NAME_SIZE);   
3534
  } else if (!user) {
3535
    tstrncpy(g_queryInfo.user, "root", MAX_DB_NAME_SIZE); ;
3536 3537 3538 3539
  }

  cJSON* password = cJSON_GetObjectItem(root, "password");
  if (password && password->type == cJSON_String && password->valuestring != NULL) {
3540
    tstrncpy(g_queryInfo.password, password->valuestring, MAX_DB_NAME_SIZE);
3541
  } else if (!password) {
3542
    tstrncpy(g_queryInfo.password, "taosdata", MAX_DB_NAME_SIZE);;
3543
  }
H
hzcheng 已提交
3544

3545
  cJSON *answerPrompt = cJSON_GetObjectItem(root, "confirm_parameter_prompt"); // yes, no,
3546 3547
  if (answerPrompt && answerPrompt->type == cJSON_String
          && answerPrompt->valuestring != NULL) {
3548 3549 3550 3551
    if (0 == strncasecmp(answerPrompt->valuestring, "yes", 3)) {
      g_args.answer_yes = false;
    } else if (0 == strncasecmp(answerPrompt->valuestring, "no", 2)) {
      g_args.answer_yes = true;
H
hzcheng 已提交
3552
    } else {
3553
      g_args.answer_yes = false;
H
hzcheng 已提交
3554
    }
3555 3556 3557
  } else if (!answerPrompt) {
    g_args.answer_yes = false;
  } else {
3558
    printf("ERROR: failed to read json, confirm_parameter_prompt not found\n");
3559 3560 3561 3562 3563
    goto PARSE_OVER;
  }  

  cJSON* dbs = cJSON_GetObjectItem(root, "databases");
  if (dbs && dbs->type == cJSON_String && dbs->valuestring != NULL) {
3564
    tstrncpy(g_queryInfo.dbName, dbs->valuestring, MAX_DB_NAME_SIZE);
3565
  } else if (!dbs) {
3566
    printf("ERROR: failed to read json, databases not found\n");
3567
    goto PARSE_OVER;
H
hzcheng 已提交
3568
  }
3569 3570 3571

  cJSON* queryMode = cJSON_GetObjectItem(root, "query_mode");
  if (queryMode && queryMode->type == cJSON_String && queryMode->valuestring != NULL) {
3572
    tstrncpy(g_queryInfo.queryMode, queryMode->valuestring, MAX_TB_NAME_SIZE);
3573
  } else if (!queryMode) {
3574
    tstrncpy(g_queryInfo.queryMode, "taosc", MAX_TB_NAME_SIZE);
3575
  } else {
3576
    printf("ERROR: failed to read json, query_mode not found\n");
3577
    goto PARSE_OVER;
H
hzcheng 已提交
3578
  }
3579 3580 3581 3582 3583 3584 3585
  
  // super_table_query 
  cJSON *superQuery = cJSON_GetObjectItem(root, "specified_table_query");
  if (!superQuery) {
    g_queryInfo.superQueryInfo.concurrent = 0;
    g_queryInfo.superQueryInfo.sqlCount = 0;
  } else if (superQuery->type != cJSON_Object) {
3586
    printf("ERROR: failed to read json, super_table_query not found\n");
3587 3588 3589 3590 3591 3592 3593 3594 3595 3596 3597 3598 3599 3600 3601 3602 3603 3604 3605 3606 3607 3608 3609
    goto PARSE_OVER;
  } else {  
    cJSON* rate = cJSON_GetObjectItem(superQuery, "query_interval");
    if (rate && rate->type == cJSON_Number) {
      g_queryInfo.superQueryInfo.rate = rate->valueint;
    } else if (!rate) {
      g_queryInfo.superQueryInfo.rate = 0;
    }
  
    cJSON* concurrent = cJSON_GetObjectItem(superQuery, "concurrent");
    if (concurrent && concurrent->type == cJSON_Number) {
      g_queryInfo.superQueryInfo.concurrent = concurrent->valueint;
    } else if (!concurrent) {
      g_queryInfo.superQueryInfo.concurrent = 1;
    }
  
    cJSON* mode = cJSON_GetObjectItem(superQuery, "mode");
    if (mode && mode->type == cJSON_String && mode->valuestring != NULL) {
      if (0 == strcmp("sync", mode->valuestring)) {      
        g_queryInfo.superQueryInfo.subscribeMode = 0;
      } else if (0 == strcmp("async", mode->valuestring)) {      
        g_queryInfo.superQueryInfo.subscribeMode = 1;
      } else {
3610
        printf("ERROR: failed to read json, subscribe mod error\n");
3611 3612 3613 3614 3615 3616 3617 3618 3619 3620 3621 3622 3623 3624 3625 3626 3627 3628 3629 3630 3631 3632
        goto PARSE_OVER;
      }
    } else {
      g_queryInfo.superQueryInfo.subscribeMode = 0;
    }
    
    cJSON* interval = cJSON_GetObjectItem(superQuery, "interval");
    if (interval && interval->type == cJSON_Number) {
      g_queryInfo.superQueryInfo.subscribeInterval = interval->valueint;
    } else if (!interval) {    
      //printf("failed to read json, subscribe interval no found\n");
      //goto PARSE_OVER;
      g_queryInfo.superQueryInfo.subscribeInterval = 10000;
    }
  
    cJSON* restart = cJSON_GetObjectItem(superQuery, "restart");
    if (restart && restart->type == cJSON_String && restart->valuestring != NULL) {
      if (0 == strcmp("yes", restart->valuestring)) {      
        g_queryInfo.superQueryInfo.subscribeRestart = 1;
      } else if (0 == strcmp("no", restart->valuestring)) {      
        g_queryInfo.superQueryInfo.subscribeRestart = 0;
      } else {
3633
        printf("ERROR: failed to read json, subscribe restart error\n");
3634 3635 3636 3637 3638 3639 3640
        goto PARSE_OVER;
      }
    } else {
      g_queryInfo.superQueryInfo.subscribeRestart = 1;
    }
  
    cJSON* keepProgress = cJSON_GetObjectItem(superQuery, "keepProgress");
3641 3642 3643
    if (keepProgress
            && keepProgress->type == cJSON_String
            && keepProgress->valuestring != NULL) {
3644 3645 3646 3647 3648
      if (0 == strcmp("yes", keepProgress->valuestring)) {      
        g_queryInfo.superQueryInfo.subscribeKeepProgress = 1;
      } else if (0 == strcmp("no", keepProgress->valuestring)) {      
        g_queryInfo.superQueryInfo.subscribeKeepProgress = 0;
      } else {
3649
        printf("ERROR: failed to read json, subscribe keepProgress error\n");
3650 3651 3652 3653 3654 3655 3656 3657 3658 3659 3660
        goto PARSE_OVER;
      }
    } else {
      g_queryInfo.superQueryInfo.subscribeKeepProgress = 0;
    }

    // sqls   
    cJSON* superSqls = cJSON_GetObjectItem(superQuery, "sqls");
    if (!superSqls) {
      g_queryInfo.superQueryInfo.sqlCount = 0;
    } else if (superSqls->type != cJSON_Array) {
3661
      printf("ERROR: failed to read json, super sqls not found\n");
3662 3663 3664 3665
      goto PARSE_OVER;
    } else {  
      int superSqlSize = cJSON_GetArraySize(superSqls);
      if (superSqlSize > MAX_QUERY_SQL_COUNT) {
3666
        printf("ERROR: failed to read json, query sql size overflow, max is %d\n", MAX_QUERY_SQL_COUNT);
3667 3668
        goto PARSE_OVER;
      }
3669

3670 3671 3672 3673 3674 3675 3676
      g_queryInfo.superQueryInfo.sqlCount = superSqlSize;
      for (int j = 0; j < superSqlSize; ++j) {
        cJSON* sql = cJSON_GetArrayItem(superSqls, j);
        if (sql == NULL) continue;
      
        cJSON *sqlStr = cJSON_GetObjectItem(sql, "sql");
        if (!sqlStr || sqlStr->type != cJSON_String || sqlStr->valuestring == NULL) {
3677
          printf("ERROR: failed to read json, sql not found\n");
3678 3679
          goto PARSE_OVER;
        }
3680
        tstrncpy(g_queryInfo.superQueryInfo.sql[j], sqlStr->valuestring, MAX_QUERY_SQL_LENGTH);
H
hzcheng 已提交
3681

3682 3683
        cJSON *result = cJSON_GetObjectItem(sql, "result");
        if (NULL != result && result->type == cJSON_String && result->valuestring != NULL) {
3684
          tstrncpy(g_queryInfo.superQueryInfo.result[j], result->valuestring, MAX_FILE_NAME_LEN);
3685 3686 3687
        } else if (NULL == result) {
          memset(g_queryInfo.superQueryInfo.result[j], 0, MAX_FILE_NAME_LEN);
        } else {
3688
          printf("ERROR: failed to read json, super query result file not found\n");
3689 3690 3691 3692 3693
          goto PARSE_OVER;
        } 
      }
    }
  }
3694

3695 3696 3697 3698 3699 3700
  // sub_table_query 
  cJSON *subQuery = cJSON_GetObjectItem(root, "super_table_query");
  if (!subQuery) {
    g_queryInfo.subQueryInfo.threadCnt = 0;
    g_queryInfo.subQueryInfo.sqlCount = 0;
  } else if (subQuery->type != cJSON_Object) {
3701
    printf("ERROR: failed to read json, sub_table_query not found\n");
3702 3703
    ret = true;
    goto PARSE_OVER;
H
hzcheng 已提交
3704
  } else {
3705 3706 3707 3708 3709 3710
    cJSON* subrate = cJSON_GetObjectItem(subQuery, "query_interval");
    if (subrate && subrate->type == cJSON_Number) {
      g_queryInfo.subQueryInfo.rate = subrate->valueint;
    } else if (!subrate) {
      g_queryInfo.subQueryInfo.rate = 0;
    }
3711

3712 3713 3714 3715 3716 3717
    cJSON* threads = cJSON_GetObjectItem(subQuery, "threads");
    if (threads && threads->type == cJSON_Number) {
      g_queryInfo.subQueryInfo.threadCnt = threads->valueint;
    } else if (!threads) {
      g_queryInfo.subQueryInfo.threadCnt = 1;
    }
3718

3719 3720 3721 3722 3723 3724
    //cJSON* subTblCnt = cJSON_GetObjectItem(subQuery, "childtable_count");
    //if (subTblCnt && subTblCnt->type == cJSON_Number) {
    //  g_queryInfo.subQueryInfo.childTblCount = subTblCnt->valueint;
    //} else if (!subTblCnt) {
    //  g_queryInfo.subQueryInfo.childTblCount = 0;
    //}
3725

3726 3727
    cJSON* stblname = cJSON_GetObjectItem(subQuery, "stblname");
    if (stblname && stblname->type == cJSON_String && stblname->valuestring != NULL) {
3728
      tstrncpy(g_queryInfo.subQueryInfo.sTblName, stblname->valuestring, MAX_TB_NAME_SIZE);
3729
    } else {
3730
      printf("ERROR: failed to read json, super table name not found\n");
3731 3732
      goto PARSE_OVER;
    }
3733

3734 3735 3736 3737 3738 3739 3740
    cJSON* submode = cJSON_GetObjectItem(subQuery, "mode");
    if (submode && submode->type == cJSON_String && submode->valuestring != NULL) {
      if (0 == strcmp("sync", submode->valuestring)) {      
        g_queryInfo.subQueryInfo.subscribeMode = 0;
      } else if (0 == strcmp("async", submode->valuestring)) {      
        g_queryInfo.subQueryInfo.subscribeMode = 1;
      } else {
3741
        printf("ERROR: failed to read json, subscribe mod error\n");
3742 3743 3744 3745 3746
        goto PARSE_OVER;
      }
    } else {
      g_queryInfo.subQueryInfo.subscribeMode = 0;
    }
3747

3748 3749 3750 3751 3752 3753 3754 3755 3756 3757 3758 3759 3760 3761 3762 3763
    cJSON* subinterval = cJSON_GetObjectItem(subQuery, "interval");
    if (subinterval && subinterval->type == cJSON_Number) {
      g_queryInfo.subQueryInfo.subscribeInterval = subinterval->valueint;
    } else if (!subinterval) {    
      //printf("failed to read json, subscribe interval no found\n");
      //goto PARSE_OVER;
      g_queryInfo.subQueryInfo.subscribeInterval = 10000;
    }
  
    cJSON* subrestart = cJSON_GetObjectItem(subQuery, "restart");
    if (subrestart && subrestart->type == cJSON_String && subrestart->valuestring != NULL) {
      if (0 == strcmp("yes", subrestart->valuestring)) {      
        g_queryInfo.subQueryInfo.subscribeRestart = 1;
      } else if (0 == strcmp("no", subrestart->valuestring)) {      
        g_queryInfo.subQueryInfo.subscribeRestart = 0;
      } else {
3764
        printf("ERROR: failed to read json, subscribe restart error\n");
3765 3766 3767 3768 3769 3770 3771 3772 3773 3774 3775 3776 3777
        goto PARSE_OVER;
      }
    } else {
      g_queryInfo.subQueryInfo.subscribeRestart = 1;
    }
  
    cJSON* subkeepProgress = cJSON_GetObjectItem(subQuery, "keepProgress");
    if (subkeepProgress && subkeepProgress->type == cJSON_String && subkeepProgress->valuestring != NULL) {
      if (0 == strcmp("yes", subkeepProgress->valuestring)) {      
        g_queryInfo.subQueryInfo.subscribeKeepProgress = 1;
      } else if (0 == strcmp("no", subkeepProgress->valuestring)) {      
        g_queryInfo.subQueryInfo.subscribeKeepProgress = 0;
      } else {
3778
        printf("ERROR: failed to read json, subscribe keepProgress error\n");
3779 3780 3781 3782 3783 3784 3785 3786 3787 3788 3789
        goto PARSE_OVER;
      }
    } else {
      g_queryInfo.subQueryInfo.subscribeKeepProgress = 0;
    }  

    // sqls     
    cJSON* subsqls = cJSON_GetObjectItem(subQuery, "sqls");
    if (!subsqls) {
      g_queryInfo.subQueryInfo.sqlCount = 0;
    } else if (subsqls->type != cJSON_Array) {
3790
      printf("ERROR: failed to read json, super sqls not found\n");
3791 3792 3793 3794
      goto PARSE_OVER;
    } else {  
      int superSqlSize = cJSON_GetArraySize(subsqls);
      if (superSqlSize > MAX_QUERY_SQL_COUNT) {
3795
        printf("ERROR: failed to read json, query sql size overflow, max is %d\n", MAX_QUERY_SQL_COUNT);
3796 3797 3798 3799 3800 3801 3802 3803 3804 3805
        goto PARSE_OVER;
      }
    
      g_queryInfo.subQueryInfo.sqlCount = superSqlSize;
      for (int j = 0; j < superSqlSize; ++j) {      
        cJSON* sql = cJSON_GetArrayItem(subsqls, j);
        if (sql == NULL) continue;
        
        cJSON *sqlStr = cJSON_GetObjectItem(sql, "sql");
        if (!sqlStr || sqlStr->type != cJSON_String || sqlStr->valuestring == NULL) {
3806
          printf("ERROR: failed to read json, sql not found\n");
3807 3808
          goto PARSE_OVER;
        }
3809
        tstrncpy(g_queryInfo.subQueryInfo.sql[j], sqlStr->valuestring, MAX_QUERY_SQL_LENGTH);
3810 3811 3812

        cJSON *result = cJSON_GetObjectItem(sql, "result");
        if (result != NULL && result->type == cJSON_String && result->valuestring != NULL){
3813
          tstrncpy(g_queryInfo.subQueryInfo.result[j], result->valuestring, MAX_FILE_NAME_LEN);
3814 3815 3816
        } else if (NULL == result) {
          memset(g_queryInfo.subQueryInfo.result[j], 0, MAX_FILE_NAME_LEN);
        }  else {
3817
          printf("ERROR: failed to read json, sub query result file not found\n");
3818 3819 3820 3821
          goto PARSE_OVER;
        } 
      }
    }
H
hzcheng 已提交
3822 3823
  }

3824
  ret = true;
H
hzcheng 已提交
3825

3826 3827 3828 3829 3830 3831
PARSE_OVER:
  //free(content);
  //cJSON_Delete(root);
  //fclose(fp);
  return ret;
}
H
hzcheng 已提交
3832

3833
static bool getInfoFromJsonFile(char* file) {
3834
    debugPrint("%s %d %s\n", __func__, __LINE__, file);
3835

3836 3837 3838 3839 3840
  FILE *fp = fopen(file, "r");
  if (!fp) {
    printf("failed to read %s, reason:%s\n", file, strerror(errno));
    return false;
  }
H
Hui Li 已提交
3841

3842 3843 3844 3845 3846 3847 3848 3849 3850
  bool  ret = false;
  int   maxLen = 64000;
  char *content = calloc(1, maxLen + 1);
  int   len = fread(content, 1, maxLen, fp);
  if (len <= 0) {
    free(content);
    fclose(fp);
    printf("failed to read %s, content is null", file);
    return false;
H
hzcheng 已提交
3851
  }
H
Hui Li 已提交
3852

3853 3854 3855
  content[len] = 0;
  cJSON* root = cJSON_Parse(content);
  if (root == NULL) {
3856
    printf("ERROR: failed to cjson parse %s, invalid json format\n", file);
3857 3858
    goto PARSE_OVER;
  }
H
Hui Li 已提交
3859

3860 3861 3862
  cJSON* filetype = cJSON_GetObjectItem(root, "filetype");
  if (filetype && filetype->type == cJSON_String && filetype->valuestring != NULL) {
    if (0 == strcasecmp("insert", filetype->valuestring)) {
3863
      g_args.test_mode = INSERT_MODE;
3864
    } else if (0 == strcasecmp("query", filetype->valuestring)) {
3865
      g_args.test_mode = QUERY_MODE;
3866
    } else if (0 == strcasecmp("subscribe", filetype->valuestring)) {
3867
      g_args.test_mode = SUBSCRIBE_MODE;
3868
    } else {
3869
      printf("ERROR: failed to read json, filetype not support\n");
3870 3871 3872
      goto PARSE_OVER;
    }
  } else if (!filetype) {
3873
    g_args.test_mode = INSERT_MODE;
3874
  } else {
3875
    printf("ERROR: failed to read json, filetype not found\n");
3876 3877
    goto PARSE_OVER;
  }
H
hzcheng 已提交
3878

3879
  if (INSERT_MODE == g_args.test_mode) {
3880
    ret = getMetaFromInsertJsonFile(root);
3881
  } else if (QUERY_MODE == g_args.test_mode) {
3882
    ret = getMetaFromQueryJsonFile(root);
3883
  } else if (SUBSCRIBE_MODE == g_args.test_mode) {
3884 3885
    ret = getMetaFromQueryJsonFile(root);
  } else {
3886
    printf("ERROR: input json file type error! please input correct file type: insert or query or subscribe\n");
3887
    goto PARSE_OVER;
3888
  } 
3889 3890 3891 3892

PARSE_OVER:
  free(content);
  cJSON_Delete(root);
H
hzcheng 已提交
3893
  fclose(fp);
3894 3895
  return ret;
}
H
hzcheng 已提交
3896

3897
void prepareSampleData() {
3898 3899 3900 3901 3902
  for (int i = 0; i < g_Dbs.dbCount; i++) {    
    for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) {
      //if (0 == strncasecmp(g_Dbs.db[i].superTbls[j].dataSource, "sample", 6)) {
      //  readSampleFromFileToMem(&g_Dbs.db[i].superTbls[j]);
      //}
3903

3904 3905
      if (g_Dbs.db[i].superTbls[j].tagsFile[0] != 0) {
        (void)readTagFromCsvFileToMem(&g_Dbs.db[i].superTbls[j]);
S
Shuaiqiang Chang 已提交
3906
      }
3907 3908 3909 3910 3911 3912 3913 3914
    }
  }
}

void postFreeResource() {
  tmfclose(g_fpOfInsertResult);
  for (int i = 0; i < g_Dbs.dbCount; i++) {    
    for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) {
3915 3916 3917
      if (0 != g_Dbs.db[i].superTbls[j].colsOfCreateChildTable) {
        free(g_Dbs.db[i].superTbls[j].colsOfCreateChildTable);
        g_Dbs.db[i].superTbls[j].colsOfCreateChildTable = NULL;
S
Shuaiqiang Chang 已提交
3918
      }
3919 3920 3921 3922 3923 3924 3925 3926 3927 3928 3929 3930
      if (0 != g_Dbs.db[i].superTbls[j].sampleDataBuf) {
        free(g_Dbs.db[i].superTbls[j].sampleDataBuf);
        g_Dbs.db[i].superTbls[j].sampleDataBuf = NULL;
      }
      if (0 != g_Dbs.db[i].superTbls[j].tagDataBuf) {
        free(g_Dbs.db[i].superTbls[j].tagDataBuf);
        g_Dbs.db[i].superTbls[j].tagDataBuf = NULL;
      }
      if (0 != g_Dbs.db[i].superTbls[j].childTblName) {
        free(g_Dbs.db[i].superTbls[j].childTblName);
        g_Dbs.db[i].superTbls[j].childTblName = NULL;
      }
S
Shuaiqiang Chang 已提交
3931
    }
3932 3933
  }
}
S
Shuaiqiang Chang 已提交
3934

3935 3936
static int getRowDataFromSample(char*  dataBuf, int maxLen, int64_t timestamp,
      SSuperTable* superTblInfo, int* sampleUsePos) {
3937
  if ((*sampleUsePos) == MAX_SAMPLES_ONCE_FROM_FILE) {
3938
    int ret = readSampleFromCsvFileToMem(superTblInfo);
3939
    if (0 != ret) {
3940
      tmfree(superTblInfo->sampleDataBuf);
3941 3942 3943 3944
      return -1;
    }
    *sampleUsePos = 0;
  }
S
Shuaiqiang Chang 已提交
3945

3946
  int    dataLen = 0;
3947 3948 3949 3950
  dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen,
          "(%" PRId64 ", ", timestamp);
  dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen,
          "%s", superTblInfo->sampleDataBuf + superTblInfo->lenOfOneRow * (*sampleUsePos));
3951
  dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, ")");
S
Shuaiqiang Chang 已提交
3952

3953
  (*sampleUsePos)++;
3954
 
3955 3956
  return dataLen;
}
S
Shuaiqiang Chang 已提交
3957

3958
static int generateRowData(char*  dataBuf, int maxLen, int64_t timestamp, SSuperTable* stbInfo) {
3959 3960 3961 3962 3963 3964 3965 3966 3967 3968 3969 3970 3971 3972 3973 3974 3975 3976 3977 3978 3979 3980 3981 3982 3983 3984 3985 3986 3987 3988 3989 3990 3991 3992 3993 3994 3995
  int    dataLen = 0;
  dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "(%" PRId64 ", ", timestamp);
  for (int i = 0; i < stbInfo->columnCount; i++) {    
    if ((0 == strncasecmp(stbInfo->columns[i].dataType, "binary", 6)) || (0 == strncasecmp(stbInfo->columns[i].dataType, "nchar", 5))) {
      if (stbInfo->columns[i].dataLen > TSDB_MAX_BINARY_LEN) {
        printf("binary or nchar length overflow, max size:%u\n", (uint32_t)TSDB_MAX_BINARY_LEN);
        return (-1);
      }
    
      char* buf = (char*)calloc(stbInfo->columns[i].dataLen+1, 1);
      if (NULL == buf) {
        printf("calloc failed! size:%d\n", stbInfo->columns[i].dataLen);
        return (-1);
      }
      rand_string(buf, stbInfo->columns[i].dataLen);
      dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "\'%s\', ", buf);
      tmfree(buf);
    } else if (0 == strncasecmp(stbInfo->columns[i].dataType, "int", 3)) {
      dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "%d, ", rand_int());
    } else if (0 == strncasecmp(stbInfo->columns[i].dataType, "bigint", 6)) {
      dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "%"PRId64", ", rand_bigint());
    }  else if (0 == strncasecmp(stbInfo->columns[i].dataType, "float", 5)) {
      dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "%f, ", rand_float());
    }  else if (0 == strncasecmp(stbInfo->columns[i].dataType, "double", 6)) {
      dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "%f, ", rand_double());
    }  else if (0 == strncasecmp(stbInfo->columns[i].dataType, "smallint", 8)) {
      dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "%d, ", rand_smallint());
    }  else if (0 == strncasecmp(stbInfo->columns[i].dataType, "tinyint", 7)) {
      dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "%d, ", rand_tinyint());
    }  else if (0 == strncasecmp(stbInfo->columns[i].dataType, "bool", 4)) {
      dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "%d, ", rand_bool());
    }  else if (0 == strncasecmp(stbInfo->columns[i].dataType, "timestamp", 9)) {
      dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, "%"PRId64", ", rand_bigint());
    }  else {
      printf("No support data type: %s\n", stbInfo->columns[i].dataType);
      return (-1);
    }
S
Shuaiqiang Chang 已提交
3996
  }
3997

3998 3999
  dataLen -= 2;
  dataLen += snprintf(dataBuf + dataLen, maxLen - dataLen, ")");
4000

4001 4002
  return dataLen;
}
S
Shuaiqiang Chang 已提交
4003

4004
static void syncWriteForNumberOfTblInOneSql(
4005
        threadInfo *winfo, char* sampleDataBuf) {
4006 4007 4008 4009 4010 4011 4012 4013 4014 4015 4016 4017 4018 4019 4020 4021 4022 4023 4024
  SSuperTable* superTblInfo = winfo->superTblInfo;

  int   samplePos = 0;

  //printf("========threadID[%d], table rang: %d - %d \n", winfo->threadID, winfo->start_table_id, winfo->end_table_id);
  int64_t    lastPrintTime = taosGetTimestampMs();

  char* buffer = calloc(superTblInfo->maxSqlLen+1, 1);
  if (NULL == buffer) {
    printf("========calloc size[ %d ] fail!\n", superTblInfo->maxSqlLen);
    return;
  }

  int32_t  numberOfTblInOneSql = superTblInfo->numberOfTblInOneSql;
  int32_t  tbls = winfo->end_table_id - winfo->start_table_id + 1;
  if (numberOfTblInOneSql > tbls) {
    numberOfTblInOneSql = tbls;
  }

4025
  uint64_t time_counter = winfo->start_time;
4026 4027
  int sampleUsePos;

4028
  int insert_interval = superTblInfo?superTblInfo->insertInterval:g_args.insert_interval;
4029
  int64_t st = 0;
4030
  int64_t et = 0xffffffff;
4031 4032 4033

  int64_t insertRows = (superTblInfo)?superTblInfo->insertRows:g_args.num_of_DPT;
  for (int i = 0; i < insertRows;) {
4034
    int32_t  tbl_id = 0;
4035
    for (int tableSeq = winfo->start_table_id; tableSeq <= winfo->end_table_id; ) {
4036
      int64_t start_time = 0;
4037 4038
      int inserted = i;

4039
      for (int k = 0; k < g_args.num_of_RPR;) {
4040 4041 4042 4043
        int len = 0;
        memset(buffer, 0, superTblInfo->maxSqlLen);
        char *pstr = buffer;

4044
        int32_t  end_tbl_id = tableSeq + numberOfTblInOneSql;
4045 4046 4047
        if (end_tbl_id > winfo->end_table_id) {
          end_tbl_id = winfo->end_table_id+1;
        }
4048

4049
        for (tbl_id = tableSeq ; tbl_id < end_tbl_id; tbl_id++) {  
4050 4051 4052 4053 4054 4055
          sampleUsePos = samplePos;
          if (AUTO_CREATE_SUBTBL == superTblInfo->autoCreateTable) {
            char* tagsValBuf = NULL;
            if (0 == superTblInfo->tagSource) {
              tagsValBuf = generateTagVaulesForStb(superTblInfo);
            } else {
4056 4057
              tagsValBuf = getTagValueFromTagSample(
                      superTblInfo, tbl_id % superTblInfo->tagSampleCount);
4058 4059 4060 4061 4062 4063
            }
            if (NULL == tagsValBuf) {
              goto free_and_statistics;
            }

            if (0 == len) {
4064 4065
              len += snprintf(pstr + len, 
                      superTblInfo->maxSqlLen - len, 
4066
                      "insert into %s.%s%d using %s.%s tags %s values ",
4067 4068 4069 4070 4071 4072
                      winfo->db_name, 
                      superTblInfo->childTblPrefix, 
                      tbl_id, 
                      winfo->db_name, 
                      superTblInfo->sTblName, 
                      tagsValBuf);
4073
            } else {
4074 4075
              len += snprintf(pstr + len, 
                      superTblInfo->maxSqlLen - len, 
4076
                      " %s.%s%d using %s.%s tags %s values ",
4077 4078 4079 4080 4081 4082
                      winfo->db_name, 
                      superTblInfo->childTblPrefix, 
                      tbl_id, 
                      winfo->db_name, 
                      superTblInfo->sTblName, 
                      tagsValBuf);
4083 4084 4085 4086
            }
            tmfree(tagsValBuf);
          } else if (TBL_ALREADY_EXISTS == superTblInfo->childTblExists) {
            if (0 == len) {
4087 4088
              len += snprintf(pstr + len, 
                      superTblInfo->maxSqlLen - len, 
4089
                      "insert into %s.%s values ",
4090 4091
                      winfo->db_name, 
                      superTblInfo->childTblName + tbl_id * TSDB_TABLE_NAME_LEN);
4092
            } else {
4093 4094
              len += snprintf(pstr + len, 
                      superTblInfo->maxSqlLen - len, 
4095
                      " %s.%s values ",
4096 4097
                      winfo->db_name, 
                      superTblInfo->childTblName + tbl_id * TSDB_TABLE_NAME_LEN);
4098 4099 4100
            }
          } else {  // pre-create child table
            if (0 == len) {
4101 4102
              len += snprintf(pstr + len, 
                      superTblInfo->maxSqlLen - len, 
4103
                      "insert into %s.%s%d values ",
4104 4105 4106
                      winfo->db_name, 
                      superTblInfo->childTblPrefix, 
                      tbl_id);
4107
            } else {
4108 4109
              len += snprintf(pstr + len, 
                      superTblInfo->maxSqlLen - len, 
4110
                      " %s.%s%d values ",
4111 4112 4113
                      winfo->db_name, 
                      superTblInfo->childTblPrefix, 
                      tbl_id);
4114 4115
            }            
          }
4116

4117
          start_time = time_counter;
4118
          for (int j = 0; j < superTblInfo->rowsPerTbl;) {
4119
            int retLen = 0;
4120 4121
            if (0 == strncasecmp(superTblInfo->dataSource, 
                        "sample", strlen("sample"))) {
4122 4123 4124 4125
              retLen = getRowDataFromSample(pstr + len,
                      superTblInfo->maxSqlLen - len,
                      start_time += superTblInfo->timeStampStep,
                      superTblInfo,
4126
                      &sampleUsePos);
4127 4128 4129
              if (retLen < 0) {
                goto free_and_statistics;
              }
4130 4131
            } else if (0 == strncasecmp(
                        superTblInfo->dataSource, "rand", strlen("rand"))) {
4132
              int rand_num = rand_tinyint() % 100;            
4133 4134
              if (0 != superTblInfo->disorderRatio 
                      && rand_num < superTblInfo->disorderRatio) {
4135
                int64_t d = start_time - taosRandom() % superTblInfo->disorderRange;
4136 4137 4138
                retLen = generateRowData(pstr + len,
                        superTblInfo->maxSqlLen - len,
                        d,
4139
                        superTblInfo);
4140
              } else {
4141 4142
                retLen = generateRowData(pstr + len,
                        superTblInfo->maxSqlLen - len,
4143
                        start_time += superTblInfo->timeStampStep,
4144
                        superTblInfo);
4145 4146 4147 4148 4149 4150 4151
              }
              if (retLen < 0) {
                goto free_and_statistics;
              }
            }
            len += retLen;
            //inserted++;
4152
            j++;
4153
            winfo->totalInsertRows++;
4154

4155 4156
            if (inserted >= superTblInfo->insertRows || 
                    (superTblInfo->maxSqlLen - len) < (superTblInfo->lenOfOneRow + 128)) {
4157
              tableSeq = tbl_id + 1;
4158 4159
              printf("config rowsPerTbl and numberOfTblInOneSql not match with max_sql_lenth, please reconfig![lenOfOneRow:%d]\n", 
                      superTblInfo->lenOfOneRow);
4160 4161 4162 4163 4164
              goto send_to_server;
            }
          }
        }

4165
        tableSeq = tbl_id;
4166 4167
        inserted += superTblInfo->rowsPerTbl;

4168
send_to_server:
4169 4170
        if (insert_interval) {
            st = taosGetTimestampUs();
4171

4172 4173
          if (insert_interval > ((et - st)/1000)) {
            int sleep_time = insert_interval - (et -st);
4174
            printf("sleep: %d ms insert interval\n", sleep_time);
4175
            taosMsleep(sleep_time); // ms
4176
          }
4177 4178
        }

4179 4180 4181
        if (0 == strncasecmp(superTblInfo->insertMode, 
                    "taosc", 
                    strlen("taosc"))) {    
4182 4183
          //printf("multi table===== sql: %s \n\n", buffer);
          //int64_t t1 = taosGetTimestampMs();
4184 4185 4186 4187
          int64_t startTs;
          int64_t endTs;
          startTs = taosGetTimestampUs();

4188
          debugPrint("%s() LN%d buff: %s\n", __func__, __LINE__, buffer);
4189 4190
          int affectedRows = queryDbExec(
                  winfo->taos, buffer, INSERT_TYPE);
4191

4192
          if (0 < affectedRows) {
4193 4194 4195 4196 4197 4198
            endTs = taosGetTimestampUs();
            int64_t delay = endTs - startTs;
            if (delay > winfo->maxDelay) winfo->maxDelay = delay;
            if (delay < winfo->minDelay) winfo->minDelay = delay;
            winfo->cntDelay++;
            winfo->totalDelay += delay;
4199 4200
            winfo->avgDelay = (double)winfo->totalDelay / winfo->cntDelay;
            winfo->totalAffectedRows += affectedRows;
4201 4202 4203
          } else {
            fprintf(stderr, "queryDbExec() buffer:\n%s\naffected rows is %d", buffer, affectedRows);
            goto free_and_statistics;
4204
          }
4205 4206 4207

          int64_t  currentPrintTime = taosGetTimestampMs();
          if (currentPrintTime - lastPrintTime > 30*1000) {
4208 4209
            printf("thread[%d] has currently inserted rows: %"PRId64 ", affected rows: %"PRId64 "\n",
                    winfo->threadID,
4210
                    winfo->totalInsertRows,
4211
                    winfo->totalAffectedRows);
4212 4213 4214
            lastPrintTime = currentPrintTime;
          }
          //int64_t t2 = taosGetTimestampMs();          
4215
          //printf("taosc insert sql return, Spent %.4f seconds \n", (double)(t2 - t1)/1000.0); 
4216 4217
        } else {
          //int64_t t1 = taosGetTimestampMs();
4218
          int retCode = postProceSql(g_Dbs.host, g_Dbs.port, buffer);
4219 4220
          //int64_t t2 = taosGetTimestampMs();          
          //printf("http insert sql return, Spent %ld ms \n", t2 - t1);
4221

4222
          if (0 != retCode) {
4223
            printf("========restful return fail, threadID[%d]\n", winfo->threadID);
4224 4225 4226
            goto free_and_statistics;
          }
        }
4227 4228
        if (insert_interval) {
            et = taosGetTimestampUs();
4229 4230
        }

4231 4232 4233
        break;
      }

4234
      if (tableSeq > winfo->end_table_id) {
4235
        if (0 == strncasecmp(superTblInfo->dataSource, "sample", strlen("sample"))) {
4236 4237 4238
          samplePos = sampleUsePos;
        }
        i = inserted;
4239
        time_counter = start_time;
4240 4241 4242 4243
      }
    }   
    
    //printf("========loop %d childTables duration:%"PRId64 "========inserted rows:%d\n", winfo->end_table_id - winfo->start_table_id, et - st, i);
H
hzcheng 已提交
4244 4245
  }

4246
free_and_statistics:
4247
  tmfree(buffer);    
4248 4249
  printf("====thread[%d] completed total inserted rows: %"PRId64 ", affected rows: %"PRId64 "====\n",
          winfo->threadID, winfo->totalInsertRows, winfo->totalAffectedRows);
4250
  return;
H
hzcheng 已提交
4251 4252
}

4253
int32_t generateData(char *res, char **data_type,
4254
        int num_of_cols, int64_t timestamp, int lenOfBinary) {
4255 4256 4257 4258 4259 4260 4261 4262 4263 4264 4265 4266 4267 4268 4269 4270 4271 4272 4273 4274 4275 4276 4277 4278 4279 4280 4281 4282 4283 4284 4285
  memset(res, 0, MAX_DATA_SIZE);
  char *pstr = res;
  pstr += sprintf(pstr, "(%" PRId64, timestamp);
  int c = 0;

  for (; c < MAX_NUM_DATATYPE; c++) {
    if (data_type[c] == NULL) {
      break;
    }
  }

  if (0 == c) {
    perror("data type error!");
    exit(-1);
  }

  for (int i = 0; i < num_of_cols; i++) {
    if (strcasecmp(data_type[i % c], "tinyint") == 0) {
      pstr += sprintf(pstr, ", %d", rand_tinyint() );
    } else if (strcasecmp(data_type[i % c], "smallint") == 0) {
      pstr += sprintf(pstr, ", %d", rand_smallint());
    } else if (strcasecmp(data_type[i % c], "int") == 0) {
      pstr += sprintf(pstr, ", %d", rand_int()); 
    } else if (strcasecmp(data_type[i % c], "bigint") == 0) {
      pstr += sprintf(pstr, ", %" PRId64, rand_bigint());
    } else if (strcasecmp(data_type[i % c], "float") == 0) {
      pstr += sprintf(pstr, ", %10.4f", rand_float());
    } else if (strcasecmp(data_type[i % c], "double") == 0) {
      double t = rand_double();
      pstr += sprintf(pstr, ", %20.8f", t);
    } else if (strcasecmp(data_type[i % c], "bool") == 0) {
4286
      bool b = taosRandom() & 1;
4287 4288
      pstr += sprintf(pstr, ", %s", b ? "true" : "false");
    } else if (strcasecmp(data_type[i % c], "binary") == 0) {
4289 4290
      char *s = malloc(lenOfBinary);
      rand_string(s, lenOfBinary);
4291 4292 4293
      pstr += sprintf(pstr, ", \"%s\"", s);
      free(s);
    }else if (strcasecmp(data_type[i % c], "nchar") == 0) {
4294 4295
      char *s = malloc(lenOfBinary);
      rand_string(s, lenOfBinary);
4296 4297 4298 4299 4300 4301 4302 4303 4304 4305 4306 4307 4308 4309 4310
      pstr += sprintf(pstr, ", \"%s\"", s);
      free(s);
    }

    if (pstr - res > MAX_DATA_SIZE) {
      perror("column length too long, abort");
      exit(-1);
    }
  }

  pstr += sprintf(pstr, ")");

  return (int32_t)(pstr - res);
}

4311
static int prepareSampleDataForSTable(SSuperTable *superTblInfo) {
4312 4313 4314
  char* sampleDataBuf = NULL;

  // each thread read sample data from csv file 
4315 4316 4317 4318 4319
  if (0 == strncasecmp(superTblInfo->dataSource, 
              "sample", 
              strlen("sample"))) {
    sampleDataBuf = calloc(
            superTblInfo->lenOfOneRow * MAX_SAMPLES_ONCE_FROM_FILE, 1);
4320
    if (sampleDataBuf == NULL) {
4321
      fprintf(stderr, "Failed to calloc %d Bytes, reason:%s\n", 
4322 4323
              superTblInfo->lenOfOneRow * MAX_SAMPLES_ONCE_FROM_FILE, 
              strerror(errno));
4324
      return -1;
4325
    }
4326

4327
    int ret = readSampleFromCsvFileToMem(superTblInfo);
4328
    if (0 != ret) {
4329
      tmfree(sampleDataBuf);
4330
      return -1;
4331
    }
H
Hui Li 已提交
4332
  }
4333

4334 4335 4336 4337 4338
  superTblInfo->sampleDataBuf = sampleDataBuf;

  return 0;
}

4339 4340 4341 4342 4343 4344
static int execInsert(threadInfo *winfo, char *buffer, int k)
{
  int affectedRows;
  SSuperTable* superTblInfo = winfo->superTblInfo;

  if (superTblInfo) {
4345
    if (0 == strncasecmp(superTblInfo->insertMode, "taosc", strlen("taosc"))) {
4346 4347
      verbosePrint("%s() LN%d %s\n", __func__, __LINE__, buffer);
      affectedRows = queryDbExec(winfo->taos, buffer, INSERT_TYPE);
4348
    } else {
4349 4350
      verbosePrint("%s() LN%d %s\n", __func__, __LINE__, buffer);
      int retCode = postProceSql(g_Dbs.host, g_Dbs.port, buffer);
4351

4352 4353 4354 4355 4356 4357
      if (0 != retCode) {
        affectedRows = -1;
        printf("========restful return fail, threadID[%d]\n", winfo->threadID);
      } else {
        affectedRows = k;
      }
4358
    }
4359
  } else {
4360 4361
    verbosePrint("%s() LN%d %s\n", __func__, __LINE__, buffer);
    affectedRows = queryDbExec(winfo->taos, buffer, 1);
4362 4363 4364 4365 4366
  }

  return affectedRows;
}

4367 4368
static int generateDataBuffer(int32_t tableSeq,
        threadInfo *pThreadInfo, char *buffer,
4369 4370 4371 4372
        int64_t insertRows,
        int64_t startFrom, int64_t startTime, int *pSampleUsePos)
{
  SSuperTable* superTblInfo = pThreadInfo->superTblInfo;
4373

4374 4375
  int ncols_per_record = 1; // count first col ts

4376
  if (superTblInfo == NULL) {
4377 4378 4379 4380 4381
    int datatypeSeq = 0;
    while(g_args.datatype[datatypeSeq]) {
        datatypeSeq ++;
        ncols_per_record ++;
    }
4382 4383
  }

4384 4385
  assert(buffer != NULL);

4386 4387 4388
  char *pChildTblName;
  int childTblCount;

4389 4390 4391 4392 4393 4394
  pChildTblName = calloc(TSDB_TABLE_NAME_LEN, 1);
  if (NULL == pChildTblName) {
    fprintf(stderr, "failed to alloc memory %d\n", TSDB_TABLE_NAME_LEN);
    return -1;
  }

4395 4396 4397 4398 4399 4400 4401 4402 4403 4404 4405
  if (superTblInfo && (superTblInfo->childTblOffset > 0)) {
      // select tbname from stb limit 1 offset tableSeq
    getChildNameOfSuperTableWithLimitAndOffset(pThreadInfo->taos,
            pThreadInfo->db_name, superTblInfo->sTblName,
            &pChildTblName, &childTblCount,
            1, tableSeq);
  } else {
      snprintf(pChildTblName, TSDB_TABLE_NAME_LEN, "%s%d",
        superTblInfo?superTblInfo->childTblPrefix:g_args.tb_prefix, tableSeq);
  }

4406
  memset(buffer, 0, superTblInfo?superTblInfo->maxSqlLen:g_args.max_sql_len);
4407

4408
  char *pstr = buffer;
4409

4410 4411 4412 4413
  if (superTblInfo) {
    if (AUTO_CREATE_SUBTBL == superTblInfo->autoCreateTable) {
      char* tagsValBuf = NULL;
      if (0 == superTblInfo->tagSource) {
4414
            tagsValBuf = generateTagVaulesForStb(superTblInfo);
4415
      } else {
4416
            tagsValBuf = getTagValueFromTagSample(
4417
                    superTblInfo,
4418
                    tableSeq % superTblInfo->tagSampleCount);
4419 4420
      }
      if (NULL == tagsValBuf) {
4421 4422 4423
        free(pChildTblName);
        fprintf(stderr, "tag buf failed to allocate  memory\n");
        return -1;
4424
      }
4425

4426
      pstr += snprintf(pstr,
4427
                  superTblInfo->maxSqlLen,
4428
                  "insert into %s.%s using %s.%s tags %s values",
4429
                  pThreadInfo->db_name,
4430
                  pChildTblName,
4431
                  pThreadInfo->db_name,
4432
                  superTblInfo->sTblName,
4433
                  tagsValBuf);
4434 4435 4436
      tmfree(tagsValBuf);
    } else if (TBL_ALREADY_EXISTS == superTblInfo->childTblExists) {
      pstr += snprintf(pstr,
4437 4438
                  superTblInfo->maxSqlLen,
                  "insert into %s.%s values",
4439
                  pThreadInfo->db_name,
4440
                  superTblInfo->childTblName + tableSeq * TSDB_TABLE_NAME_LEN);
4441 4442
    } else {
      pstr += snprintf(pstr,
4443
                  (superTblInfo?superTblInfo->maxSqlLen:g_args.max_sql_len),
4444
                  "insert into %s.%s values",
4445
                  pThreadInfo->db_name,
4446
                  pChildTblName);
4447 4448 4449
    }
  } else {
      pstr += snprintf(pstr,
4450
                  g_args.max_sql_len,
4451
                  "insert into %s.%s values",
4452
                  pThreadInfo->db_name,
4453
                  pChildTblName);
4454
  }
4455

4456 4457
  int k;
  int len = 0;
4458

4459 4460 4461
  verbosePrint("%s() LN%d num_of_RPR=%d\n", __func__, __LINE__, g_args.num_of_RPR);
  for (k = 0; k < g_args.num_of_RPR;) {
    if (superTblInfo) {
4462 4463
          int retLen = 0;

4464 4465 4466
        if (0 == strncasecmp(superTblInfo->dataSource,
                    "sample", strlen("sample"))) {
          retLen = getRowDataFromSample(
4467 4468
                    pstr + len, 
                    superTblInfo->maxSqlLen - len, 
4469
                    startTime + superTblInfo->timeStampStep * startFrom,
4470
                    superTblInfo, 
4471
                    pSampleUsePos);
4472 4473 4474 4475
       } else if (0 == strncasecmp(superTblInfo->dataSource,
                   "rand", strlen("rand"))) {
          int rand_num = rand_tinyint() % 100;
          if (0 != superTblInfo->disorderRatio 
4476
                    && rand_num < superTblInfo->disorderRatio) {
4477
            int64_t d = startTime - taosRandom() % superTblInfo->disorderRange;
4478
            retLen = generateRowData(
4479
                      pstr + len, 
4480 4481
                      superTblInfo->maxSqlLen - len,
                      d, 
4482
                      superTblInfo);
4483
              //printf("disorder rows, rand_num:%d, last ts:%"PRId64" current ts:%"PRId64"\n", rand_num, start_time, d);
4484
        } else {
4485
          retLen = generateRowData(
4486 4487
                      pstr + len, 
                      superTblInfo->maxSqlLen - len, 
4488
                      startTime + superTblInfo->timeStampStep * startFrom,
4489
                      superTblInfo);
4490
        }
4491

4492
        if (retLen < 0) {
4493 4494
          free(pChildTblName);
          return -1;
4495
        }
4496

4497
        len += retLen;
4498 4499 4500 4501 4502

        if (len >= (superTblInfo->maxSqlLen - 256)) {    // reserve for overwrite
            k++;
            break;
        }
4503 4504
      }
    } else {
4505
      int rand_num = taosRandom() % 100;
4506 4507 4508 4509
          char data[MAX_DATA_SIZE];
          char **data_type = g_args.datatype;
          int lenOfBinary = g_args.len_of_binary;

4510
      if ((g_args.disorderRatio != 0)
4511 4512
                && (rand_num < g_args.disorderRange)) {
             
4513
        int64_t d = startTime - taosRandom() % 1000000 + rand_num;
4514
        len = generateData(data, data_type,
4515
                  ncols_per_record, d, lenOfBinary);
4516
      } else {
4517
            len = generateData(data, data_type,
4518 4519 4520 4521
                  ncols_per_record,
                  startTime + DEFAULT_TIMESTAMP_STEP * startFrom,
                  lenOfBinary);
      }
4522

4523
      pstr += sprintf(pstr, " %s", data);
4524
      //assert(len + pstr - buffer < BUFFER_SIZE);
4525 4526 4527
      if (len + pstr - buffer >= (g_args.max_sql_len - 256)) { // too long
          k++;
          break;
4528 4529 4530
      }
    }

4531
    verbosePrint("%s() LN%d len=%d k=%d \nbuffer=%p\n", __func__, __LINE__, len, k, buffer);
4532 4533 4534

    k++;
    startFrom ++;
4535

4536 4537 4538
    if (startFrom >= insertRows)
      break;
  }
4539

4540 4541
  if (pChildTblName)
      free(pChildTblName);
4542

4543 4544
  return k;
}
4545

4546 4547 4548 4549 4550 4551 4552 4553 4554 4555 4556 4557 4558
// sync insertion
/*
   1 thread: 100 tables * 2000  rows/s
   1 thread: 10  tables * 20000 rows/s
   6 thread: 300 tables * 2000  rows/s

   2 taosinsertdata , 1 thread:  10  tables * 20000 rows/s
*/
static void* syncWrite(void *sarg) {

  threadInfo *winfo = (threadInfo *)sarg; 
  SSuperTable* superTblInfo = winfo->superTblInfo;

4559 4560 4561 4562 4563 4564 4565
  char* buffer = calloc(superTblInfo?superTblInfo->maxSqlLen:g_args.max_sql_len, 1);
  if (NULL == buffer) {
    fprintf(stderr, "Failed to alloc %d Bytes, reason:%s\n",
              superTblInfo?superTblInfo->maxSqlLen:g_args.max_sql_len,
              strerror(errno));
    return NULL;
  }
4566 4567

  if (superTblInfo) {
4568
    if (0 != prepareSampleDataForSTable(superTblInfo))
4569 4570 4571 4572 4573 4574 4575 4576 4577 4578 4579 4580 4581 4582 4583
        return NULL;

    if (superTblInfo->numberOfTblInOneSql > 0) {
      syncWriteForNumberOfTblInOneSql(winfo, superTblInfo->sampleDataBuf);
      tmfree(superTblInfo->sampleDataBuf);
      return NULL;
    }
  }
 
  int   samplePos     = 0;

  int64_t lastPrintTime = taosGetTimestampMs();
  int64_t startTs = taosGetTimestampUs();
  int64_t endTs;

4584 4585
  int insert_interval = superTblInfo?superTblInfo->insertInterval:
      g_args.insert_interval;
4586 4587 4588 4589 4590 4591 4592 4593
  uint64_t st = 0;
  uint64_t et = 0xffffffff;

  winfo->totalInsertRows = 0;
  winfo->totalAffectedRows = 0;

  int sampleUsePos;

4594 4595
  for (uint32_t tableSeq = winfo->start_table_id; tableSeq <= winfo->end_table_id;
        tableSeq ++) {
4596 4597 4598 4599 4600 4601 4602 4603
    int64_t start_time = winfo->start_time;

    int64_t insertRows = (superTblInfo)?superTblInfo->insertRows:g_args.num_of_DPT;
    verbosePrint("%s() LN%d insertRows=%"PRId64"\n", __func__, __LINE__, insertRows);

    for (int64_t i = 0; i < insertRows;) {
      if (insert_interval) {
            st = taosGetTimestampUs();
4604
      }
4605 4606 4607

      sampleUsePos = samplePos;

4608
      int generated = generateDataBuffer(tableSeq, winfo, buffer, insertRows,
4609 4610 4611 4612 4613 4614 4615
            i, start_time, &sampleUsePos);
      if (generated > 0)
        i += generated;
      else
        goto free_and_statistics_2;

      int affectedRows = execInsert(winfo, buffer, generated);
4616 4617
      if (affectedRows < 0)
        goto free_and_statistics_2;
4618

4619

4620
      winfo->totalInsertRows += generated;
4621
      winfo->totalAffectedRows += affectedRows;
4622 4623 4624 4625 4626 4627 4628 4629 4630 4631

      endTs = taosGetTimestampUs();
      int64_t delay = endTs - startTs;
      if (delay > winfo->maxDelay) winfo->maxDelay = delay;
      if (delay < winfo->minDelay) winfo->minDelay = delay;
      winfo->cntDelay++;
      winfo->totalDelay += delay;

      int64_t  currentPrintTime = taosGetTimestampMs();
      if (currentPrintTime - lastPrintTime > 30*1000) {
4632 4633
        printf("thread[%d] has currently inserted rows: %"PRId64 ", affected rows: %"PRId64 "\n",
                    winfo->threadID,
4634
                    winfo->totalInsertRows,
4635 4636 4637 4638
                    winfo->totalAffectedRows);
        lastPrintTime = currentPrintTime;
      }

4639
      if (i >= insertRows)
4640
        break;
4641 4642 4643 4644 4645 4646

      if (insert_interval) {
        et = taosGetTimestampUs();

        if (insert_interval > ((et - st)/1000) ) {
            int sleep_time = insert_interval - (et -st)/1000;
4647
            verbosePrint("%s() LN%d sleep: %d ms for insert interval\n", __func__, __LINE__, sleep_time);
4648 4649
            taosMsleep(sleep_time); // ms
        }
4650 4651 4652
      }
    }   // num_of_DPT

4653
    if ((tableSeq == winfo->end_table_id) && superTblInfo &&
4654 4655
        (0 == strncasecmp(
                    superTblInfo->dataSource, "sample", strlen("sample")))) {
4656
          samplePos = sampleUsePos;
4657
    }
4658
  } // tableSeq
4659

4660
free_and_statistics_2:
4661
  tmfree(buffer);
4662 4663
  if (superTblInfo)
    tmfree(superTblInfo->sampleDataBuf);
4664

4665 4666
  printf("====thread[%d] completed total inserted rows: %"PRId64 ", total affected rows: %"PRId64 "====\n", 
          winfo->threadID, 
4667
          winfo->totalInsertRows, 
4668
          winfo->totalAffectedRows);
4669
  return NULL;
H
Hui Li 已提交
4670 4671
}

4672 4673
void callBack(void *param, TAOS_RES *res, int code) {
  threadInfo* winfo = (threadInfo*)param; 
4674
  SSuperTable* superTblInfo = winfo->superTblInfo;
4675

4676 4677 4678 4679 4680
  int insert_interval = superTblInfo?superTblInfo->insertInterval:g_args.insert_interval;
  if (insert_interval) {
    winfo->et = taosGetTimestampUs();
    if (((winfo->et - winfo->st)/1000) < insert_interval) {
      taosMsleep(insert_interval - (winfo->et - winfo->st)/1000); // ms
4681
    }
H
Hui Li 已提交
4682 4683
  }
  
4684 4685 4686 4687
  char *buffer = calloc(1, winfo->superTblInfo->maxSqlLen);
  char *data   = calloc(1, MAX_DATA_SIZE);
  char *pstr = buffer;
  pstr += sprintf(pstr, "insert into %s.%s%d values", winfo->db_name, winfo->tb_prefix, winfo->start_table_id);
4688
//  if (winfo->counter >= winfo->superTblInfo->insertRows) {
4689
  if (winfo->counter >= g_args.num_of_RPR) {
4690 4691 4692 4693 4694 4695 4696 4697 4698 4699
    winfo->start_table_id++;
    winfo->counter = 0;
  }
  if (winfo->start_table_id > winfo->end_table_id) {
    tsem_post(&winfo->lock_sem);
    free(buffer);
    free(data);
    taos_free_result(res);
    return;
  }
H
Hui Li 已提交
4700
  
4701
  for (int i = 0; i < g_args.num_of_RPR; i++) {
4702
    int rand_num = taosRandom() % 100;
4703 4704
    if (0 != winfo->superTblInfo->disorderRatio && rand_num < winfo->superTblInfo->disorderRatio)
    {
4705
      int64_t d = winfo->lastTs - taosRandom() % 1000000 + rand_num;
4706 4707 4708
      //generateData(data, datatype, ncols_per_record, d, len_of_binary);
      (void)generateRowData(data, MAX_DATA_SIZE, d, winfo->superTblInfo);
    } else {
4709
      //generateData(data, datatype, ncols_per_record, start_time += 1000, len_of_binary);
4710
      (void)generateRowData(data, MAX_DATA_SIZE, winfo->lastTs += 1000, winfo->superTblInfo);
H
Hui Li 已提交
4711
    }
4712 4713
    pstr += sprintf(pstr, "%s", data);
    winfo->counter++;
H
Hui Li 已提交
4714

4715 4716
    if (winfo->counter >= winfo->superTblInfo->insertRows) {
      break;
H
Hui Li 已提交
4717 4718
    }
  }
4719

4720 4721
  if (insert_interval) {
    winfo->st = taosGetTimestampUs();
4722 4723 4724 4725
  }
  taos_query_a(winfo->taos, buffer, callBack, winfo);
  free(buffer);
  free(data);
H
Hui Li 已提交
4726

4727
  taos_free_result(res);
H
Hui Li 已提交
4728 4729
}

4730 4731
void *asyncWrite(void *sarg) {
  threadInfo *winfo = (threadInfo *)sarg;
4732
  SSuperTable* superTblInfo = winfo->superTblInfo;
4733 4734 4735 4736 4737

  winfo->st = 0;
  winfo->et = 0;
  winfo->lastTs = winfo->start_time;
  
4738 4739 4740
  int insert_interval = superTblInfo?superTblInfo->insertInterval:g_args.insert_interval;
  if (insert_interval) {
    winfo->st = taosGetTimestampUs();
H
Hui Li 已提交
4741
  }
4742 4743 4744
  taos_query_a(winfo->taos, "show databases", callBack, winfo);

  tsem_wait(&(winfo->lock_sem));
H
Hui Li 已提交
4745 4746 4747 4748

  return NULL;
}

4749 4750
static void startMultiThreadInsertData(int threads, char* db_name,
        char* precision,SSuperTable* superTblInfo) {
4751 4752

    pthread_t *pids = malloc(threads * sizeof(pthread_t));
4753 4754
    assert(pids != NULL);

4755
    threadInfo *infos = malloc(threads * sizeof(threadInfo));
4756 4757
    assert(infos != NULL);

4758 4759
    memset(pids, 0, threads * sizeof(pthread_t));
    memset(infos, 0, threads * sizeof(threadInfo));
H
Hui Li 已提交
4760

4761 4762
    int ntables = 0;
    if (superTblInfo)
4763 4764 4765 4766
        if (superTblInfo->childTblOffset)
            ntables = superTblInfo->childTblLimit;
        else
            ntables = superTblInfo->childTblCount;
4767 4768
    else
        ntables = g_args.num_of_tables;
H
Hui Li 已提交
4769

4770 4771 4772 4773 4774 4775 4776 4777 4778 4779
    int a = ntables / threads;
    if (a < 1) {
        threads = ntables;
        a = 1;
    }

    int b = 0;
    if (threads != 0) {
        b = ntables % threads;
    }
4780 4781 4782 4783 4784 4785 4786 4787 4788 4789 4790 4791 4792 4793 4794 4795 4796

  //TAOS* taos;
  //if (0 == strncasecmp(superTblInfo->insertMode, "taosc", 5)) {
  //  taos = taos_connect(g_Dbs.host, g_Dbs.user, g_Dbs.password, db_name, g_Dbs.port);
  //  if (NULL == taos) {
  //    printf("connect to server fail, reason: %s\n", taos_errstr(NULL));
  //    exit(-1);
  //  }
  //}

  int32_t timePrec = TSDB_TIME_PRECISION_MILLI;
  if (0 != precision[0]) {
    if (0 == strncasecmp(precision, "ms", 2)) {
      timePrec = TSDB_TIME_PRECISION_MILLI;
    }  else if (0 == strncasecmp(precision, "us", 2)) {
      timePrec = TSDB_TIME_PRECISION_MICRO;
    }  else {
4797
      fprintf(stderr, "No support precision: %s\n", precision);
4798 4799 4800 4801
      exit(-1);
    }
  }

4802 4803 4804 4805
  int64_t start_time; 
  if (superTblInfo) {
    if (0 == strncasecmp(superTblInfo->startTimestamp, "now", 3)) {
        start_time = taosGetTimestamp(timePrec);
4806
    } else {
4807
        if (TSDB_CODE_SUCCESS != taosParseTime(
4808 4809 4810
            superTblInfo->startTimestamp,
            &start_time,
            strlen(superTblInfo->startTimestamp),
4811 4812 4813 4814
            timePrec, 0)) {
            printf("ERROR to parse time!\n");
            exit(-1);
        }
4815 4816 4817
    }
  } else {
     start_time = 1500000000000;
4818 4819 4820
  }

  double start = getCurrentTime();
4821
 
4822 4823 4824 4825 4826 4827 4828
  int last;

  if ((superTblInfo) && (superTblInfo->childTblOffset))
      last = superTblInfo->childTblOffset;
  else
      last = 0;

H
Hui Li 已提交
4829
  for (int i = 0; i < threads; i++) {
4830
    threadInfo *t_info = infos + i;
H
Hui Li 已提交
4831 4832
    t_info->threadID = i;
    tstrncpy(t_info->db_name, db_name, MAX_DB_NAME_SIZE);
4833 4834 4835
    t_info->superTblInfo = superTblInfo;

    t_info->start_time = start_time;
4836
    t_info->minDelay = INT16_MAX;
4837

4838 4839
    if ((NULL == superTblInfo) ||
            (0 == strncasecmp(superTblInfo->insertMode, "taosc", 5))) {
4840
      //t_info->taos = taos;
4841
      t_info->taos = taos_connect(
4842
              g_Dbs.host, g_Dbs.user,
4843
              g_Dbs.password, db_name, g_Dbs.port);
4844
      if (NULL == t_info->taos) {
4845 4846
        printf("connect to server fail from insert sub thread, reason: %s\n",
                taos_errstr(NULL));
4847 4848 4849 4850 4851 4852
        exit(-1);
      }
    } else {
      t_info->taos = NULL;
    }

4853 4854
    if ((NULL == superTblInfo)
            || (0 == superTblInfo->multiThreadWriteOneTbl)) {
4855 4856 4857 4858 4859 4860 4861 4862 4863 4864 4865
      t_info->start_table_id = last;
      t_info->end_table_id = i < b ? last + a : last + a - 1;
      last = t_info->end_table_id + 1;
    } else {
      t_info->start_table_id = 0;
      t_info->end_table_id = superTblInfo->childTblCount - 1;
      t_info->start_time = t_info->start_time + rand_int() % 10000 - rand_tinyint();
    }

    tsem_init(&(t_info->lock_sem), 0, 0);
    if (SYNC == g_Dbs.queryMode) {
4866
      pthread_create(pids + i, NULL, syncWrite, t_info);
4867
    } else {
4868 4869
      pthread_create(pids + i, NULL, asyncWrite, t_info);
    }
H
Hui Li 已提交
4870
  }
4871

H
Hui Li 已提交
4872 4873 4874 4875
  for (int i = 0; i < threads; i++) {
    pthread_join(pids[i], NULL);
  }

4876 4877 4878
  int64_t totalDelay = 0;
  int64_t maxDelay = 0;
  int64_t minDelay = INT16_MAX;
4879
  int64_t cntDelay = 1;
4880 4881
  double  avgDelay = 0;

H
Hui Li 已提交
4882
  for (int i = 0; i < threads; i++) {
4883 4884
    threadInfo *t_info = infos + i;

S
TD-1057  
Shengliang Guan 已提交
4885
    tsem_destroy(&(t_info->lock_sem));
4886 4887
    taos_close(t_info->taos);

4888 4889
    if (superTblInfo) {
        superTblInfo->totalAffectedRows += t_info->totalAffectedRows;
4890
        superTblInfo->totalInsertRows += t_info->totalInsertRows;
4891
    }
4892 4893 4894 4895

    totalDelay  += t_info->totalDelay;
    cntDelay   += t_info->cntDelay;
    if (t_info->maxDelay > maxDelay) maxDelay = t_info->maxDelay;
4896
    if (t_info->minDelay < minDelay) minDelay = t_info->minDelay;
H
Hui Li 已提交
4897
  }
4898
  cntDelay -= 1;
H
Hui Li 已提交
4899

4900
  if (cntDelay == 0)    cntDelay = 1;
4901 4902
  avgDelay = (double)totalDelay / cntDelay;

4903
  double end = getCurrentTime();
4904
  double t = end - start;
4905 4906

  if (superTblInfo) {
4907
    printf("Spent %.4f seconds to insert rows: %"PRId64", affected rows: %"PRId64" with %d thread(s) into %s.%s. %2.f records/second\n\n",
4908
          t, superTblInfo->totalInsertRows,
4909 4910
          superTblInfo->totalAffectedRows,
          threads, db_name, superTblInfo->sTblName,
4911
          superTblInfo->totalInsertRows / t);
4912
    fprintf(g_fpOfInsertResult,
4913
          "Spent %.4f seconds to insert rows: %"PRId64", affected rows: %"PRId64" with %d thread(s) into %s.%s. %2.f records/second\n\n",
4914
          t, superTblInfo->totalInsertRows,
4915 4916
          superTblInfo->totalAffectedRows,
          threads, db_name, superTblInfo->sTblName,
4917
          superTblInfo->totalInsertRows/ t);
4918
  }
4919 4920 4921 4922 4923

  printf("insert delay, avg: %10.6fms, max: %10.6fms, min: %10.6fms\n\n",
          avgDelay/1000.0, (double)maxDelay/1000.0, (double)minDelay/1000.0);
  fprintf(g_fpOfInsertResult, "insert delay, avg:%10.6fms, max: %10.6fms, min: %10.6fms\n\n",
          avgDelay/1000.0, (double)maxDelay/1000.0, (double)minDelay/1000.0);
4924 4925 4926
  
  //taos_close(taos);

H
Hui Li 已提交
4927
  free(pids);
4928
  free(infos);
H
Hui Li 已提交
4929 4930
}

H
hzcheng 已提交
4931
void *readTable(void *sarg) {
4932 4933
#if 1  
  threadInfo *rinfo = (threadInfo *)sarg;
H
hzcheng 已提交
4934 4935
  TAOS *taos = rinfo->taos;
  char command[BUFFER_SIZE] = "\0";
4936
  uint64_t sTime = rinfo->start_time;
H
hzcheng 已提交
4937 4938
  char *tb_prefix = rinfo->tb_prefix;
  FILE *fp = fopen(rinfo->fp, "a");
H
Hui Li 已提交
4939 4940 4941 4942
  if (NULL == fp) {
    printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
    return NULL;
  }
4943

4944
    int num_of_DPT;
4945
/*  if (rinfo->superTblInfo) {
4946 4947
    num_of_DPT = rinfo->superTblInfo->insertRows; //  nrecords_per_table;
  } else {
4948
  */
4949
      num_of_DPT = g_args.num_of_DPT;
4950
//  }
4951

H
hzcheng 已提交
4952 4953
  int num_of_tables = rinfo->end_table_id - rinfo->start_table_id + 1;
  int totalData = num_of_DPT * num_of_tables;
4954
  bool do_aggreFunc = g_Dbs.do_aggreFunc;
H
hzcheng 已提交
4955 4956 4957 4958 4959 4960 4961 4962 4963 4964 4965 4966

  int n = do_aggreFunc ? (sizeof(aggreFunc) / sizeof(aggreFunc[0])) : 2;
  if (!do_aggreFunc) {
    printf("\nThe first field is either Binary or Bool. Aggregation functions are not supported.\n");
  }
  printf("%d records:\n", totalData);
  fprintf(fp, "| QFunctions |    QRecords    |   QSpeed(R/s)   |  QLatency(ms) |\n");

  for (int j = 0; j < n; j++) {
    double totalT = 0;
    int count = 0;
    for (int i = 0; i < num_of_tables; i++) {
L
lihui 已提交
4967
      sprintf(command, "select %s from %s%d where ts>= %" PRId64, aggreFunc[j], tb_prefix, i, sTime);
H
hzcheng 已提交
4968 4969

      double t = getCurrentTime();
S
Shuaiqiang Chang 已提交
4970 4971
      TAOS_RES *pSql = taos_query(taos, command);
      int32_t code = taos_errno(pSql);
H
hzcheng 已提交
4972

S
Shuaiqiang Chang 已提交
4973
      if (code != 0) {
H
Haojun Liao 已提交
4974
        fprintf(stderr, "Failed to query:%s\n", taos_errstr(pSql));
S
Shuaiqiang Chang 已提交
4975
        taos_free_result(pSql);
H
hzcheng 已提交
4976
        taos_close(taos);
4977
        fclose(fp);
4978
        return NULL;
H
hzcheng 已提交
4979 4980
      }

S
Shuaiqiang Chang 已提交
4981
      while (taos_fetch_row(pSql) != NULL) {
H
hzcheng 已提交
4982 4983 4984 4985 4986 4987
        count++;
      }

      t = getCurrentTime() - t;
      totalT += t;

S
Shuaiqiang Chang 已提交
4988
      taos_free_result(pSql);
H
hzcheng 已提交
4989 4990 4991
    }

    fprintf(fp, "|%10s  |   %10d   |  %12.2f   |   %10.2f  |\n",
S
slguan 已提交
4992
            aggreFunc[j][0] == '*' ? "   *   " : aggreFunc[j], totalData,
H
hzcheng 已提交
4993 4994 4995 4996 4997
            (double)(num_of_tables * num_of_DPT) / totalT, totalT * 1000);
    printf("select %10s took %.6f second(s)\n", aggreFunc[j], totalT);
  }
  fprintf(fp, "\n");
  fclose(fp);
4998
#endif
H
hzcheng 已提交
4999 5000 5001 5002
  return NULL;
}

void *readMetric(void *sarg) {
5003 5004
#if 1  
  threadInfo *rinfo = (threadInfo *)sarg;
H
hzcheng 已提交
5005 5006 5007
  TAOS *taos = rinfo->taos;
  char command[BUFFER_SIZE] = "\0";
  FILE *fp = fopen(rinfo->fp, "a");
H
Hui Li 已提交
5008 5009 5010 5011
  if (NULL == fp) {
    printf("fopen %s fail, reason:%s.\n", rinfo->fp, strerror(errno));
    return NULL;
  }
5012 5013

  int num_of_DPT = rinfo->superTblInfo->insertRows;
H
hzcheng 已提交
5014 5015
  int num_of_tables = rinfo->end_table_id - rinfo->start_table_id + 1;
  int totalData = num_of_DPT * num_of_tables;
5016
  bool do_aggreFunc = g_Dbs.do_aggreFunc;
H
hzcheng 已提交
5017 5018 5019 5020 5021 5022 5023 5024 5025

  int n = do_aggreFunc ? (sizeof(aggreFunc) / sizeof(aggreFunc[0])) : 2;
  if (!do_aggreFunc) {
    printf("\nThe first field is either Binary or Bool. Aggregation functions are not supported.\n");
  }
  printf("%d records:\n", totalData);
  fprintf(fp, "Querying On %d records:\n", totalData);

  for (int j = 0; j < n; j++) {
sangshuduo's avatar
sangshuduo 已提交
5026
    char condition[BUFFER_SIZE - 30] = "\0";
B
Bomin Zhang 已提交
5027
    char tempS[64] = "\0";
H
hzcheng 已提交
5028 5029 5030 5031 5032

    int m = 10 < num_of_tables ? 10 : num_of_tables;

    for (int i = 1; i <= m; i++) {
      if (i == 1) {
5033
        sprintf(tempS, "t1 = %d", i);
H
hzcheng 已提交
5034
      } else {
5035
        sprintf(tempS, " or t1 = %d ", i);
H
hzcheng 已提交
5036 5037 5038
      }
      strcat(condition, tempS);

L
liu0x54 已提交
5039
      sprintf(command, "select %s from meters where %s", aggreFunc[j], condition);
H
hzcheng 已提交
5040 5041 5042 5043 5044 5045

      printf("Where condition: %s\n", condition);
      fprintf(fp, "%s\n", command);

      double t = getCurrentTime();

S
Shuaiqiang Chang 已提交
5046 5047 5048 5049
      TAOS_RES *pSql = taos_query(taos, command);
      int32_t code = taos_errno(pSql);

      if (code != 0) {
H
Haojun Liao 已提交
5050
        fprintf(stderr, "Failed to query:%s\n", taos_errstr(pSql));
S
Shuaiqiang Chang 已提交
5051
        taos_free_result(pSql);
H
hzcheng 已提交
5052
        taos_close(taos);
5053
        fclose(fp);
5054
        return NULL;
H
hzcheng 已提交
5055 5056
      }
      int count = 0;
S
Shuaiqiang Chang 已提交
5057
      while (taos_fetch_row(pSql) != NULL) {
H
hzcheng 已提交
5058 5059 5060 5061 5062 5063 5064
        count++;
      }
      t = getCurrentTime() - t;

      fprintf(fp, "| Speed: %12.2f(per s) | Latency: %.4f(ms) |\n", num_of_tables * num_of_DPT / t, t * 1000);
      printf("select %10s took %.6f second(s)\n\n", aggreFunc[j], t);

S
Shuaiqiang Chang 已提交
5065
      taos_free_result(pSql);
H
hzcheng 已提交
5066 5067 5068 5069
    }
    fprintf(fp, "\n");
  }
  fclose(fp);
5070
#endif
H
hzcheng 已提交
5071 5072 5073
  return NULL;
}

H
Hui Li 已提交
5074

5075
static int insertTestProcess() {
5076

5077 5078 5079
  setupForAnsiEscape();
  int ret = printfInsertMeta();
  resetAfterAnsiEscape();
5080

5081 5082 5083
  if (ret == -1)
    exit(EXIT_FAILURE);

5084
  debugPrint("%d result file: %s\n", __LINE__, g_Dbs.resultFile);
5085 5086 5087 5088 5089
  g_fpOfInsertResult = fopen(g_Dbs.resultFile, "a");
  if (NULL == g_fpOfInsertResult) {
    fprintf(stderr, "Failed to open %s for save result\n", g_Dbs.resultFile);
    return -1;
  }
5090

5091 5092
  printfInsertMetaToFile(g_fpOfInsertResult);

5093 5094 5095 5096
  if (!g_args.answer_yes) {
    printf("Press enter key to continue\n\n");
    (void)getchar();
  }
5097
 
5098 5099 5100
  init_rand_data();

  // create database and super tables
5101
  if(createDatabases() != 0) {
5102 5103 5104
    fclose(g_fpOfInsertResult);
    return -1;
  }
5105 5106

  // pretreatement
5107
  prepareSampleData();
5108

5109 5110 5111 5112 5113 5114 5115
  double start;
  double end;

  // create child tables
  start = getCurrentTime();
  createChildTables();
  end = getCurrentTime();
5116

5117
  if (g_totalChildTables > 0) {
5118
    printf("Spent %.4f seconds to create %d tables with %d thread(s)\n\n",
5119
            end - start, g_totalChildTables, g_Dbs.threadCount);
5120 5121
    fprintf(g_fpOfInsertResult,
            "Spent %.4f seconds to create %d tables with %d thread(s)\n\n",
5122
            end - start, g_totalChildTables, g_Dbs.threadCount);
5123
  }
5124

5125
  taosMsleep(1000);
5126 5127
  // create sub threads for inserting data
  //start = getCurrentTime();
5128 5129 5130 5131 5132 5133
  for (int i = 0; i < g_Dbs.dbCount; i++) {
    if (g_Dbs.db[i].superTblCount > 0) {
      for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) {
        SSuperTable* superTblInfo = &g_Dbs.db[i].superTbls[j];
        if (0 == g_Dbs.db[i].superTbls[j].insertRows) {
          continue;
5134
        }
5135 5136 5137 5138 5139 5140 5141 5142 5143 5144 5145 5146
        startMultiThreadInsertData(
          g_Dbs.threadCount, 
          g_Dbs.db[i].dbName, 
          g_Dbs.db[i].dbCfg.precision, 
          superTblInfo);
        }
    } else {
        startMultiThreadInsertData(
          g_Dbs.threadCount, 
          g_Dbs.db[i].dbName, 
          g_Dbs.db[i].dbCfg.precision, 
          NULL);
H
Hui Li 已提交
5147
    }
5148
  }
5149
  //end = getCurrentTime();
5150

5151
  //int64_t    totalInsertRows = 0;
5152 5153 5154
  //int64_t    totalAffectedRows = 0;
  //for (int i = 0; i < g_Dbs.dbCount; i++) {    
  //  for (int j = 0; j < g_Dbs.db[i].superTblCount; j++) {
5155
  //  totalInsertRows+= g_Dbs.db[i].superTbls[j].totalInsertRows;
5156 5157
  //  totalAffectedRows += g_Dbs.db[i].superTbls[j].totalAffectedRows;
  //}
5158
  //printf("Spent %.4f seconds to insert rows: %"PRId64", affected rows: %"PRId64" with %d thread(s)\n\n", end - start, totalInsertRows, totalAffectedRows, g_Dbs.threadCount);
5159
  postFreeResource();
5160

5161 5162 5163 5164 5165 5166 5167 5168 5169 5170 5171 5172 5173
  return 0;
}

void *superQueryProcess(void *sarg) {
  threadInfo *winfo = (threadInfo *)sarg; 

  //char sqlStr[MAX_TB_NAME_SIZE*2];
  //sprintf(sqlStr, "use %s", g_queryInfo.dbName);
  //queryDB(winfo->taos, sqlStr);
  
  int64_t st = 0;
  int64_t et = 0;
  while (1) {
5174
    if (g_queryInfo.superQueryInfo.rate && (et - st) < (int64_t)g_queryInfo.superQueryInfo.rate*1000) {
5175 5176 5177 5178
      taosMsleep(g_queryInfo.superQueryInfo.rate*1000 - (et - st)); // ms
      //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_id, winfo->end_table_id);
    }

5179
    st = taosGetTimestampUs();
5180 5181 5182 5183 5184 5185 5186 5187 5188
    for (int i = 0; i < g_queryInfo.superQueryInfo.sqlCount; i++) {
      if (0 == strncasecmp(g_queryInfo.queryMode, "taosc", 5)) {          
        int64_t t1 = taosGetTimestampUs();
        char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
        if (g_queryInfo.superQueryInfo.result[i][0] != 0) {
          sprintf(tmpFile, "%s-%d", g_queryInfo.superQueryInfo.result[i], winfo->threadID);
        }
        selectAndGetResult(winfo->taos, g_queryInfo.superQueryInfo.sql[i], tmpFile); 
        int64_t t2 = taosGetTimestampUs();          
5189 5190
        printf("=[taosc] thread[%"PRId64"] complete one sql, Spent %f s\n", 
                taosGetSelfPthreadId(), (t2 - t1)/1000000.0);
5191 5192
      } else {
        int64_t t1 = taosGetTimestampUs();
5193 5194
        int retCode = postProceSql(g_queryInfo.host, 
                g_queryInfo.port, g_queryInfo.superQueryInfo.sql[i]);
5195
        int64_t t2 = taosGetTimestampUs();          
5196 5197
        printf("=[restful] thread[%"PRId64"] complete one sql, Spent %f s\n", 
                taosGetSelfPthreadId(), (t2 - t1)/1000000.0);
5198 5199
        
        if (0 != retCode) {
5200
          printf("====restful return fail, threadID[%d]\n", winfo->threadID);
5201 5202 5203 5204
          return NULL;
        }
      }   
    }
5205
    et = taosGetTimestampUs();
5206 5207
    printf("==thread[%"PRId64"] complete all sqls to specify tables once queries duration:%.6fs\n\n", 
            taosGetSelfPthreadId(), (double)(et - st)/1000.0);
H
Hui Li 已提交
5208
  }
5209 5210
  return NULL;
}
H
Hui Li 已提交
5211

5212
static void replaceSubTblName(char* inSql, char* outSql, int tblIndex) {
5213 5214
  char sourceString[32] = "xxxx";
  char subTblName[MAX_TB_NAME_SIZE*3];
5215 5216 5217
  sprintf(subTblName, "%s.%s", 
          g_queryInfo.dbName, 
          g_queryInfo.subQueryInfo.childTblName + tblIndex*TSDB_TABLE_NAME_LEN);
5218 5219 5220 5221 5222 5223

  //printf("inSql: %s\n", inSql);
  
  char* pos = strstr(inSql, sourceString);
  if (0 == pos) {
    return; 
H
Hui Li 已提交
5224 5225
  }
  
5226
  tstrncpy(outSql, inSql, pos - inSql + 1);
5227 5228 5229 5230 5231
  //printf("1: %s\n", outSql);
  strcat(outSql, subTblName);  
  //printf("2: %s\n", outSql);  
  strcat(outSql, pos+strlen(sourceString));  
  //printf("3: %s\n", outSql); 
H
Hui Li 已提交
5232 5233
}

5234
static void *subQueryProcess(void *sarg) {
5235 5236 5237
  char sqlstr[1024];
  threadInfo *winfo = (threadInfo *)sarg; 
  int64_t st = 0;
5238
  int64_t et = (int64_t)g_queryInfo.subQueryInfo.rate*1000;
5239
  while (1) {
5240
    if (g_queryInfo.subQueryInfo.rate && (et - st) < (int64_t)g_queryInfo.subQueryInfo.rate*1000) {
5241 5242 5243
      taosMsleep(g_queryInfo.subQueryInfo.rate*1000 - (et - st)); // ms
      //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_id, winfo->end_table_id);
    }
H
Hui Li 已提交
5244

5245
    st = taosGetTimestampUs();
5246
    for (int i = winfo->start_table_id; i <= winfo->end_table_id; i++) {
5247
      for (int j = 0; j < g_queryInfo.subQueryInfo.sqlCount; j++) {
5248
        memset(sqlstr,0,sizeof(sqlstr));
5249
        replaceSubTblName(g_queryInfo.subQueryInfo.sql[j], sqlstr, i);
5250 5251
        char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
        if (g_queryInfo.subQueryInfo.result[i][0] != 0) {
5252 5253 5254
          sprintf(tmpFile, "%s-%d", 
                  g_queryInfo.subQueryInfo.result[i], 
                  winfo->threadID);
5255 5256 5257
        }
        selectAndGetResult(winfo->taos, sqlstr, tmpFile); 
      }
H
Hui Li 已提交
5258
    }
5259
    et = taosGetTimestampUs();
5260 5261 5262 5263
    printf("####thread[%"PRId64"] complete all sqls to allocate all sub-tables[%d - %d] once queries duration:%.4fs\n\n", 
            taosGetSelfPthreadId(), 
            winfo->start_table_id, 
            winfo->end_table_id, 
5264
            (double)(et - st)/1000000.0);
5265 5266 5267 5268
  }
  return NULL;
}

5269
static int queryTestProcess() {
5270
  TAOS * taos = NULL;  
5271 5272 5273 5274 5275
  taos = taos_connect(g_queryInfo.host, 
          g_queryInfo.user, 
          g_queryInfo.password, 
          NULL, 
          g_queryInfo.port);
5276 5277 5278 5279 5280 5281
  if (taos == NULL) {
    fprintf(stderr, "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL));
    exit(-1);
  }

  if (0 != g_queryInfo.subQueryInfo.sqlCount) {
5282 5283 5284 5285
    getAllChildNameOfSuperTable(taos,
            g_queryInfo.dbName,
            g_queryInfo.subQueryInfo.sTblName,
            &g_queryInfo.subQueryInfo.childTblName,
5286
            &g_queryInfo.subQueryInfo.childTblCount);
5287 5288 5289 5290 5291 5292 5293 5294 5295 5296 5297 5298 5299 5300 5301
  }  
  
  printfQueryMeta();
  
  if (!g_args.answer_yes) {
    printf("Press enter key to continue\n\n");
    (void)getchar();
  }
  
  printfQuerySystemInfo(taos);
  
  pthread_t  *pids  = NULL;
  threadInfo *infos = NULL;
  //==== create sub threads for query from specify table
  if (g_queryInfo.superQueryInfo.sqlCount > 0 && g_queryInfo.superQueryInfo.concurrent > 0) {
5302

5303 5304 5305 5306 5307 5308 5309
    pids  = malloc(g_queryInfo.superQueryInfo.concurrent * sizeof(pthread_t));
    infos = malloc(g_queryInfo.superQueryInfo.concurrent * sizeof(threadInfo));
    if ((NULL == pids) || (NULL == infos)) {
      printf("malloc failed for create threads\n");
      taos_close(taos);
      exit(-1);
    }
5310 5311

    for (int i = 0; i < g_queryInfo.superQueryInfo.concurrent; i++) {
5312
      threadInfo *t_info = infos + i;
5313
      t_info->threadID = i;
5314 5315 5316
  
      if (0 == strncasecmp(g_queryInfo.queryMode, "taosc", 5)) {
        t_info->taos = taos;
5317

5318 5319
        char sqlStr[MAX_TB_NAME_SIZE*2];
        sprintf(sqlStr, "use %s", g_queryInfo.dbName);
5320
        verbosePrint("%s() %d sqlStr: %s\n", __func__, __LINE__, sqlStr);
5321 5322 5323 5324 5325
        (void)queryDbExec(t_info->taos, sqlStr, NO_INSERT_TYPE);
      } else {
        t_info->taos = NULL;
      }
  
5326 5327
      pthread_create(pids + i, NULL, superQueryProcess, t_info);
    }
5328 5329 5330
  }else {
    g_queryInfo.superQueryInfo.concurrent = 0;
  }
5331

5332 5333 5334
  pthread_t  *pidsOfSub  = NULL;
  threadInfo *infosOfSub = NULL;
  //==== create sub threads for query from all sub table of the super table
5335 5336
  if ((g_queryInfo.subQueryInfo.sqlCount > 0)
          && (g_queryInfo.subQueryInfo.threadCnt > 0)) {
5337 5338 5339 5340 5341 5342 5343
    pidsOfSub  = malloc(g_queryInfo.subQueryInfo.threadCnt * sizeof(pthread_t));
    infosOfSub = malloc(g_queryInfo.subQueryInfo.threadCnt * sizeof(threadInfo));
    if ((NULL == pidsOfSub) || (NULL == infosOfSub)) {
      printf("malloc failed for create threads\n");
      taos_close(taos);
      exit(-1);
    }
5344

5345 5346 5347 5348 5349 5350 5351 5352
    int ntables = g_queryInfo.subQueryInfo.childTblCount;
    int threads = g_queryInfo.subQueryInfo.threadCnt;
  
    int a = ntables / threads;
    if (a < 1) {
      threads = ntables;
      a = 1;
    }
5353
 
5354 5355 5356 5357
    int b = 0;
    if (threads != 0) {
      b = ntables % threads;
    }
5358

5359 5360 5361 5362 5363 5364 5365 5366 5367 5368 5369 5370 5371 5372 5373 5374 5375 5376 5377
    int last = 0;
    for (int i = 0; i < threads; i++) {  
      threadInfo *t_info = infosOfSub + i;
      t_info->threadID = i;
      
      t_info->start_table_id = last;
      t_info->end_table_id = i < b ? last + a : last + a - 1;
      last = t_info->end_table_id + 1;
      t_info->taos = taos;
      pthread_create(pidsOfSub + i, NULL, subQueryProcess, t_info);
    }

    g_queryInfo.subQueryInfo.threadCnt = threads;
  }else {
    g_queryInfo.subQueryInfo.threadCnt = 0;
  }  
  
  for (int i = 0; i < g_queryInfo.superQueryInfo.concurrent; i++) {
    pthread_join(pids[i], NULL);
5378
  }
S
Shuaiqiang Chang 已提交
5379

5380 5381 5382 5383 5384
  tmfree((char*)pids);
  tmfree((char*)infos);  
  
  for (int i = 0; i < g_queryInfo.subQueryInfo.threadCnt; i++) {
    pthread_join(pidsOfSub[i], NULL);
H
hzcheng 已提交
5385
  }
H
Hui Li 已提交
5386

5387 5388 5389 5390 5391 5392 5393 5394 5395 5396 5397 5398 5399 5400 5401
  tmfree((char*)pidsOfSub);
  tmfree((char*)infosOfSub);  
  
  taos_close(taos);
  return 0;
}

static void subscribe_callback(TAOS_SUB* tsub, TAOS_RES *res, void* param, int code) {  
  if (res == NULL || taos_errno(res) != 0) {
    printf("failed to subscribe result, code:%d, reason:%s\n", code, taos_errstr(res));
    return;
  }
  
  getResult(res, (char*)param);
  taos_free_result(res);
H
hzcheng 已提交
5402 5403
}

5404 5405
static TAOS_SUB* subscribeImpl(TAOS *taos, char *sql, char* topic, char* resultFileName) {
  TAOS_SUB* tsub = NULL;  
H
hzcheng 已提交
5406

5407
  if (g_queryInfo.superQueryInfo.subscribeMode) {
5408 5409 5410 5411
    tsub = taos_subscribe(taos, 
            g_queryInfo.superQueryInfo.subscribeRestart, 
            topic, sql, subscribe_callback, (void*)resultFileName, 
            g_queryInfo.superQueryInfo.subscribeInterval);
5412
  } else {
5413 5414 5415
    tsub = taos_subscribe(taos, 
            g_queryInfo.superQueryInfo.subscribeRestart, 
            topic, sql, NULL, NULL, 0);
5416
  }
5417

5418 5419 5420 5421
  if (tsub == NULL) {
    printf("failed to create subscription. topic:%s, sql:%s\n", topic, sql);
    return NULL;
  } 
5422

5423 5424
  return tsub;
}
H
hzcheng 已提交
5425

5426 5427 5428
void *subSubscribeProcess(void *sarg) {
  threadInfo *winfo = (threadInfo *)sarg; 
  char subSqlstr[1024];
H
hzcheng 已提交
5429

5430 5431
  char sqlStr[MAX_TB_NAME_SIZE*2];
  sprintf(sqlStr, "use %s", g_queryInfo.dbName);
5432
  debugPrint("%s() %d sqlStr: %s\n", __func__, __LINE__, sqlStr);
5433 5434 5435 5436 5437 5438 5439 5440 5441 5442 5443 5444 5445 5446 5447 5448 5449 5450 5451 5452 5453 5454 5455 5456 5457 5458 5459 5460
  if (0 != queryDbExec(winfo->taos, sqlStr, NO_INSERT_TYPE)){
    return NULL;
  }
  
  //int64_t st = 0;
  //int64_t et = 0;
  do {
    //if (g_queryInfo.superQueryInfo.rate && (et - st) < g_queryInfo.superQueryInfo.rate*1000) {
    //  taosMsleep(g_queryInfo.superQueryInfo.rate*1000 - (et - st)); // ms
    //  //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_id, winfo->end_table_id);
    //}

    //st = taosGetTimestampMs();
    char topic[32] = {0};
    for (int i = 0; i < g_queryInfo.subQueryInfo.sqlCount; i++) {
      sprintf(topic, "taosdemo-subscribe-%d", i);
      memset(subSqlstr,0,sizeof(subSqlstr));
      replaceSubTblName(g_queryInfo.subQueryInfo.sql[i], subSqlstr, i);
      char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
      if (g_queryInfo.subQueryInfo.result[i][0] != 0) {
        sprintf(tmpFile, "%s-%d", g_queryInfo.subQueryInfo.result[i], winfo->threadID);
      }
      g_queryInfo.subQueryInfo.tsub[i] = subscribeImpl(winfo->taos, subSqlstr, topic, tmpFile); 
      if (NULL == g_queryInfo.subQueryInfo.tsub[i]) {
        return NULL;
      }
    }
    //et = taosGetTimestampMs();
5461
    //printf("========thread[%"PRId64"] complete all sqls to super table once queries duration:%.4fs\n", taosGetSelfPthreadId(), (double)(et - st)/1000.0);
5462 5463 5464 5465 5466 5467 5468 5469 5470
  } while (0);

  // start loop to consume result
  TAOS_RES* res = NULL;
  while (1) {
    for (int i = 0; i < g_queryInfo.subQueryInfo.sqlCount; i++) {
      if (1 == g_queryInfo.subQueryInfo.subscribeMode) {
        continue;
      }
H
Hui Li 已提交
5471
      
5472 5473 5474 5475
      res = taos_consume(g_queryInfo.subQueryInfo.tsub[i]);
      if (res) {
        char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
        if (g_queryInfo.subQueryInfo.result[i][0] != 0) {
5476 5477 5478
          sprintf(tmpFile, "%s-%d", 
                  g_queryInfo.subQueryInfo.result[i], 
                  winfo->threadID);
5479 5480
        }
        getResult(res, tmpFile);
H
hzcheng 已提交
5481 5482 5483
      }
    }
  }
5484 5485 5486
  taos_free_result(res);
  
  for (int i = 0; i < g_queryInfo.subQueryInfo.sqlCount; i++) {
5487 5488
    taos_unsubscribe(g_queryInfo.subQueryInfo.tsub[i], 
            g_queryInfo.subQueryInfo.subscribeKeepProgress);
5489
  }
H
hzcheng 已提交
5490 5491 5492
  return NULL;
}

5493 5494
void *superSubscribeProcess(void *sarg) {
  threadInfo *winfo = (threadInfo *)sarg; 
H
hzcheng 已提交
5495

5496 5497
  char sqlStr[MAX_TB_NAME_SIZE*2];
  sprintf(sqlStr, "use %s", g_queryInfo.dbName);
5498
  debugPrint("%s() %d sqlStr: %s\n", __func__, __LINE__, sqlStr);
5499 5500 5501 5502 5503 5504 5505 5506 5507 5508 5509 5510 5511 5512 5513 5514 5515 5516
  if (0 != queryDbExec(winfo->taos, sqlStr, NO_INSERT_TYPE)) {
    return NULL;
  }
  
  //int64_t st = 0;
  //int64_t et = 0;
  do {
    //if (g_queryInfo.superQueryInfo.rate && (et - st) < g_queryInfo.superQueryInfo.rate*1000) {
    //  taosMsleep(g_queryInfo.superQueryInfo.rate*1000 - (et - st)); // ms
    //  //printf("========sleep duration:%"PRId64 "========inserted rows:%d, table range:%d - %d\n", (1000 - (et - st)), i, winfo->start_table_id, winfo->end_table_id);
    //}

    //st = taosGetTimestampMs();
    char topic[32] = {0};
    for (int i = 0; i < g_queryInfo.superQueryInfo.sqlCount; i++) {
      sprintf(topic, "taosdemo-subscribe-%d", i);
      char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
      if (g_queryInfo.subQueryInfo.result[i][0] != 0) {
5517 5518
        sprintf(tmpFile, "%s-%d", 
                g_queryInfo.superQueryInfo.result[i], winfo->threadID);
5519
      }
5520 5521 5522 5523
      g_queryInfo.superQueryInfo.tsub[i] = 
          subscribeImpl(winfo->taos, 
                  g_queryInfo.superQueryInfo.sql[i], 
                  topic, tmpFile); 
5524 5525 5526 5527 5528
      if (NULL == g_queryInfo.superQueryInfo.tsub[i]) {
        return NULL;
      }
    }
    //et = taosGetTimestampMs();
5529
    //printf("========thread[%"PRId64"] complete all sqls to super table once queries duration:%.4fs\n", taosGetSelfPthreadId(), (double)(et - st)/1000.0);
5530 5531 5532 5533 5534 5535 5536 5537 5538 5539 5540 5541 5542 5543
  } while (0);

  // start loop to consume result
  TAOS_RES* res = NULL;
  while (1) {
    for (int i = 0; i < g_queryInfo.superQueryInfo.sqlCount; i++) {
      if (1 == g_queryInfo.superQueryInfo.subscribeMode) {
        continue;
      }
      
      res = taos_consume(g_queryInfo.superQueryInfo.tsub[i]);
      if (res) {
        char tmpFile[MAX_FILE_NAME_LEN*2] = {0};
        if (g_queryInfo.superQueryInfo.result[i][0] != 0) {
5544 5545
          sprintf(tmpFile, "%s-%d", 
                  g_queryInfo.superQueryInfo.result[i], winfo->threadID);
5546 5547 5548 5549 5550 5551
        }
        getResult(res, tmpFile);
      }
    }
  }
  taos_free_result(res);
5552

5553
  for (int i = 0; i < g_queryInfo.superQueryInfo.sqlCount; i++) {
5554 5555
    taos_unsubscribe(g_queryInfo.superQueryInfo.tsub[i], 
            g_queryInfo.superQueryInfo.subscribeKeepProgress);
5556
  }
H
hzcheng 已提交
5557 5558 5559
  return NULL;
}

5560
static int subscribeTestProcess() {
5561
  printfQueryMeta();
H
hzcheng 已提交
5562

5563 5564 5565
  if (!g_args.answer_yes) {
    printf("Press enter key to continue\n\n");
    (void)getchar();
Y
yihaoDeng 已提交
5566
  }
5567 5568

  TAOS * taos = NULL;  
5569 5570 5571 5572 5573
  taos = taos_connect(g_queryInfo.host, 
          g_queryInfo.user, 
          g_queryInfo.password, 
          g_queryInfo.dbName, 
          g_queryInfo.port);
5574 5575 5576 5577 5578 5579
  if (taos == NULL) {
    fprintf(stderr, "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL));
    exit(-1);
  }

  if (0 != g_queryInfo.subQueryInfo.sqlCount) {
5580
    getAllChildNameOfSuperTable(taos, 
5581 5582 5583 5584
            g_queryInfo.dbName, 
            g_queryInfo.subQueryInfo.sTblName, 
            &g_queryInfo.subQueryInfo.childTblName, 
            &g_queryInfo.subQueryInfo.childTblCount);
5585 5586 5587 5588 5589
  }

  pthread_t  *pids = NULL;
  threadInfo *infos = NULL;
  //==== create sub threads for query from super table
5590 5591
  if (g_queryInfo.superQueryInfo.sqlCount > 0 
          && g_queryInfo.superQueryInfo.concurrent > 0) {
5592 5593 5594
    pids  = malloc(g_queryInfo.superQueryInfo.concurrent * sizeof(pthread_t));
    infos = malloc(g_queryInfo.superQueryInfo.concurrent * sizeof(threadInfo));
    if ((NULL == pids) || (NULL == infos)) {
5595
      printf("malloc failed for create threads\n");
5596 5597 5598
      taos_close(taos);
      exit(-1);
    }
5599 5600

    for (int i = 0; i < g_queryInfo.superQueryInfo.concurrent; i++) {
5601 5602 5603 5604 5605
      threadInfo *t_info = infos + i;
      t_info->threadID = i;
      t_info->taos = taos;
      pthread_create(pids + i, NULL, superSubscribeProcess, t_info);
    }
H
hzcheng 已提交
5606
  }
5607
 
5608 5609 5610
  //==== create sub threads for query from sub table  
  pthread_t  *pidsOfSub  = NULL;
  threadInfo *infosOfSub = NULL;
5611
  if ((g_queryInfo.subQueryInfo.sqlCount > 0)
5612
          && (g_queryInfo.subQueryInfo.threadCnt > 0)) {
5613
    pidsOfSub  = malloc(g_queryInfo.subQueryInfo.threadCnt *
5614 5615 5616
            sizeof(pthread_t));
    infosOfSub = malloc(g_queryInfo.subQueryInfo.threadCnt * 
            sizeof(threadInfo));
5617
    if ((NULL == pidsOfSub) || (NULL == infosOfSub)) {
5618
      printf("malloc failed for create threads\n");
5619 5620
      taos_close(taos);
      exit(-1);
S
Shuaiqiang Chang 已提交
5621
    }
5622

5623 5624
    int ntables = g_queryInfo.subQueryInfo.childTblCount;
    int threads = g_queryInfo.subQueryInfo.threadCnt;
5625
 
5626 5627 5628 5629 5630
    int a = ntables / threads;
    if (a < 1) {
      threads = ntables;
      a = 1;
    }
5631
 
5632 5633 5634 5635
    int b = 0;
    if (threads != 0) {
      b = ntables % threads;
    }
5636
 
5637
    int last = 0;
5638
    for (int i = 0; i < threads; i++) {
5639 5640
      threadInfo *t_info = infosOfSub + i;
      t_info->threadID = i;
5641

5642 5643 5644 5645
      t_info->start_table_id = last;
      t_info->end_table_id = i < b ? last + a : last + a - 1;
      t_info->taos = taos;
      pthread_create(pidsOfSub + i, NULL, subSubscribeProcess, t_info);
H
hzcheng 已提交
5646
    }
5647
    g_queryInfo.subQueryInfo.threadCnt = threads;
H
hzcheng 已提交
5648
  }
5649
 
5650 5651 5652
  for (int i = 0; i < g_queryInfo.superQueryInfo.concurrent; i++) {
    pthread_join(pids[i], NULL);
  }  
H
hzcheng 已提交
5653

5654
  tmfree((char*)pids);
5655
  tmfree((char*)infos);
H
hzcheng 已提交
5656

5657 5658
  for (int i = 0; i < g_queryInfo.subQueryInfo.threadCnt; i++) {
    pthread_join(pidsOfSub[i], NULL);
H
hzcheng 已提交
5659 5660
  }

5661
  tmfree((char*)pidsOfSub);
5662
  tmfree((char*)infosOfSub);
5663 5664
  taos_close(taos);
  return 0;
H
hzcheng 已提交
5665 5666
}

5667 5668 5669
void initOfInsertMeta() {
  memset(&g_Dbs, 0, sizeof(SDbs));
   
5670 5671 5672 5673 5674 5675 5676
  // set default values
  tstrncpy(g_Dbs.host, "127.0.0.1", MAX_DB_NAME_SIZE);
  g_Dbs.port = 6030;
  tstrncpy(g_Dbs.user, TSDB_DEFAULT_USER, MAX_DB_NAME_SIZE);
  tstrncpy(g_Dbs.password, TSDB_DEFAULT_PASS, MAX_DB_NAME_SIZE);
  g_Dbs.threadCount = 2;
  g_Dbs.use_metric = true;
L
Liu Tao 已提交
5677 5678
}

5679 5680 5681
void initOfQueryMeta() {
  memset(&g_queryInfo, 0, sizeof(SQueryMetaInfo));
   
5682 5683 5684 5685 5686
  // set default values
  tstrncpy(g_queryInfo.host, "127.0.0.1", MAX_DB_NAME_SIZE);
  g_queryInfo.port = 6030;
  tstrncpy(g_queryInfo.user, TSDB_DEFAULT_USER, MAX_DB_NAME_SIZE);
  tstrncpy(g_queryInfo.password, TSDB_DEFAULT_PASS, MAX_DB_NAME_SIZE);
L
Liu Tao 已提交
5687 5688
}

5689 5690 5691 5692
void setParaFromArg(){
  if (g_args.host) {
    strcpy(g_Dbs.host, g_args.host);
  } else {
5693
    tstrncpy(g_Dbs.host, "127.0.0.1", MAX_DB_NAME_SIZE);
5694
  }
L
Liu Tao 已提交
5695

5696 5697 5698 5699 5700 5701 5702
  if (g_args.user) {
    strcpy(g_Dbs.user, g_args.user);
  } 

  if (g_args.password) {
    strcpy(g_Dbs.password, g_args.password);
  } 
L
Liu Tao 已提交
5703
  
5704 5705 5706
  if (g_args.port) {
    g_Dbs.port = g_args.port;
  } 
L
Liu Tao 已提交
5707

5708 5709 5710
  g_Dbs.threadCount = g_args.num_of_threads;
  g_Dbs.threadCountByCreateTbl = g_args.num_of_threads;

5711 5712
  g_Dbs.dbCount = 1;
  g_Dbs.db[0].drop = 1;
5713

5714
  tstrncpy(g_Dbs.db[0].dbName, g_args.database, MAX_DB_NAME_SIZE);
5715
  g_Dbs.db[0].dbCfg.replica = g_args.replica;
5716
  tstrncpy(g_Dbs.db[0].dbCfg.precision, "ms", MAX_DB_NAME_SIZE);
L
Liu Tao 已提交
5717

5718
  tstrncpy(g_Dbs.resultFile, g_args.output_file, MAX_FILE_NAME_LEN);
5719 5720 5721 5722 5723

  g_Dbs.use_metric = g_args.use_metric;
  g_Dbs.insert_only = g_args.insert_only;

  g_Dbs.do_aggreFunc = true;
L
Liu Tao 已提交
5724

5725 5726 5727 5728
  char dataString[STRING_LEN];
  char **data_type = g_args.datatype;
  
  memset(dataString, 0, STRING_LEN);
L
Liu Tao 已提交
5729

5730 5731 5732
  if (strcasecmp(data_type[0], "BINARY") == 0 
          || strcasecmp(data_type[0], "BOOL") == 0 
          || strcasecmp(data_type[0], "NCHAR") == 0 ) {
5733
    g_Dbs.do_aggreFunc = false;
L
Liu Tao 已提交
5734
  }
H
hzcheng 已提交
5735

5736 5737 5738 5739 5740 5741 5742 5743 5744 5745 5746 5747 5748 5749 5750 5751 5752 5753 5754
  if (g_args.use_metric) {
    g_Dbs.db[0].superTblCount = 1;
    tstrncpy(g_Dbs.db[0].superTbls[0].sTblName, "meters", MAX_TB_NAME_SIZE);
    g_Dbs.db[0].superTbls[0].childTblCount = g_args.num_of_tables;
    g_Dbs.threadCount = g_args.num_of_threads;
    g_Dbs.threadCountByCreateTbl = 1;
    g_Dbs.queryMode = g_args.mode;
  
    g_Dbs.db[0].superTbls[0].autoCreateTable = PRE_CREATE_SUBTBL;
    g_Dbs.db[0].superTbls[0].superTblExists = TBL_NO_EXISTS;
    g_Dbs.db[0].superTbls[0].childTblExists = TBL_NO_EXISTS;
    g_Dbs.db[0].superTbls[0].disorderRange = g_args.disorderRange;
    g_Dbs.db[0].superTbls[0].disorderRatio = g_args.disorderRatio;
    tstrncpy(g_Dbs.db[0].superTbls[0].childTblPrefix, 
            g_args.tb_prefix, MAX_TB_NAME_SIZE);
    tstrncpy(g_Dbs.db[0].superTbls[0].dataSource, "rand", MAX_TB_NAME_SIZE);
    tstrncpy(g_Dbs.db[0].superTbls[0].insertMode, "taosc", MAX_TB_NAME_SIZE);
    tstrncpy(g_Dbs.db[0].superTbls[0].startTimestamp, 
            "2017-07-14 10:40:00.000", MAX_TB_NAME_SIZE);
5755
    g_Dbs.db[0].superTbls[0].timeStampStep = DEFAULT_TIMESTAMP_STEP;
5756
  
5757
    g_Dbs.db[0].superTbls[0].insertRows = g_args.num_of_DPT;
5758
    g_Dbs.db[0].superTbls[0].maxSqlLen = TSDB_PAYLOAD_SIZE;
5759

5760 5761 5762 5763 5764 5765 5766 5767 5768
    g_Dbs.db[0].superTbls[0].columnCount = 0;
    for (int i = 0; i < MAX_NUM_DATATYPE; i++) {
      if (data_type[i] == NULL) {
        break;
      }
  
      tstrncpy(g_Dbs.db[0].superTbls[0].columns[i].dataType, 
              data_type[i], MAX_TB_NAME_SIZE);
      g_Dbs.db[0].superTbls[0].columns[i].dataLen = g_args.len_of_binary;    
5769 5770
      g_Dbs.db[0].superTbls[0].columnCount++;
    }
5771 5772 5773 5774 5775 5776 5777 5778 5779 5780
  
    if (g_Dbs.db[0].superTbls[0].columnCount > g_args.num_of_CPR) {
      g_Dbs.db[0].superTbls[0].columnCount = g_args.num_of_CPR;
    } else {
      for (int i = g_Dbs.db[0].superTbls[0].columnCount; i < g_args.num_of_CPR; i++) {
        tstrncpy(g_Dbs.db[0].superTbls[0].columns[i].dataType, "INT", MAX_TB_NAME_SIZE);
        g_Dbs.db[0].superTbls[0].columns[i].dataLen = 0;    
        g_Dbs.db[0].superTbls[0].columnCount++;
      }
    }
5781

5782
    tstrncpy(g_Dbs.db[0].superTbls[0].tags[0].dataType, "INT", MAX_TB_NAME_SIZE);
5783 5784
    g_Dbs.db[0].superTbls[0].tags[0].dataLen = 0;    
  
5785
    tstrncpy(g_Dbs.db[0].superTbls[0].tags[1].dataType, "BINARY", MAX_TB_NAME_SIZE);
5786 5787 5788
    g_Dbs.db[0].superTbls[0].tags[1].dataLen = g_args.len_of_binary;    
    g_Dbs.db[0].superTbls[0].tagCount = 2;  
  } else {
5789
    g_Dbs.threadCountByCreateTbl = 1;
5790
    g_Dbs.db[0].superTbls[0].tagCount = 0; 
H
hzcheng 已提交
5791
  }
5792

5793 5794 5795 5796 5797 5798
}

/* Function to do regular expression check */
static int regexMatch(const char *s, const char *reg, int cflags) {
  regex_t regex;
  char    msgbuf[100] = {0};
H
hzcheng 已提交
5799

5800 5801 5802
  /* Compile regular expression */
  if (regcomp(&regex, reg, cflags) != 0) {
    printf("Fail to compile regex\n");
H
Hui Li 已提交
5803 5804 5805
    exit(-1);
  }

5806 5807 5808 5809 5810 5811 5812 5813 5814 5815 5816 5817 5818 5819 5820 5821 5822 5823 5824 5825 5826 5827 5828 5829 5830 5831 5832 5833 5834 5835 5836
  /* Execute regular expression */
  int reti = regexec(&regex, s, 0, NULL, 0);
  if (!reti) {
    regfree(&regex);
    return 1;
  } else if (reti == REG_NOMATCH) {
    regfree(&regex);
    return 0;
  } else {
    regerror(reti, &regex, msgbuf, sizeof(msgbuf));
    printf("Regex match failed: %s\n", msgbuf);
    regfree(&regex);
    exit(-1);
  }

  return 0;
}

static int isCommentLine(char *line) {
  if (line == NULL) return 1;

  return regexMatch(line, "^\\s*#.*", REG_EXTENDED);
}

void querySqlFile(TAOS* taos, char* sqlFile)
{
  FILE *fp = fopen(sqlFile, "r");
  if (fp == NULL) {
    printf("failed to open file %s, reason:%s\n", sqlFile, strerror(errno));
    return;
  }
5837

5838 5839 5840 5841 5842 5843 5844
  int       read_len = 0;
  char *    cmd = calloc(1, MAX_SQL_SIZE);
  size_t    cmd_len = 0;
  char *    line = NULL;
  size_t    line_len = 0;

  double t = getCurrentTime();
5845

5846 5847 5848 5849 5850 5851 5852 5853 5854 5855 5856 5857 5858
  while ((read_len = tgetline(&line, &line_len, fp)) != -1) {
    if (read_len >= MAX_SQL_SIZE) continue;
    line[--read_len] = '\0';

    if (read_len == 0 || isCommentLine(line)) {  // line starts with #
      continue;
    }

    if (line[read_len - 1] == '\\') {
      line[read_len - 1] = ' ';
      memcpy(cmd + cmd_len, line, read_len);
      cmd_len += read_len;
      continue;
5859
    }
5860 5861

    memcpy(cmd + cmd_len, line, read_len);
5862
    verbosePrint("%s() LN%d cmd: %s\n", __func__, __LINE__, cmd);
5863
    queryDbExec(taos, cmd, NO_INSERT_TYPE);
5864 5865 5866 5867 5868 5869 5870
    if (0 != queryDbExec(taos, cmd, NO_INSERT_TYPE)) {
        printf("queryDbExec %s failed!\n", cmd);
        tmfree(cmd);
        tmfree(line);
        tmfclose(fp);
        return;
    }
5871 5872
    memset(cmd, 0, MAX_SQL_SIZE);
    cmd_len = 0;
H
hzcheng 已提交
5873 5874
  }

5875 5876
  t = getCurrentTime() - t;
  printf("run %s took %.6f second(s)\n\n", sqlFile, t);
5877

5878 5879 5880 5881
  tmfree(cmd);
  tmfree(line);
  tmfclose(fp);
  return;
H
hzcheng 已提交
5882 5883
}

5884
static void testMetaFile() {
5885
    if (INSERT_MODE == g_args.test_mode) {
5886
      if (g_Dbs.cfgDir[0]) taos_options(TSDB_OPTION_CONFIGDIR, g_Dbs.cfgDir);
5887
      insertTestProcess();
5888
    } else if (QUERY_MODE == g_args.test_mode) {
5889 5890
      if (g_queryInfo.cfgDir[0])
          taos_options(TSDB_OPTION_CONFIGDIR, g_queryInfo.cfgDir);
5891
      queryTestProcess();
5892
    } else if (SUBSCRIBE_MODE == g_args.test_mode) {
5893 5894
      if (g_queryInfo.cfgDir[0])
          taos_options(TSDB_OPTION_CONFIGDIR, g_queryInfo.cfgDir);
5895
      subscribeTestProcess();
5896 5897 5898
    }  else {
      ;
    }
5899
}
5900

5901
static void queryResult() {
5902 5903 5904
    // select
    if (false == g_Dbs.insert_only) {
      // query data
5905

5906 5907 5908 5909
      pthread_t read_id;
      threadInfo *rInfo = malloc(sizeof(threadInfo));
      rInfo->start_time = 1500000000000;  // 2017-07-14 10:40:00.000
      rInfo->start_table_id = 0;
5910

5911
      //rInfo->do_aggreFunc = g_Dbs.do_aggreFunc;
5912 5913 5914 5915 5916 5917 5918 5919 5920 5921
      if (g_args.use_metric) {
        rInfo->end_table_id = g_Dbs.db[0].superTbls[0].childTblCount - 1;
        rInfo->superTblInfo = &g_Dbs.db[0].superTbls[0];
        strcpy(rInfo->tb_prefix, 
              g_Dbs.db[0].superTbls[0].childTblPrefix);
      } else {
        rInfo->end_table_id = g_args.num_of_tables -1;
        strcpy(rInfo->tb_prefix, g_args.tb_prefix);
      }

5922 5923 5924 5925 5926 5927
      rInfo->taos = taos_connect(
              g_Dbs.host, 
              g_Dbs.user, 
              g_Dbs.password, 
              g_Dbs.db[0].dbName, 
              g_Dbs.port);
5928 5929 5930 5931 5932 5933
      if (rInfo->taos == NULL) {
        fprintf(stderr, "Failed to connect to TDengine, reason:%s\n", taos_errstr(NULL));
        free(rInfo);
        exit(-1);
      }

5934 5935 5936 5937 5938 5939 5940 5941 5942 5943
      strcpy(rInfo->fp, g_Dbs.resultFile);

      if (!g_Dbs.use_metric) {
        pthread_create(&read_id, NULL, readTable, rInfo);
      } else {
        pthread_create(&read_id, NULL, readMetric, rInfo);
      }
      pthread_join(read_id, NULL);
      taos_close(rInfo->taos);
      free(rInfo);
H
hzcheng 已提交
5944
    }
5945 5946
}

5947 5948 5949 5950 5951 5952 5953 5954 5955 5956 5957
static void testCmdLine() {

    g_args.test_mode = INSERT_MODE;
    insertTestProcess();

    if (g_Dbs.insert_only)
      return;
    else
        queryResult();
}

5958 5959 5960
int main(int argc, char *argv[]) {
  parse_args(argc, argv, &g_args);

5961
  debugPrint("meta file: %s\n", g_args.metaFile);
5962 5963 5964 5965 5966 5967 5968 5969 5970 5971 5972 5973 5974 5975 5976 5977 5978 5979 5980 5981 5982 5983 5984 5985 5986 5987 5988 5989

  if (g_args.metaFile) {
    initOfInsertMeta();
    initOfQueryMeta();    

    if (false == getInfoFromJsonFile(g_args.metaFile)) {
      printf("Failed to read %s\n", g_args.metaFile);
      return 1;
    }

    testMetaFile();
  } else {
    memset(&g_Dbs, 0, sizeof(SDbs));
    setParaFromArg();

    if (NULL != g_args.sqlFile) {
      TAOS* qtaos = taos_connect(
          g_Dbs.host, 
          g_Dbs.user, 
          g_Dbs.password, 
          g_Dbs.db[0].dbName, 
          g_Dbs.port);
      querySqlFile(qtaos, g_args.sqlFile);  
      taos_close(qtaos);

    } else {
      testCmdLine();
    }
H
hzcheng 已提交
5990
  }
5991 5992

  return 0;
H
hzcheng 已提交
5993
}
5994