Merge pull request #9198 from taosdata/szhou/hotfix/never-ending-schemaless

TD-12302: Add tool to profile sql performance and compare with schemaless

tests/examples/c/CMakeLists.txt

@@ -7,6 +7,8 @@ IF (TD_LINUX)
   TARGET_LINK_LIBRARIES(demo taos_static trpc tutil pthread )
   ADD_EXECUTABLE(sml schemaless.c)
   TARGET_LINK_LIBRARIES(sml taos_static trpc tutil pthread )
+  ADD_EXECUTABLE(sqlperf sqlperf.c)
+  TARGET_LINK_LIBRARIES(sqlperf taos_static trpc tutil pthread )
   ADD_EXECUTABLE(subscribe subscribe.c)
   TARGET_LINK_LIBRARIES(subscribe taos_static trpc tutil pthread )
   ADD_EXECUTABLE(epoll epoll.c)

tests/examples/c/schemaless.c

@@ -105,6 +105,35 @@ int32_t generateLine(char* line, int lineLen, char* lineTemplate, int protocol,
   return TSDB_CODE_SUCCESS;
 }
 
+int32_t setupSuperTables(TAOS* taos, char* lineTemplate, int protocol,
+                         int numSuperTables, int numChildTables, int numRowsPerChildTable,
+                         int maxBatchesPerThread, int64_t ts) {
+  printf("setup supertables...");
+  {
+    char** linesStb = calloc(numSuperTables, sizeof(char*));
+    for (int i = 0; i < numSuperTables; i++) {
+      char* lineStb = calloc(strlen(lineTemplate)+128, 1);
+      generateLine(lineStb, strlen(lineTemplate)+128, lineTemplate, protocol, i,
+                   numSuperTables * numChildTables,
+                   ts + numSuperTables * numChildTables * numRowsPerChildTable);
+      linesStb[i] = lineStb;
+    }
+    SThreadInsertArgs args = {0};
+    args.protocol = protocol;
+    args.batches = calloc(maxBatchesPerThread, sizeof(maxBatchesPerThread));
+    args.taos = taos;
+    args.batches[0].lines = linesStb;
+    args.batches[0].numLines = numSuperTables;
+    insertLines(&args);
+    free(args.batches);
+    for (int i = 0; i < numSuperTables; ++i) {
+      free(linesStb[i]);
+    }
+    free(linesStb);
+  }
+  return TSDB_CODE_SUCCESS;
+}
+
 int main(int argc, char* argv[]) {
   int numThreads = 8;
   int maxBatchesPerThread = 1024;	
@@ -117,9 +146,10 @@ int main(int argc, char* argv[]) {
   int maxLinesPerBatch = 16384;
 
   int protocol = TSDB_SML_TELNET_PROTOCOL;
+  int assembleSTables = 0;
 
   int opt;
-  while ((opt = getopt(argc, argv, "s:c:r:f:t:b:p:hv")) != -1) {
+  while ((opt = getopt(argc, argv, "s:c:r:f:t:b:p:w:hv")) != -1) {
     switch (opt) {
       case 's':
         numSuperTables = atoi(optarg);
@@ -142,6 +172,9 @@ int main(int argc, char* argv[]) {
       case 'v':
         verbose = true;
         break;
+      case 'a':
+        assembleSTables = atoi(optarg);
+        break;
       case 'p':
         if (optarg[0] == 't') {
           protocol = TSDB_SML_TELNET_PROTOCOL;
@@ -152,11 +185,11 @@ int main(int argc, char* argv[]) {
         }
         break;
       case 'h':
-        fprintf(stderr, "Usage: %s -s supertable -c childtable -r rows -f fields -t threads -b maxlines_per_batch -p [t|l|j] -v\n",
+        fprintf(stderr, "Usage: %s -s supertable -c childtable -r rows -f fields -t threads -b maxlines_per_batch -p [t|l|j] -a assemble-stables -v\n",
                 argv[0]);
         exit(0);
       default: /* '?' */
-        fprintf(stderr, "Usage: %s -s supertable -c childtable -r rows -f fields -t threads -b maxlines_per_batch -p [t|l|j] -v\n",
+        fprintf(stderr, "Usage: %s -s supertable -c childtable -r rows -f fields -t threads -b maxlines_per_batch -p [t|l|j] -a assemble-stables -v\n",
                 argv[0]);
         exit(-1);
     }
@@ -200,28 +233,9 @@ int main(int argc, char* argv[]) {
     getTelenetTemplate(lineTemplate, 65535);
   }
 
-  printf("setup supertables...");
-  {
-    char** linesStb = calloc(numSuperTables, sizeof(char*));
-    for (int i = 0; i < numSuperTables; i++) {
-      char* lineStb = calloc(strlen(lineTemplate)+128, 1);
-      generateLine(lineStb, strlen(lineTemplate)+128, lineTemplate, protocol, i,
-               numSuperTables * numChildTables,
-               ts + numSuperTables * numChildTables * numRowsPerChildTable);
-      linesStb[i] = lineStb;
-    }
-    SThreadInsertArgs args = {0};
-    args.protocol = protocol;
-    args.batches = calloc(maxBatchesPerThread, sizeof(maxBatchesPerThread));
-    args.taos = taos;
-    args.batches[0].lines = linesStb;
-    args.batches[0].numLines = numSuperTables;
-    insertLines(&args);
-    free(args.batches);
-    for (int i = 0; i < numSuperTables; ++i) {
-      free(linesStb[i]);
-    }
-    free(linesStb);
+  if (assembleSTables) {
+    setupSuperTables(taos, lineTemplate, protocol,
+                     numSuperTables, numChildTables, numRowsPerChildTable, maxBatchesPerThread, ts);
   }
 
   printf("generate lines...\n");

tests/examples/c/sqlperf.c

+#include <inttypes.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include "taos.h"  // TAOS header file
+#include "taoserror.h"
+#include "os.h"
+
+bool verbose = false;
+bool describeTableFirst = false;
+
+typedef struct{
+  TAOS* taos;
+  int numThreads;
+  int threadId;
+  int numSTables;
+} SThreadArgs;
+
+static int executeSql(TAOS *taos, char *command) {
+  if (verbose) {
+    printf("sql: %s\n", command);
+  }
+  TAOS_RES *pSql = NULL;
+  int32_t   code = TSDB_CODE_SUCCESS;
+
+  pSql = taos_query(taos, command);
+  code = taos_errno(pSql);
+
+  if (code != 0) {
+    if (verbose) fprintf(stderr, "Failed to run %s, reason: %s\n", command, taos_errstr(pSql));
+    taos_free_result(pSql);
+    return code;
+  }
+
+  taos_free_result(pSql);
+  return 0;
+}
+
+void* threadFunc(void* args) {
+  char* sqlDescribeSTable = "describe st%d";
+  char* sqlCreateSTable = "create table st%d (ts timestamp, value double) "
+      "tags(t0 nchar(20), t1 nchar(20), t2 nchar(20), t3 nchar(20), t4 nchar(20), "
+      "t5 nchar(20), t6 nchar(20), t7 nchar(20), t8 nchar(20), t9 nchar(20))";
+
+  char* sqlInsertData = "insert into t%d using st%d tags('t%d', 't%d', 't%d', 't%d', 't%d', 't%d', 't%d', 't%d', 't%d', 't%d') values(%lld, %d.%d)";
+
+  SThreadArgs* param = args;
+
+  int interval = param->numSTables/param->numThreads;
+  if (param->numSTables % param->numThreads != 0) {
+    ++interval;
+  }
+  int start = param->threadId*interval;
+  int end = (param->threadId+1)*interval > param->numSTables ? param->numSTables :  (param->threadId+1)*interval;
+  int     r = rand();
+
+  for (int i = start; i < end; ++i) {
+    int tableId = i;
+    char sql0[1024] = {0};
+    char sql1[1024] = {0};
+    char sql2[1024] = {0};
+    sprintf(sql0, sqlDescribeSTable, tableId);
+    sprintf(sql1, sqlCreateSTable, tableId);
+    time_t  ct = time(0);
+    int64_t ts = ct * 1000;
+    sprintf(sql2, sqlInsertData, tableId, tableId, r, r, r, r, r, r, r, r, r, r, ts + tableId, r, r);
+
+    if (describeTableFirst) {
+      executeSql(param->taos, sql0);
+    }
+
+    executeSql(param->taos, sql1);
+
+    executeSql(param->taos, sql2);
+  }
+  return NULL;
+}
+
+
+int main(int argc, char *argv[]) {
+  int numSTables = 20000;
+  int numThreads = 32;
+
+  int opt;
+  while ((opt = getopt(argc, argv, "s:t:fvh")) != -1) {
+    switch (opt) {
+      case 's':
+        numSTables = atoi(optarg);
+        break;
+      case 't':
+        numThreads = atoi(optarg);
+        break;
+      case 'f':
+        describeTableFirst = true;
+        break;
+      case 'v':
+        verbose = true;
+        break;	
+      case 'h':
+        fprintf(stderr, "Usage: %s -s supertable -t thread -FirstDescribeSTable -Verbose\n", argv[0]);
+        exit(0);
+      default:
+        fprintf(stderr, "Usage: %s -s supertable -t thread -FirstDescribeSTable -Verbose\n", argv[0]);
+        exit(-1);
+    }
+  }
+
+
+  // connect to server
+  TAOS *taos = taos_connect(NULL, "root", "taosdata", NULL, 0);
+  if (taos == NULL) {
+    printf("failed to connect to server, reason:%s\n", "null taos" /*taos_errstr(taos)*/);
+    exit(1);
+  }
+  executeSql(taos, "drop database if exists sqlsml");
+  executeSql(taos, "create database sqlsml");
+  executeSql(taos, "use sqlsml");
+  pthread_t* tids = calloc(numThreads, sizeof(pthread_t));
+  SThreadArgs* threadArgs = calloc(numThreads, sizeof(SThreadArgs));
+  for (int i = 0; i < numThreads; ++i) {
+    threadArgs[i].numSTables = numSTables;
+    threadArgs[i].numThreads = numThreads;
+    threadArgs[i].threadId = i;
+    threadArgs[i].taos = taos;
+  }
+
+  int64_t begin = taosGetTimestampUs();
+  for (int i = 0; i < numThreads; ++i) {
+    pthread_create(tids+i, NULL, threadFunc, threadArgs+i);
+  }
+  for (int i = 0; i < numThreads; ++i) {
+    pthread_join(tids[i], NULL);
+  }
+  int64_t end = taosGetTimestampUs();
+  printf("TIME: %d(ms)\n", (int)((end-begin)/1000));
+  printf("THROUGHPUT: %d\n", (int)((numSTables * 1e6) / (end-begin)));
+  free(threadArgs);
+  free(tids);
+  taos_close(taos);
+  taos_cleanup();
+}