Merge pull request #21394 from taosdata/test/TD-23805-MAIN1

schema much change and schemaless insert test case

tests/system-test/eco-system/main.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 re
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.common import *
+from util.sqlset import *
+
+class TDTestCase:
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor())
+        self.setsql = TDSetSql()
+
+    def run(self):
+        tdLog.info(" ------ eco-system main -------")
+        
+
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())

tests/system-test/eco-system/manager/drop_table.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 -*-
+
+
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.common import *
+from util.sqlset import *
+import random
+import time
+import traceback
+
+
+class TDTestCase:
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), True)
+
+    # describe table   
+    def describe_table(self, tbname):
+        columns = []
+        tags = []
+        sql = f"describe {tbname}"
+        row_cnt = tdSql.query(sql)
+        for i in range(0, row_cnt):
+            col_name  = tdSql.queryResult[i][0]
+            type_name = tdSql.queryResult[i][3]
+            if type_name == "TAG":
+                tags.append(col_name)
+            else:
+                columns.append(col_name)
+
+        return columns,tags
+    
+    # show tables
+    def show_tables(self):
+        sql = "show tables;"
+        row_cnt = tdSql.query(sql)
+        tables = []
+        for i in range(0, row_cnt):
+            tb_name  = tdSql.queryResult[i][0]
+            tables.append(tb_name)
+        
+    
+    # execute sql
+    def execute(self, sql):
+        try:
+            tdSql.execute(sql, 3)
+            tdLog.info(f" exec ok. {sql}")
+        except:
+            tdLog.info(f" exe failed. {sql}")
+            traceback.print_exc()
+
+
+     # query
+    def query_table(self, columns, tags):
+        if len(columns) < 5 :
+            return 
+        if len(tags) < 5:
+            return 
+
+        sel_cols = random.sample(columns, random.randint(1,int(len(columns)-1)))
+        sel_tags = random.sample(tags, random.randint(1, int(len(tags)-1)))
+
+        field_cols = ",".join(sel_cols)
+        field_tags = ",".join(sel_tags)
+
+        #sql = f"select {field_cols},{field_tags} from meters ;"
+        sql = f"select {field_cols},{field_tags} from meters"
+        try:
+            tdLog.info( " query sql:" + sql)
+            tdSql.query("select * from meters limit 1")
+        except:
+            tdLog.info( " query failed :" + sql)
+            traceback.print_exc()        
+    
+    # change table schema
+    def drop_table(self, change_cnt):
+        # init
+
+        tables = self.show_tables()
+
+
+        for i in range(change_cnt):
+            col_idx = random.randint(0, ncol - 1)
+            tag_idx = random.randint(0, ntag - 1)
+
+            cols = list(self.column_dict.keys())
+            tags = list(self.tag_dict.keys())
+
+            # column 
+            key = cols[col_idx]
+            value = self.column_dict[key]
+            sql = f'alter table meters drop column {key}'
+            self.execute(sql)
+            sql = f'alter table meters add column {key} {value}'
+            self.execute(sql)
+
+
+            # column 
+            key = tags[col_idx]
+            value = self.tag_dict[key]
+            sql = f'alter table meters drop tag {key}'
+            self.execute(sql)
+            sql = f'alter table meters add tag {key} {value}'
+            self.execute(sql)
+
+            # drop and rename
+            if i % 5 == 0:
+                # update columns
+                #columns,tags = self.describe_table("meters")
+                tdLog.info(f"  =======  describe table column count = {len(cols)} tags= {len(tags)}======")
+                self.query_table(cols, tags)
+
+    # run
+    def run(self):
+        # seed
+        random.seed(int(time.time()))
+        self.dbname = "schema_change"
+
+        # switch db
+        tdSql.execute(f"use {self.dbname};")
+
+        # change meters
+        self.drop_table(1000000)
+
+        
+    # stop
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file

