importPerTabe.c 11.0 KB
Newer Older
S
slguan 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18
/*
 * 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/>.
 */

#define _DEFAULT_SOURCE
#include "os.h"
#include "taos.h"
S
slguan 已提交
19
#include "tulog.h"
S
slguan 已提交
20 21
#include "ttimer.h"
#include "tutil.h"
S
slguan 已提交
22
#include "tglobal.h"
S
slguan 已提交
23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63

#define MAX_RANDOM_POINTS 20000
#define GREEN "\033[1;32m"
#define NC "\033[0m"

typedef struct {
  int64_t   rowsPerTable;
  int64_t   pointsPerTable;
  int64_t   tableBeginIndex;
  int64_t   tableEndIndex;
  int       threadIndex;
  char      dbName[32];
  char      stableName[64];
  pthread_t thread;
} SInfo;

void *syncTest(void *param);
void  generateRandomPoints();
void  shellParseArgument(int argc, char *argv[]);
void  createDbAndTable();
void  insertData();

int32_t randomData[MAX_RANDOM_POINTS];
int64_t rowsPerTable = 10000;
int64_t pointsPerTable = 1;
int64_t numOfThreads = 1;
int64_t numOfTablesPerThread = 1;
char    dbName[32] = "db";
char    stableName[64] = "st";
int32_t cache = 16384;
int32_t tables = 1000;

int main(int argc, char *argv[]) {
  shellParseArgument(argc, argv);
  generateRandomPoints();
  taos_init();
  createDbAndTable();
  insertData();
}

void createDbAndTable() {
S
slguan 已提交
64
  pPrint("start to create table");
S
slguan 已提交
65 66 67 68 69 70 71 72

  TAOS *         con;
  struct timeval systemTime;
  int64_t        st, et;
  char           qstr[64000];

  con = taos_connect(tsMasterIp, tsDefaultUser, tsDefaultPass, NULL, 0);
  if (con == NULL) {
S
slguan 已提交
73
    pError("failed to connect to DB, reason:%s", taos_errstr(con));
S
slguan 已提交
74 75 76 77 78
    exit(1);
  }

  sprintf(qstr, "create database if not exists %s cache %d tables %d", dbName, cache, tables);
  if (taos_query(con, qstr)) {
S
slguan 已提交
79
    pError("failed to create database:%s, code:%d reason:%s", dbName, taos_errno(con), taos_errstr(con));
S
slguan 已提交
80 81 82 83 84
    exit(0);
  }

  sprintf(qstr, "use %s", dbName);
  if (taos_query(con, qstr)) {
S
slguan 已提交
85
    pError("failed to use db, code:%d reason:%s", taos_errno(con), taos_errstr(con));
S
slguan 已提交
86 87 88 89 90 91 92 93 94 95 96 97 98 99 100
    exit(0);
  }

  gettimeofday(&systemTime, NULL);
  st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
  int64_t totalTables = numOfTablesPerThread * numOfThreads;
    
  if (strcmp(stableName, "no") != 0) {
    int len = sprintf(qstr, "create table if not exists %s(ts timestamp", stableName);
    for (int64_t f = 0; f < pointsPerTable; ++f) {
      len += sprintf(qstr + len, ", f%ld double", f);
    }
    sprintf(qstr + len, ") tags(t int)");

    if (taos_query(con, qstr)) {
S
slguan 已提交
101
      pError("failed to create stable, code:%d reason:%s", taos_errno(con), taos_errstr(con));
S
slguan 已提交
102 103 104 105 106 107
      exit(0);
    }
    
    for (int64_t t = 0; t < totalTables; ++t) {
      sprintf(qstr, "create table if not exists %s%ld using %s tags(%ld)", stableName, t, stableName, t);
      if (taos_query(con, qstr)) {
S
slguan 已提交
108
        pError("failed to create table %s%d, reason:%s", stableName, t, taos_errstr(con));
S
slguan 已提交
109 110 111 112 113 114 115 116 117 118 119 120
        exit(0);
      }
    }
  } else {
    for (int64_t t = 0; t < totalTables; ++t) {
      int len = sprintf(qstr, "create table if not exists %s%ld(ts timestamp", stableName, t);
      for (int64_t f = 0; f < pointsPerTable; ++f) {
        len += sprintf(qstr + len, ", f%ld double", f);
      }
      sprintf(qstr + len, ")");

      if (taos_query(con, qstr)) {
S
slguan 已提交
121
        pError("failed to create table %s%ld, reason:%s", stableName, t, taos_errstr(con));
S
slguan 已提交
122 123 124 125 126 127 128
        exit(0);
      }
    }
  }

  gettimeofday(&systemTime, NULL);
  et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
S
slguan 已提交
129
  pPrint("%.1f seconds to create %ld tables", (et - st) / 1000.0 / 1000.0, totalTables);
S
slguan 已提交
130 131 132 133 134 135 136 137 138
}

