add metadata performance test

tests/pytest/query/queryMetaPerformace.py

+
+###################################################################
+#           Copyright (c) 2016 by TAOS Technologies, Inc.
+#                     All rights reserved.
+#
+#  This file is proprietary and confidential to TAOS Technologies.
+#  No part of this file may be reproduced, stored, transmitted,
+#  disclosed or used in any form or by any means other than as
+#  expressly provided by the written permission from Jianhui Tao
+#
+###################################################################
+
+# -*- coding: utf-8 -*-
+
+import sys
+import taos
+import threading
+import time
+from datetime import datetime
+import numpy as np
+
+class MyThread(threading.Thread):
+
+    def __init__(self, func, args=()):
+        super(MyThread, self).__init__()
+        self.func = func
+        self.args = args
+
+    def run(self):
+        self.result = self.func(*self.args)
+
+    def get_result(self):
+        try:
+            return self.result   # 如果子线程不使用join方法，此处可能会报没有self.result的错误
+        except Exception:
+            return None
+
+class MetadataQuery:
+    def initConnection(self):
+        self.tables = 100
+        self.records = 10
+        self.numOfTherads =5 
+        self.ts = 1537146000000
+        self.host = "127.0.0.1"
+        self.user = "root"
+        self.password = "taosdata"
+        self.config = "/etc/taos"
+        self.conn = taos.connect( self.host, self.user, self.password, self.config)
+    def connectDB(self):
+        return self.conn.cursor()
+
+    def createStable(self):
+        print("================= Create stable meters =================")
+        cursor = self.connectDB()
+        cursor.execute("drop database if exists test")
+        cursor.execute("create database test")
+        cursor.execute("use test")
+        cursor.execute('''create table if not exists meters (ts timestamp, speed int) tags(
+                    tgcol1 tinyint, tgcol2 smallint, tgcol3 int, tgcol4 bigint, tgcol5 float, tgcol6 double, tgcol7 bool, tgcol8 binary(20), tgcol9 nchar(20),
+                    tgcol10 tinyint, tgcol11 smallint, tgcol12 int, tgcol13 bigint, tgcol14 float, tgcol15 double, tgcol16 bool, tgcol17 binary(20), tgcol18 nchar(20),
+                    tgcol19 tinyint, tgcol20 smallint, tgcol21 int, tgcol22 bigint, tgcol23 float, tgcol24 double, tgcol25 bool, tgcol26 binary(20), tgcol27 nchar(20),
+                    tgcol28 tinyint, tgcol29 smallint, tgcol30 int, tgcol31 bigint, tgcol32 float, tgcol33 double, tgcol34 bool, tgcol35 binary(20), tgcol36 nchar(20),
+                    tgcol37 tinyint, tgcol38 smallint, tgcol39 int, tgcol40 bigint, tgcol41 float, tgcol42 double, tgcol43 bool, tgcol44 binary(20), tgcol45 nchar(20),
+                    tgcol46 tinyint, tgcol47 smallint, tgcol48 int, tgcol49 bigint, tgcol50 float, tgcol51 double, tgcol52 bool, tgcol53 binary(20), tgcol54 nchar(20))''')
+        cursor.close()
+
+    def createTablesAndInsertData(self, threadID):
+        cursor = self.connectDB()
+        cursor.execute("use test")
+        base = threadID * self.tables
+
+        tablesPerThread = int (self.tables / self.numOfTherads)
+        for i in range(tablesPerThread):
+            cursor.execute(
+                '''create table t%d using meters tags(
+                %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
+                %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
+                %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
+                %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
+                %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d',
+                %d, %d, %d, %d, %f, %f, %d, 'taosdata%d', '涛思数据%d')''' %
+                (base + i + 1, 
+                (base + i) %100, (base + i) %10000, (base + i) %1000000, (base + i) %100000000, (base + i) %100 * 1.1, (base + i) %100 * 2.3, (base + i) %2, (base + i) %100, (base + i) %100, 
+                (base + i) %100, (base + i) %10000, (base + i) %1000000, (base + i) %100000000, (base + i) %100 * 1.1, (base + i) %100 * 2.3, (base + i) %2, (base + i) %100, (base + i) %100, 
+                (base + i) %100, (base + i) %10000, (base + i) %1000000, (base + i) %100000000, (base + i) %100 * 1.1, (base + i) %100 * 2.3, (base + i) %2, (base + i) %100, (base + i) %100, 
+                (base + i) %100, (base + i) %10000, (base + i) %1000000, (base + i) %100000000, (base + i) %100 * 1.1, (base + i) %100 * 2.3, (base + i) %2, (base + i) %100, (base + i) %100, 
+                (base + i) %100, (base + i) %10000, (base + i) %1000000, (base + i) %100000000, (base + i) %100 * 1.1, (base + i) %100 * 2.3, (base + i) %2, (base + i) %100, (base + i) %100, 
+                (base + i) %100, (base + i) %10000, (base + i) %1000000, (base + i) %100000000, (base + i) %100 * 1.1, (base + i) %100 * 2.3, (base + i) %2, (base + i) %100, (base + i) %100))
+            for j in range(self.records):
+                cursor.execute(
+                    "insert into t%d values(%d, %d)" %
+                    (base + i + 1, self.ts + j, j))
+        cursor.close()
+    def queryWithTagId(self, threadId, tagId, queryNum): 
+        print("---------thread%d start-----------"%threadId)
+        query = '''select tgcol1, tgcol2, tgcol3, tgcol4, tgcol5, tgcol6, tgcol7, tgcol8, tgcol9,
+                tgcol10, tgcol11, tgcol12, tgcol13, tgcol14, tgcol15, tgcol16, tgcol17, tgcol18,
+                tgcol19, tgcol20, tgcol21, tgcol22, tgcol23, tgcol24, tgcol25, tgcol26, tgcol27,
+                tgcol28, tgcol29, tgcol30, tgcol31, tgcol32, tgcol33, tgcol34, tgcol35, tgcol36,
+                tgcol37, tgcol38, tgcol39, tgcol40, tgcol41, tgcol42, tgcol43, tgcol44, tgcol45,
+                tgcol46, tgcol47, tgcol48, tgcol49, tgcol50, tgcol51, tgcol52, tgcol53, tgcol54
+                from meters where tgcol{id} > {condition}'''
+        latancy = []
+        cursor = self.connectDB()
+        cursor.execute("use test")
+        for i in range(queryNum): 
+            startTime = time.time()
+            cursor.execute(query.format(id = tagId, condition = i))
+            cursor.fetchall()
+            latancy.append((time.time() - startTime)) 
+        print("---------thread%d end-----------"%threadId)
+        return latancy
+    def queryData(self, query):
+        cursor = self.connectDB()
+        cursor.execute("use test")
+
+        print("================= query tag data =================")    
+        startTime = datetime.now()
+        cursor.execute(query)
+        cursor.fetchall()
+        endTime = datetime.now()
+        print(
+            "Query time for the above query is %d seconds" %
+            (endTime - startTime).seconds)
+
+        cursor.close()
+        #self.conn.close()
+
+
+if __name__ == '__main__':
+
+    t = MetadataQuery()
+    t.initConnection()
+
+    latancys = [] 
+    threads =  []
+    tagId = 1
+    queryNum = 1000 
+    for i in range(t.numOfTherads):
+        thread = MyThread(t.queryWithTagId, args = (i, tagId, queryNum))
+        threads.append(thread) 
+        thread.start()
+    for i in range(t.numOfTherads):  
+        threads[i].join()
+        latancys.extend(threads[i].get_result()) 
+    print("Total query: %d"%(queryNum * t.numOfTherads))    
+    print("statistic(s): mean= %f, P50 = %f, P75 = %f, P95 = %f, P99 = %f"
+            %(sum(latancys)/(queryNum * t.numOfTherads), np.percentile(latancys, 50), np.percentile(latancys, 75), np.percentile(latancys, 95), np.percentile(latancys, 99)))  
+