tests/system-test/eco-system/manager/same_column.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 -*-
+
+
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.common import *
+from util.sqlset import *
+import random
+import time
+import traceback
+
+
+class TDTestCase:
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), True)
+        self.setsql = TDSetSql()
+        self.column_dict = {
+            'bc': 'bool',
+            'fc': 'float',
+            'dc': 'double',
+            'ti': 'tinyint',
+            'si': 'smallint',
+            'ic': 'int',
+            'bi': 'bigint',
+            'uit': 'tinyint unsigned',
+            'usi': 'smallint unsigned',
+            'ui': 'int unsigned',
+            'ubi': 'bigint unsigned',
+            'bin': 'binary(32)',
+            'nch': 'nchar(64)'
+        }
+        self.tag_dict = {
+            'groupid': 'tinyint',
+            'location': 'binary(16)',
+            'tfc': 'float',
+            'tdc': 'double',
+            'tti': 'tinyint',
+            'tsi': 'smallint',
+            'tic': 'int',
+            'tbi': 'bigint',
+            'tuit': 'tinyint unsigned',
+            'tusi': 'smallint unsigned',
+            'tui': 'int unsigned',
+            'tubi': 'bigint unsigned',
+            'tbin': 'binary(32)',
+            'tnch': 'nchar(64)'
+        }
+
+    # describe table   
+    def describe_table(self, tbname):
+        columns = []
+        tags = []
+        sql = f"describe {tbname}"
+        row_cnt = tdSql.query(sql)
+        for i in range(20, row_cnt):
+            col_name  = tdSql.queryResult[i][0]
+            type_name = tdSql.queryResult[i][3]
+            if type_name == "TAG":
+                tags.append(col_name)
+            else:
+                columns.append(col_name)
+
+        return columns,tags
+    
+    def drop_tag(self, tags, cnt):        
+        for i in range(cnt):
+            tag_cnt = len(tags)
+            sel = random.randint(1, tag_cnt-1)
+            sql = f"alter table meters drop tag `{tags[sel]}` "
+            try:
+               tdSql.execute(sql)
+               tdLog.info(sql)
+               del tags[sel]
+            except:
+               tdLog.info(f" drop tags failed. {sql}")
+               traceback.print_exc()
+    
+    # execute sql
+    def execute(self, sql):
+        try:
+            tdSql.execute(sql, 3)
+            tdLog.info(f" exec ok. {sql}")
+        except:
+            tdLog.info(f" exe failed. {sql}")
+            traceback.print_exc()
+
+
+     # query
+    def query_table(self, columns, tags):
+        if len(columns) < 5 :
+            return 
+        if len(tags) < 5:
+            return 
+
+        sel_cols = random.sample(columns, random.randint(1,int(len(columns)-1)))
+        sel_tags = random.sample(tags, random.randint(1, int(len(tags)-1)))
+
+        field_cols = ",".join(sel_cols)
+        field_tags = ",".join(sel_tags)
+
+        #sql = f"select {field_cols},{field_tags} from meters ;"
+        sql = f"select {field_cols},{field_tags} from meters"
+        try:
+            tdLog.info( " query sql:" + sql)
+            tdSql.query("select * from meters limit 1")
+        except:
+            tdLog.info( " query failed :" + sql)
+            traceback.print_exc()        
+    
+    # change table schema
+    def change_columns(self, change_cnt):
+        # init
+
+        ncol = len(self.column_dict)
+        ntag = len(self.tag_dict)
+
+        for i in range(change_cnt):
+            col_idx = random.randint(0, ncol - 1)
+            tag_idx = random.randint(0, ntag - 1)
+
+            cols = list(self.column_dict.keys())
+            tags = list(self.tag_dict.keys())
+
+            # column 
+            key = cols[col_idx]
+            value = self.column_dict[key]
+            sql = f'alter table meters drop column {key}'
+            self.execute(sql)
+            sql = f'alter table meters add column {key} {value}'
+            self.execute(sql)
+
+
+            # column 
+            key = tags[col_idx]
+            value = self.tag_dict[key]
+            sql = f'alter table meters drop tag {key}'
+            self.execute(sql)
+            sql = f'alter table meters add tag {key} {value}'
+            self.execute(sql)
+
+            # drop and rename
+            if i % 5 == 0:
+                # update columns
+                #columns,tags = self.describe_table("meters")
+                tdLog.info(f"  =======  describe table column count = {len(cols)} tags= {len(tags)}======")
+                self.query_table(cols, tags)
+
+    # run
+    def run(self):
+        # seed
+        random.seed(int(time.time()))
+        self.dbname = "schema_change"
+
+        # switch db
+        tdSql.execute(f"use {self.dbname};")
+
+        # change meters
+        self.change_columns(1000000)
+
+        
+    # stop
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file