void insertData() {
  struct timeval systemTime;
  int64_t        st, et;

  gettimeofday(&systemTime, NULL);
  st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;

S
slguan 已提交
139
  pPrint("%d threads are spawned to import data", numOfThreads);
S
slguan 已提交
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 167 168 169 170 171 172

  pthread_attr_t thattr;
  pthread_attr_init(&thattr);
  pthread_attr_setdetachstate(&thattr, PTHREAD_CREATE_JOINABLE);
  SInfo *pInfo = (SInfo *)malloc(sizeof(SInfo) * numOfThreads);

  // Start threads to write
  for (int i = 0; i < numOfThreads; ++i) {
    pInfo[i].rowsPerTable = rowsPerTable;
    pInfo[i].pointsPerTable = pointsPerTable;
    pInfo[i].tableBeginIndex = i * numOfTablesPerThread;
    pInfo[i].tableEndIndex = (i + 1) * numOfTablesPerThread;
    pInfo[i].threadIndex = i;
    strcpy(pInfo[i].dbName, dbName);
    strcpy(pInfo[i].stableName, stableName);
    pthread_create(&(pInfo[i].thread), &thattr, syncTest, (void *)(pInfo + i));
  }

  taosMsleep(300);
  for (int i = 0; i < numOfThreads; i++) {
    pthread_join(pInfo[i].thread, NULL);
  }

  gettimeofday(&systemTime, NULL);
  et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
  double seconds = (et - st) / 1000.0 / 1000.0;

  int64_t totalTables = numOfTablesPerThread * numOfThreads;
  int64_t totalRows = totalTables * rowsPerTable;
  int64_t totalPoints = totalTables * rowsPerTable * pointsPerTable;
  double  speedOfRows = totalRows / seconds;
  double  speedOfPoints = totalPoints / seconds;

S
slguan 已提交
173
  pPrint(
S
slguan 已提交
174 175 176 177
      "%sall threads:%ld finished, use %.1lf seconds, tables:%.ld rows:%ld points:%ld, speed RowsPerSecond:%.1lf "
      "PointsPerSecond:%.1lf%s",
      GREEN, numOfThreads, seconds, totalTables, totalRows, totalPoints, speedOfRows, speedOfPoints, NC);

S
slguan 已提交
178
  pPrint("threads exit");
S
slguan 已提交
179 180 181 182 183 184 185 186 187 188 189 190 191

  pthread_attr_destroy(&thattr);
  free(pInfo);
}

void *syncTest(void *param) {
  TAOS *         con;
  SInfo *        pInfo = (SInfo *)param;
  struct timeval systemTime;
  int64_t        st, et;
  char           qstr[65000];
  int            maxBytes = 60000;

S
slguan 已提交
192
  pPrint("thread:%d, start to run", pInfo->threadIndex);
S
slguan 已提交
193 194 195

  con = taos_connect(tsMasterIp, tsDefaultUser, tsDefaultPass, NULL, 0);
  if (con == NULL) {
S
slguan 已提交
196
    pError("index:%d, failed to connect to DB, reason:%s", pInfo->threadIndex, taos_errstr(con));
S
slguan 已提交
197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223
    exit(1);
  }

  sprintf(qstr, "use %s", pInfo->dbName);
  taos_query(con, qstr);

  gettimeofday(&systemTime, NULL);
  st = systemTime.tv_sec * 1000000 + systemTime.tv_usec;

  int64_t start = 1587225600000;
  int64_t interval = 1000;  // 1000 ms

  char *sql = qstr;
  char  inserStr[] = "import into";
  int   len = sprintf(sql, "%s", inserStr);

  for (int64_t table = pInfo->tableBeginIndex; table < pInfo->tableEndIndex; ++table) {
    len += sprintf(sql + len, " %s%ld values", pInfo->stableName, table);
    for (int64_t row = 0; row < pInfo->rowsPerTable; row++) {
      len += sprintf(sql + len, "(%ld", start - row * interval);
      for (int64_t point = 0; point < pInfo->pointsPerTable; ++point) {
        len += sprintf(sql + len, ",%d", randomData[(123 * table + 456 * row + 789 * point) % MAX_RANDOM_POINTS]);
        // len += sprintf(sql + len, ",%ld", row);
      }
      len += sprintf(sql + len, ")");
      if (len > maxBytes) {
        if (taos_query(con, qstr)) {
S
slguan 已提交
224
          pError("thread:%d, failed to import table:%s%ld row:%ld, reason:%s", pInfo->threadIndex, pInfo->stableName,
S
slguan 已提交
225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247
                 table, row, taos_errstr(con));
        }

        // "insert into"
        len = sprintf(sql, "%s", inserStr);

        // "insert into st1 values"
        if (row != pInfo->rowsPerTable - 1) {
          len += sprintf(sql + len, " %s%ld values", pInfo->stableName, table);
        }
      }
    }
  }

  if (len != strlen(inserStr)) {
    taos_query(con, qstr);
  }

  gettimeofday(&systemTime, NULL);
  et = systemTime.tv_sec * 1000000 + systemTime.tv_usec;
  int64_t totalTables = pInfo->tableEndIndex - pInfo->tableBeginIndex;
  int64_t totalRows = totalTables * pInfo->rowsPerTable;
  int64_t totalPoints = totalRows * pInfo->pointsPerTable;
S
slguan 已提交
248
  pPrint("thread:%d, import finished, use %.2f seconds, tables:%ld rows:%ld points:%ld", pInfo->threadIndex,
S
slguan 已提交
249 250 251 252 253 254 255 256 257 258 259 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 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312
         (et - st) / 1000.0 / 1000.0, totalTables, totalRows, totalPoints);

  return NULL;
}

void generateRandomPoints() {
  for (int r = 0; r < MAX_RANDOM_POINTS; ++r) {
    randomData[r] = rand() % 1000;
  }
}

void printHelp() {
  char indent[10] = "        ";
  printf("Used to test the performance of TDengine\n After writing all the data in one table, start the next table\n");

  printf("%s%s\n", indent, "-d");
  printf("%s%s%s%s\n", indent, indent, "The name of the database to be created, default is ", dbName);
  printf("%s%s\n", indent, "-s");
  printf("%s%s%s%s%s\n", indent, indent, "The name of the super table to be created, default is ", stableName, ", if 'no' then create normal table");
  printf("%s%s\n", indent, "-c");
  printf("%s%s%s%s\n", indent, indent, "Configuration directory, default is ", configDir);
  printf("%s%s\n", indent, "-r");
  printf("%s%s%s%ld\n", indent, indent, "Number of records to write to each table, default is ", rowsPerTable);
  printf("%s%s\n", indent, "-p");
  printf("%s%s%s%" PRId64 "\n", indent, indent, "Number of columns per table, default is ", pointsPerTable);
  printf("%s%s\n", indent, "-t");
  printf("%s%s%s%" PRId64 "\n", indent, indent, "Number of threads to be used, default is ", numOfThreads);
  printf("%s%s\n", indent, "-n");
  printf("%s%s%s%" PRId64 "\n", indent, indent, "Number of tables per thread, default is ", numOfTablesPerThread);
  printf("%s%s\n", indent, "-tables");
  printf("%s%s%s%d\n", indent, indent, "Database parameters tables, default is ", tables);
  printf("%s%s\n", indent, "-cache");
  printf("%s%s%s%d\n", indent, indent, "Database parameters cache, default is ", cache);

  exit(EXIT_SUCCESS);
}

void shellParseArgument(int argc, char *argv[]) {
  for (int i = 1; i < argc; i++) {
    if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
      printHelp();
      exit(0);
    } else if (strcmp(argv[i], "-d") == 0) {
      strcpy(dbName, argv[++i]);
    } else if (strcmp(argv[i], "-c") == 0) {
      strcpy(configDir, argv[++i]);
    } else if (strcmp(argv[i], "-s") == 0) {
      strcpy(stableName, argv[++i]);
    } else if (strcmp(argv[i], "-r") == 0) {
      rowsPerTable = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-p") == 0) {
      pointsPerTable = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-t") == 0) {
      numOfThreads = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-n") == 0) {
      numOfTablesPerThread = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-tables") == 0) {
      tables = atoi(argv[++i]);
    } else if (strcmp(argv[i], "-cache") == 0) {
      cache = atoi(argv[++i]);
    } else {
    }
  }

S
slguan 已提交
313 314 315 316 317 318 319 320 321
  pPrint("%srowsPerTable:%" PRId64 "%s", GREEN, rowsPerTable, NC);
  pPrint("%spointsPerTable:%" PRId64 "%s", GREEN, pointsPerTable, NC);
  pPrint("%snumOfThreads:%" PRId64 "%s", GREEN, numOfThreads, NC);
  pPrint("%snumOfTablesPerThread:%" PRId64 "%s", GREEN, numOfTablesPerThread, NC);
  pPrint("%scache:%" PRId64 "%s", GREEN, cache, NC);
  pPrint("%stables:%" PRId64 "%s", GREEN, tables, NC);
  pPrint("%sdbName:%s%s", GREEN, dbName, NC);
  pPrint("%stableName:%s%s", GREEN, stableName, NC);
  pPrint("%sstart to run%s", GREEN, NC);
S
slguan 已提交
322
}