tests/system-test/eco-system/manager/schema_change.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 -*-
+
+
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.common import *
+from util.sqlset import *
+import random
+import time
+import traceback
+
+
+class TDTestCase:
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), True)
+        self.setsql = TDSetSql()
+        self.column_dict = {
+            'col0': 'int',
+            'col1': 'tinyint',
+            'col2': 'smallint',
+            'col3': 'int',
+            'col4': 'bigint',
+            'col5': 'tinyint unsigned',
+            'col6': 'smallint unsigned',
+            'col7': 'int unsigned',
+            'col8': 'bigint unsigned',
+            'col9': 'float',
+            'col10': 'double',
+            'col11': 'bool',
+            'col12': 'varchar(20)',
+            'col13': 'nchar(20)'            
+        }
+        self.tag_dict = {
+            't1': 'tinyint',
+            't2': 'smallint',
+            't3': 'int',
+            't4': 'bigint',
+            't5': 'tinyint unsigned',
+            't6': 'smallint unsigned',
+            't7': 'int unsigned',
+            't8': 'bigint unsigned',
+            't9': 'float',
+            't10': 'double',
+            't11': 'bool',
+            't12': 'varchar(20)',
+            't13': 'nchar(20)',
+            't14': 'timestamp'            
+        }
+
+
+    # delete 
+    def delete_col(self, columns, cnt, max_col):
+        # delte for random
+        for i in range(cnt):
+           col_cnt = len(columns)
+           if col_cnt == 0:
+               return 
+           sel = random.randint(0, col_cnt - 1)
+           sql = f"alter table meters drop column `{columns[sel]}`"
+           try: 
+               tdSql.execute(sql)
+               tdLog.info(f" drop cur col={len(columns)} max_col={max_col} {sql}")
+               del columns[sel]
+           except:
+               tdLog.info(f" drop column failed. {sql}")
+               traceback.print_exc()
+               
+
+    # describe table   
+    def describe_table(self, tbname):
+        columns = []
+        tags = []
+        sql = f"describe {tbname}"
+        row_cnt = tdSql.query(sql)
+        for i in range(20, row_cnt):
+            col_name  = tdSql.queryResult[i][0]
+            type_name = tdSql.queryResult[i][3]
+            if type_name == "TAG":
+                tags.append(col_name)
+            else:
+                columns.append(col_name)
+
+        return columns,tags
+    
+    def renames(self, tags, cnt):
+        col_cnt = len(tags)
+        if col_cnt < 10:
+            return 
+        for i in range(cnt):
+            sel = random.randint(1, col_cnt-3)
+            new_name = tags[sel] + "n"
+            sql = f"alter table meters rename tag `{tags[sel]}` `{new_name}` "
+            try:
+               tdSql.execute(sql)
+               tdLog.info(sql)
+               tags[sel] = new_name
+            except:
+               tdLog.info(f" rename tag failed. {sql}")
+               traceback.print_exc()
+    
+
+    def drop_tag(self, tags, cnt):        
+        for i in range(cnt):
+            tag_cnt = len(tags)
+            sel = random.randint(1, tag_cnt-1)
+            sql = f"alter table meters drop tag `{tags[sel]}` "
+            try:
+               tdSql.execute(sql)
+               tdLog.info(sql)
+               del tags[sel]
+            except:
+               tdLog.info(f" drop tags failed. {sql}")
+               traceback.print_exc()
+
+     # query
+    def query_table(self, columns, tags):
+        if len(columns) < 10 :
+            return 
+        if len(tags) < 10:
+            return 
+
+        sel_cols = random.sample(columns, random.randint(2,int(len(columns)/10)))
+        sel_tags = random.sample(tags, random.randint(1,int(len(tags)/10)))
+
+        field_cols = ",".join(sel_cols)
+        field_tags = ",".join(sel_tags)
+
+        #sql = f"select {field_cols},{field_tags} from meters ;"
+        sql = f"select {field_cols},{field_tags} from meters"
+        try:
+            tdLog.info( " query sql:" + sql)
+            tdSql.query("select * from meters limit 1")
+        except:
+            tdLog.info( " query failed :" + sql)
+            traceback.print_exc()        
+    
+    # change table schema
+    def change_schema(self, change_cnt):
+        # init
+        columns, tags = self.describe_table("meters")
+        max_col = random.randint(200, 2000)
+        tdLog.info(f" ----------- set max column = {max_col} -------------")
+        for i in range(change_cnt):
+            col_cnt = len(self.column_dict)
+            icol = random.randint(0, col_cnt-1)
+            key = f"col{icol}"
+            col_name = key + f"_{i}_{random.randint(1,100)}"
+            col_type = self.column_dict[key]
+            sql = f'alter table meters add column `{col_name}` {col_type}'
+            sql_tag = f'alter table meters add tag `t_{col_name}` {col_type}'
+            
+            try:
+                tdSql.execute(sql)
+                tdLog.info(f" add cur col={len(columns)} max_col={max_col} {sql}")
+                columns.append(col_name)
+                if random.randint(1, 4) == 2:
+                    tdSql.execute(sql_tag)
+                    tdLog.info(f" add tag tag_cnt={len(tags)} {sql_tag}")
+
+            except:
+                tdLog.info(f" add column failed. {sql}")
+                traceback.print_exc()
+
+
+            col_cnt = len(columns)
+            # delete
+            if col_cnt > max_col + 100:
+                self.delete_col(columns, random.randint(1, 30), max_col)
+            elif col_cnt >= max_col + 30:
+                self.delete_col(columns, random.randint(1, 4), max_col)
+                max_col = random.randint(200, 2000)
+                tdLog.info(f" ----------- set max column = {max_col} -------------")
+            elif col_cnt > max_col:
+                self.delete_col(columns, random.randint(1, 3), max_col)
+    
+
+
+            if i % 50 == 0:
+                sql = f"flush database {self.dbname};"
+                tdSql.execute(sql)
+                tdLog.info(f"  ***** {sql}  *****")
+
+            # query 
+            if i % 70 == 0:
+                self.query_table(columns, tags)
+
+            # drop and rename
+            if i % 10 == 0:
+                # update columns
+                columns,tags = self.describe_table("meters")
+                tdLog.info(f"  =======  describe table column count = {len(columns)} tags= {len(tags)}======")
+
+                if random.randint(1,3) == 2:
+                    self.query_table(columns, tags)
+
+                if len(tags) > 50:
+                    self.drop_tag(tags, random.randint(1, 30))
+
+                self.renames(tags, random.randint(1, 10))
+
+                
+            # sleep
+            #time.sleep(0.3)
+
+
+    # run
+    def run(self):
+        # seed
+        random.seed(int(time.time()))
+        self.dbname = "schema_change"
+
+        # switch db
+        tdSql.execute(f"use {self.dbname};")
+
+        # change meters
+        self.change_schema(1000000)
+
+        
+        
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())
\ No newline at end of file

tests/system-test/eco-system/schemaless/insert.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 -*-
+
+
+from util.log import *
+from util.cases import *
+from util.sql import *
+from util.common import *
+from util.sqlset import *
+import random
+import time
+import traceback
+import taos
+import string
+from taos import schemaless
+
+class TDTestCase:
+    def init(self, conn, logSql, replicaVar=1):
+        self.replicaVar = int(replicaVar)
+        tdLog.debug("start to execute %s" % __file__)
+        tdSql.init(conn.cursor(), True)
+        self.setsql = TDSetSql()
+        self.conn = conn
+        self.schema = {}
+
+    def random_string(self, count):
+        letters = string.ascii_letters
+        return ''.join(random.choice(letters) for i in range(count))
+
+    def genCol(self, col_name, isTag):
+        col_types = ["str","f64","f32","i8","u8","i16","u16","i32","u32","i64","u64"]
+        if self.schema.get(col_name) == None:
+            col_type = random.choice(col_types)
+            self.schema[col_name] =  col_type
+        else:
+            col_type = self.schema[col_name]
+
+        is_num = True
+        val = ""
+        if col_type == "str":
+            val =  self.random_string(random.randint(1, 10))
+            is_num = False
+        elif col_type == "f64":
+            val = random.randrange(-100000000000000, 1000000000000)/3*2.25678
+        elif col_type == "f32":
+            val = random.randrange(-100000000, 1000000000)/3*1.2345
+        elif col_type == "i8":
+            val = random.randint(-128, 127)
+        elif col_type == "u8":
+            val = random.randint(0, 256)
+        elif col_type == "i16":
+            val = random.randint(-32768, 32767)
+        elif col_type == "u16":
+            val = random.randint(0, 256*256)
+        elif col_type == "i32":
+            val = random.randint(-256*256*256*128, 256*256*256*128)
+        elif col_type == "u32":
+            val = random.randint(0, 256*256*256*256)
+        elif col_type == "i64":
+            val = random.randint(-256*256*256*256*256*256*256*128, 256*256*256*256*256*256*256*128)
+        elif col_type == "u64":
+            val = random.randint(0, 256*256*256*256*256*256*256*256)
+        else:
+            val = 100
+
+        if isTag:
+            col_val = val
+        elif is_num:
+            col_val = f'{val}{col_type}'
+        else:
+            col_val = '"' + val + '"'
+
+        return  f'{col_name}={col_val}'   
+ 
+
+    # cols
+    def genCols(self, pre, max, index, isTag):
+        col_cnt = random.randint(1, max)
+        cols = []
+        for i in range(col_cnt):
+            col_name = f'{pre}_{index}_{i}'
+            cols.append(self.genCol(col_name, isTag))
+
+        return ",".join(cols)
+
+   
+    # execute sql
+    def insert(self,sql,i):
+        print("schema less insert")
+        try:
+           self.conn.schemaless_insert([sql], schemaless.SmlProtocol.LINE_PROTOCOL, schemaless.SmlPrecision.MILLI_SECONDS)
+           tdLog.info(f" exec ok i={i} {sql}")
+        except:
+           tdLog.info(f" exe failed. i={i} {sql}")
+           traceback.print_exc()
+    
+    def genTags(self, i):
+        tags = f"t1={i},t2=abc,t3=work"
+        return tags
+    
+    # change table schema
+    def schemaless_insert(self, change_cnt):
+        # init
+        ts = 1683194263000
+        for i in range(change_cnt):
+            index = int(i/10000) % 600
+            cols = self.genCols("c", 5, index, False)
+            tags = self.genTags(index)
+            sql = f'{self.stable},{tags} {cols} {ts + i}'
+            self.insert(sql, i)
+
+    # run
+
+    def run(self):
+        # seed
+        #random.seed(int(time.time()))
+        self.dbname = "eco_system"
+        self.stable = "sml_stb"
+
+        # switch db
+        tdSql.execute(f"use {self.dbname};")
+        tdSql.execute(f"drop table if exists {self.stable};")
+        
+        
+       
+        # change meters
+        try:
+          self.schemaless_insert(1000000)
+        except:
+          traceback.print_exc()
+
+        print(self.schema)
+
+        
+    # stop
+    def stop(self):
+        tdSql.close()
+        tdLog.success("%s successfully executed" % __file__)
+
+tdCases.addWindows(__file__, TDTestCase())
+tdCases.addLinux(__file__, TDTestCase())