# -*- coding: utf-8 -*- # """ 从Oracle库获取SPA SQL语句,然后在TiDB执行 """ import os, sys, re, time, json, socket, platform from optparse import OptionParser from configparser import ConfigParser import ast import hashlib # from jinja2 import Environment, FileSystemLoader ####################################################################################################################### parser = OptionParser(prog="SqlAnalyzer.py", usage="python3 %prog --config ./SqlAnalyzer.ini ", description="分析SQL语句的写法和执行计划,找出疑似差的执行计划" ) parser.add_option("--config", action="store", dest="config", default="SqlAnalyzer.ini", help="配置文件名称") parser.add_option("--sqlfromoraspa", action="store", dest="sqlfromoraspa", help="SQL文件是否来自Oracle spa") parser.add_option("--sqlfile", action="store", dest="sqlfile", help="SQL文件名称") parser.add_option("--filetype", action="store", dest="filetype", default="json", help="文件类型,支持json/txt;txt要求以;分隔,JSON是指从Oracle导出的语句") parser.add_option("--ignoresqlfile", action="store", dest="ignoresqlfile", help="忽略执行的SQL语句,以TXT文本格式提供SQLID清单") parser.add_option("--rewritesql", action="store", dest="rewritesql", default="yes", help="从Oracle SPA抓取的SQL需要改写,TXT不改写,JSON文件指定不改写,则需要包含TIDB_SQLSTMT") parser.add_option("--dmltypes", action="store", dest="dmltypes", default="select", help="待分析的SQL语句类型,支持select,insert,update,delete,merge,replace...,默认只运行select") parser.add_option("--explainonly", action="store_const", dest="explainonly", const=True, default=False, help="是否仅执行explain,默认执行explain analyze") parser.add_option("--loglevel", action="store", dest="loglevel", default="INFO", help="日志级别,支持DEBUG(4)/INFO(3)/WARN(2)/ERROR(1),默认INFO(3)") ####################################################################################################################### PLATFORM_NAME = platform.system() CUR_HOSTNAME = socket.gethostname() CUR_USERNAME = os.getenv('USER', None) CUR_USERHOME = os.getenv('HOME', '/tmp') SCRIPTNAME = os.path.basename(__file__) WORKDIR = os.getcwd() LOGDIR = "%s/logs" % (WORKDIR) RUNTS = time.strftime("%Y%m%d.%H%M%S", time.localtime()) LOGFILE = "%s/logs/%s_%s.log" % (WORKDIR, SCRIPTNAME, RUNTS) DBNAME = 'test' TABLENAME = "o2t_sqlset_exec_info" loglevel_dict = {"DEBUG": 4, "INFO": 3, "WARN": 2, "ERROR": 1, "STOP": 1, "SUCC": 1} loglevel_int = 3 logflag_int = 3 ####################################################################################################################### if not os.path.exists(LOGDIR): os.mkdir(LOGDIR) ####################################################################################################################### def f_PrintfLog(logtext='', logflag='info'): '''打印日志,并写入日志文件''' if logflag.upper() in loglevel_dict: logflag_int = loglevel_dict[logflag.upper()] if logflag_int <= loglevel_int: print('%s [%s] %s@%s: %s ' % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), logflag.upper(), CUR_USERNAME, CUR_HOSTNAME, logtext)) with open(LOGFILE, mode='a') as f: f.write('%s [%s] %s@%s: %s \n' % ( time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), logflag.upper(), CUR_USERNAME, CUR_HOSTNAME, logtext)) if logflag == "ERROR": sys.exit(1) def f_pick_para(sql=''): ''' 从字符串中提取:开头的变量名称 ''' # par_list = re.findall(r':\d+|:\"SYS_.+?\"|:B\d+', sql) if re.search(':\"SYS_.+?\"', sql): sql = sql.replace('\"', '') par_list = re.findall(r':[\w:]+\b|:SYS_.+?\b', sql) f_PrintfLog(par_list, "DEBUG") # para list par_del_list = [] for par in par_list: tmp_list = re.findall(r':[\w]+', par) if len(tmp_list) == 1: par_del_list.append(par) return par_del_list def f_replace_limit(sql=''): ''' 移除rownum ''' sql_str = sql sql_str = re.sub(r'(fetch +first )(.*)(rows +only)', r'limit \2', sql_str, flags=re.IGNORECASE) sql_str = re.sub(r'(and +rownum[<>= ]+)([\d]+)', r'limit \2', sql_str, flags=re.IGNORECASE) sql_str = re.sub(r'(where +rownum[<>= ]+)([\d]+)', r'limit \2', sql_str, flags=re.IGNORECASE) return sql_str def f_remove_hint(sql=''): ''' 移除hint,\" /* */ ''' hint_list = re.findall(r'/\*.+?\*/', sql) for k in hint_list: sql = sql.replace(k, '') sql = sql.replace('\n', ' ') sql = sql.replace('\t', ' ') sql = sql.replace('\"', '') return sql def f_ts_format(ts=''): """ 01-JUN-22 12.00.00.000000 AM to YYYY-MM-DD HH24.MI.SI.SSSSSS """ mon_dict = {"JAN": "01", "FEB": "02", "MAR": "03", "APR": "04", "MAY": "05", "JUN": "06", "JUL": "07", "AUG": "08", "SEP": "09", "OCT": "10", "NOV": "11", "DEC": "12"} ts_lst = ts.replace('\'', '').split(' ') ts_dt = ts_lst[0].split('-') ts_dt2 = "20%s-%s-%s" % (ts_dt[2], mon_dict[ts_dt[1]], ts_dt[0]) if ts_lst[2] == "PM": ts_tm = ts_lst[1].split('.') if int(ts_tm[0]) < 12: ts_tm[0] = str(int(ts_tm[0]) + 12) ts_tm2 = ".".join(ts_tm) else: ts_tm2 = ts_lst[1] return "'%s %s'" % (ts_dt2, ts_tm2) def f_check_exists(sql=''): ''' check not exists/not in ''' dt_list = re.findall(r'not +exists|not +in', sql, flags=re.IGNORECASE) if dt_list: f_PrintfLog("find %s from sql:%s..." % (dt_list, sql[:100]), "DEBUG") return True, str(dt_list) else: return False, None def f_check_func(sql=''): ''' check func in where ''' dt_list = re.findall(r'(and|or|where)( +\w+\([\w.+-, /*%]+\))', sql, flags=re.IGNORECASE) fu_lst = [] if dt_list: f_PrintfLog("find %s from sql:%s..." % (dt_list, sql[:100]), "DEBUG") for fu in dt_list: if type(fu) == type(()): fu_lst.append(fu[1]) else: fu_lst.append(fu) return True, str(fu_lst) else: return False, None def f_sql_rewrite(sql='', para_all=[], para_pos=[], para_val=[]): ''' 将SQL语句的参数值替换为真实值 ''' # print(sql) for k in reversed(para_all): k_pos = str(para_all.index(k) + 1) # time format=:11 if k_pos in para_pos: # print("%s %s in %s value is %s" % (k_pos, k, para_pos, para_val[para_pos.index(k_pos)])) v1_str = para_val[para_pos.index(k_pos)] if not re.search(':', v1_str): # timestamp foramt if re.search(r'\d{2}-[A-Z]{3}-\d{2}[\d. ]+[AP]M', v1_str): v1_str = f_ts_format(v1_str) sql = sql.replace(str(k), v1_str, 1) else: # print("%s not in %s " %(k_pos, para_pos)) sql = sql.replace(k, '\'\'') for k in reversed(para_all): k_pos = str(para_all.index(k) + 1) # time format=:11 if k_pos in para_pos: # print("%s %s in %s value is %s" % (k_pos, k, para_pos, para_val[para_pos.index(k_pos)])) if re.search(':', para_val[para_pos.index(k_pos)]): sql = sql.replace(str(k), para_val[para_pos.index(k_pos)], 1) # print(sql) return sql def caculate_ms(tm=''): """ 时间单位格式化到毫秒 """ hr_lst = re.findall(r'\d+h', tm) mi_lst = re.findall(r'\d+m\d', tm) se_lst = re.findall(r'[\d.]+s', tm) ms_lst = re.findall(r'[\d.]+ms', tm) µs_lst = re.findall(r'[\d.]+µs', tm) ns_lst = re.findall(r'[\d.]+ns', tm) ms = 0 if len(hr_lst) > 0: ms = ms + float(hr_lst[0].replace('h', '')) * 3600 * 1000 if len(mi_lst) > 0: mi2_lst = re.findall(r'\d+m', mi_lst[0]) ms = ms + float(mi2_lst[0].replace('m', '')) * 60 * 1000 if len(se_lst) > 0: ms = ms + float(se_lst[0].replace('s', '')) * 1000 if len(ms_lst) > 0: ms = ms + float(ms_lst[0].replace('ms', '')) if len(µs_lst) > 0: ms = ms + float(µs_lst[0].replace('µs', '')) / 1000 if len(ns_lst) > 0: ms = ms + float(ns_lst[0].replace('ns', '')) / 1000 / 1000 return ms def f_runsql_tidb(sql_prefix='', sql='', sqltype='', sql_database=''): ''' try mysql ''' t0 = time.time() try: if sql_database != '': cursor.execute("""use %s""" % (sql_database.lower())) cursor.execute("""%s %s""" % (sql_prefix, sql)) elapse_ms = (time.time() - t0) * 1000 rows = cursor.fetchall() f_dict_plus1(succSqltype_dict, sqltype) f_PrintfLog("%s [SUCC] run successful. %s" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), sql), "DEBUG") except Exception as e: elapse_ms = (time.time() - t0) * 1000 f_dict_plus1(errCode_dict, e.errno) f_dict_plus1(failSqltype_dict, sqltype) if e.errno not in errCodeInfo_dict.keys(): errCodeInfo_dict[e.errno] = str(e)[:200] f_PrintfLog("%s [ERROR] run failed, Error:%s" % (time.strftime("%Y-%m-%d-%H.%M.%S", time.localtime()), str(e)[:250]), "DEBUG") return (False, str(e)[:250], None, round(elapse_ms,2), None) # engine exec time if sql_prefix == "explain": exec_ts = 0.0 else: #explain analyze exec_info = rows[0][5].split(',')[0] if exec_info.startswith('time:'): exec_info = exec_info.replace('time:', '') exec_ts = caculate_ms(exec_info) else: f_PrintfLog("not a valid time:%s" % (str(exec_ts)), "WARN") exec_ts = None plan_str = "" for row_plan in rows: # print(row_plan) plan_str = plan_str + str(row_plan) + '\n' return (True, 'successful', plan_str, round(elapse_ms, 2), round(exec_ts, 2)) # return (True, 'successful', str(rows), round(elapse_ms,2), round(exec_ts,2)) def f_dict_plus1(d={}, k=''): ''' count+1 ''' if k in d.keys(): d[k] = d[k] + 1 else: d[k] = 1 def f_create_config_tab(dbName='test'): ''' 创建日志表,用于存储json数据 ''' f_PrintfLog("start to run f_create_config_tab") f_PrintfLog("start to create table test.o2t_sqlset_exec_info on database.") stmt = "select count(1) from information_schema.tables where table_schema='test' and table_name='o2t_sqlset_exec_info'" cursor.execute(stmt) for row in cursor.fetchall(): if row[0] > 0: f_PrintfLog("table already exists in test, not need create.") return config_tabs = {} config_tabs["o2t_sqlset_exec_info"] = """ CREATE TABLE test.o2t_sqlset_exec_info ( ID int(11) NOT NULL AUTO_INCREMENT, UPDATETIME datetime DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, JSON_FILE_NAME varchar(512), TIDB_SQLSTMT longtext , TIDB_EXEC_STATUS varchar(512) , TIDB_PLAN longtext , TIDB_ELAPSED_TIME_MS decimal(38,20) , TIDB_ENGINE_TIME_MS decimal(38,20) , TIDB_EXEC_TIME datetime , TIDB_DIGEST varchar(64) , TIDB_USE_WHERE_FUNC varchar(512) , TIDB_USE_NOT_EXISTS varchar(128) , TIDB_PLAN_NOT_GOOD varchar(128) , ORA_SQLSET_NAME varchar(64) , ORA_SQLSET_OWNER varchar(64) , ORA_MODULE varchar(64) , ORA_SQL_ID varchar(32) , ORA_SQL_TEXT longtext , ORA_PARSING_SCHEMA_NAME varchar(32) , ORA_PLAN_HASH_VALUE varchar(64) , ORA_PARAMETER_POSITION varchar(2000) , ORA_PARAMETER_VALUE longtext , ORA_EXECUTIONS int(11) , ORA_ELAPSED_TIME_MS decimal(38,20) , ORA_LAST_EXEC_START_TIME varchar(32) , ORA_PLAN_TIMESTAMP datetime , STMT_MD5 varchar(128), USER_OPERATE_FLAG varchar(128), PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin """ try: cursor.execute("use test") cursor.execute(config_tabs["o2t_sqlset_exec_info"]) cnx.commit() f_PrintfLog("table o2t_sqlset_exec_info create successfully.") except mysql.connector.Error as e: f_PrintfLog("table o2t_sqlset_exec_info create failed, Error:%s.".format(e)) def f_getcols(tableName='o2t_sqlset_exec_info', dbName='test'): '''查询表结构''' f_PrintfLog("start to run f_getcols") col_lst = [] qry_col = """select upper(column_name) from information_schema.columns where lower(table_schema)='%s' and lower(table_name)='%s' and length(extra)=0 order by ordinal_position""" % ( dbName.lower(), tableName.lower()) cursor.execute(qry_col) rows = cursor.fetchall() for row in rows: col_lst.append(row[0]) return col_lst def f_insert_data(tableName='', colName=[], data={}): '''生产可写入表的SQL语句''' f_PrintfLog("start to run f_insert_data", "DEBUG") insert_cols = ','.join(colName) value_cols = ')s, %('.join(colName) stmt_insert = """insert into %s(%s) values( %%(%s)s )""" % (tableName, insert_cols, value_cols) f_PrintfLog("Start to run sql:%s" % (stmt_insert), "DEBUG") #prepare insert data for col in set(colName).difference(data.keys()): data[col] = None cursor.execute(stmt_insert, data) rows_insert = cursor.rowcount cnx.commit() f_PrintfLog("insert data finish. insert rows count:%s" % (rows_insert), "DEBUG") def f_readignoresql(file_name=''): '''获取不需要执行的sqlid''' with open(file_name, mode='r') as f: f_PrintfLog("start to read file: %s" % (f.name)) for line in f: ignoresql_lst.append(line.strip()) f_PrintfLog("total %s sqls in ignore file: %s" % (len(ignoresql_lst), file_name)) def f_txt2json_main(file_txt='', file_json=''): '''SQL文本文件转成JSON文件''' f_PrintfLog("start to run f_txt2json_main") stmt = "" stmt_dict = {} sqls_lst = [] with open(file_txt, mode='r') as f: f_PrintfLog("start to read file: %s" % (f.name)) for line in f: line = line.strip() # 读取SQL文本文件,分号作为结束符 if line.endswith(';'): stmt = "%s %s" % (stmt, line[:-1]) stmt_dict["STMT_MD5"] = hashlib.md5(stmt.strip().encode(encoding='utf-8')).hexdigest() stmt_dict["ORA_SQL_ID"] = stmt_dict["STMT_MD5"] stmt_dict["ORA_SQL_TEXT"] = stmt.strip() stmt_dict["TIDB_SQLSTMT"] = stmt.strip() # check not in/not exists (result, checkLabel) = f_check_exists(stmt.strip()) if result: stmt_dict['TIDB_USE_NOT_EXISTS'] = checkLabel # check where function (result, checkLabel) = f_check_func(stmt.strip()) if result: stmt_dict['TIDB_USE_WHERE_FUNC'] = checkLabel # append sql stmt sqls_lst.append(stmt_dict) stmt = "" stmt_dict = {} else: stmt = "%s %s" % (stmt, line) with open(file_json, mode='w', encoding='utf-8') as f: f_PrintfLog("start to write file: %s" % (f.name)) f.write(json.dumps({"sqls": sqls_lst}, indent=2, ensure_ascii=False)) f_PrintfLog("txt2json finish. from txt sqlfile %s to %s" % (file_txt, file_json)) del sqls_lst def f_rewriteorasql_main(file_ora='', file_tidb=''): '''Oracle SQL改写,执行SQL文本的参数替换''' f_PrintfLog("start to run f_rewriteorasql_main") with open(file_ora, mode='r') as f: f_PrintfLog(f.name, "DEBUG") file_json = json.loads(f.read()) f_PrintfLog("Find %s sqls from sqlfile %s" % (len(file_json['sqls']), file_ora), "INFO") for sql_dict in file_json['sqls']: f_PrintfLog("start to rewite sql, sql_id:%s" % (sql_dict['ORA_SQL_ID']), "DEBUG") sqlStmtID_str = sql_dict['ORA_SQL_ID'] sqlStmt_str = sql_dict['ORA_SQL_TEXT'] paraPos_lst = [] paraVal_lst = [] paraVal_dict = {} if sql_dict['ORA_PARAMETER_POSITION']: paraPos_lst = sql_dict['ORA_PARAMETER_POSITION'].split(';,') paraVal_lst = sql_dict['ORA_PARAMETER_VALUE'].split(';,') # paraVal_dict = zip(paraPos_lst, paraVal_lst) # print(type(paraVal_dict), list(paraVal_dict)) else: # print("null值的变量值: %s" % (sql_dict['PARAMETER_POSITION'])) sql_dict['ORA_PARAMETER_POSITION'] = None sql_dict['ORA_PARAMETER_VALUE'] = None sqlParas_lst = f_pick_para(sqlStmt_str) # print(sqlStmt_str) f_PrintfLog("SQL语句 %s 共找到 %s 个变量, 其中有值的变量为 %s" % (sqlStmtID_str, len(sqlParas_lst), paraPos_lst), "DEBUG") sqlStmt_str = f_remove_hint(sqlStmt_str) sqlStmt_tidb = f_sql_rewrite(sqlStmt_str, sqlParas_lst, paraPos_lst, paraVal_lst) sqlStmt_tidb = f_replace_limit(sqlStmt_tidb) sql_dict['TIDB_SQLSTMT'] = sqlStmt_tidb.lstrip() # check not in/not exists (result, checkLabel) = f_check_exists(sqlStmt_tidb) if result: sql_dict['TIDB_USE_NOT_EXISTS'] = checkLabel # check where function (result, checkLabel) = f_check_func(sqlStmt_tidb) if result: sql_dict['TIDB_USE_WHERE_FUNC'] = checkLabel f_PrintfLog("start to write data into file %s, total sql count:%s" % (file_tidb, len(file_json['sqls'])), "INFO") with open(file_tidb, 'w', encoding='utf-8') as f: f.write(json.dumps(file_json, indent=13)) f_PrintfLog("SQL rewrite finish, %s to %s" % (file_ora, file_tidb)) del file_json def f_getoraspasql_main(filename=""): '''读取Oracle测试环境的SPA数据,生成SQL JSON文本文件''' f_PrintfLog("start to run f_getoraspasql_main") import cx_Oracle global ora_conn_dict # read config config = ConfigParser() config.read(configfile) ora_conn_str = config.get('oracle', 'connstr') ora_conn_dict = ast.literal_eval(ora_conn_str) f_PrintfLog('ora_conn_dict:%s' % (ora_conn_dict), "DEBUG") sql_filter_str = config.get('sql', 'sql_factor') sql_filter_dict = ast.literal_eval(sql_filter_str) f_PrintfLog('sql_filter_dict:%s' % (sql_filter_dict), "DEBUG") order_by_str = "ORDER BY %s " %(sql_filter_dict.get('order_by', 'executions DESC')) connection = cx_Oracle.connect(user=ora_conn_dict['user'], password=ora_conn_dict['password'], dsn="%s:%s/%s" % (ora_conn_dict['host'], ora_conn_dict['port'], ora_conn_dict['sid']), encoding="UTF-8") # get spa sql stmt1 = """ SELECT t1.sqlset_name as ora_sqlset_name, t1.sqlset_owner as ora_sqlset_owner, t1.module as ora_module, t1.sql_id as ora_sql_id, t1.sql_text as ora_sql_text, t1.parsing_schema_name as ora_parsing_schema_name, t1.plan_hash_value as ora_plan_hash_value, t2.position as ora_parameter_position, t2.value as ora_parameter_value, t1.executions as ora_executions, round(t1.elapsed_time/1000/t1.executions,2) as ora_elapsed_time_ms, replace(t1.last_exec_start_time,'/',' ') as ora_last_exec_start_time, to_char(t1.plan_timestamp,'YYYY-MM-DD HH24:MI:SS') as ora_plan_timestamp FROM dba_sqlset_statements t1 LEFT OUTER JOIN ( SELECT sql_id, plan_hash_value , listagg(position, ';,') within group(order by position) as position, listagg(getdata(value), ';,') within group(order by position) as value FROM dba_sqlset_binds WHERE value is not null GROUP BY sql_id, plan_hash_value) t2 ON t1.sql_id = t2.sql_id AND t1.plan_hash_value = t2.plan_hash_value %s """ % (order_by_str) f_PrintfLog("start to get data from oracle table dba_sqlset_statements", "INFO") items_lst = [] cursor = connection.cursor() for row in cursor.execute(stmt1): f_PrintfLog('data:%s' % (row[0]), "DEBUG") i = 0 items_dic = {} for key in cursor.description: if key[0] == "ORA_SQL_TEXT": items_dic[key[0]] = row[i].read() else: items_dic[key[0]] = row[i] i = i + 1 items_lst.append(items_dic) cursor.close() connection.close() # write json files f_PrintfLog("start to write data into file %s, total sql count:%s" % (filename, len(items_lst)), "INFO") with open(filename, mode='w') as f: f.write(json.dumps({'sqls': items_lst}, indent=13)) f_PrintfLog("write file %s finish" % (filename), "info") del items_lst def f_trytidbsql_main(filename=""): '''读取SQL文本文件,在TiDB执行''' f_PrintfLog("start to run f_trytidbsql_main") import mysql.connector # read config config = ConfigParser() config.read(configfile) tidb_conn_str = config.get('tidb', 'connstr') tidb_conn_dict = ast.literal_eval(tidb_conn_str) f_PrintfLog('tidb_conn_dict:%s' % (tidb_conn_dict), "DEBUG") sql_filter_str = config.get('sql', 'sql_factor') sql_filter_dict = ast.literal_eval(sql_filter_str) f_PrintfLog('sql_filter_dict:%s' % (sql_filter_dict), "DEBUG") sql_text_filter_str = '|'.join(sql_filter_dict["sql_text_filter"]) module_filter_str = '|'.join(sql_filter_dict["module_filter"]) global cursor, cnx cnx = mysql.connector.connect(**tidb_conn_dict) cursor = cnx.cursor() global errCode_dict, sqlTypes_dict, succSqltype_dict, failSqltype_dict, errCodeInfo_dict, notRun_dict errCode_dict = {} sqlTypes_dict = {} succSqltype_dict = {} failSqltype_dict = {} errCodeInfo_dict = {} notRun_dict = {} with open(filename, mode='r') as f: f_PrintfLog("start to read file: %s" % (f.name)) file_json = json.loads(f.read()) global sqlsucc_lst, sqlfail_lst, sqlother_lst sqlsucc_lst = [] sqlfail_lst = [] sqlother_lst = [] dmltypes_lst = dmltypes.upper().split(',') if explainonly: sql_expln = "explain" else: sql_expln = "explain analyze" f_PrintfLog("start to run sqls on %s:%s, total sql count:%s" % ( tidb_conn_dict['host'], tidb_conn_dict['port'], len(file_json['sqls']))) for sql_dict in file_json['sqls']: ##skip ignore sqls if len(ignoresql_lst) > 0 and sql_dict['ORA_SQL_ID'] in ignoresql_lst: f_PrintfLog("ignore sql %s, skip." %(sql_dict['ORA_SQL_ID']), "DEBUG") continue ##skip config filter if module_filter_str and sql_dict["ORA_MODULE"] and re.search(module_filter_str, sql_dict["ORA_MODULE"]): f_PrintfLog("ignore sql %s, skip." % (sql_dict['ORA_MODULE']), "DEBUG") continue if sql_text_filter_str and re.search(sql_text_filter_str, sql_dict["ORA_SQL_TEXT"]): f_PrintfLog("ignore sql %s, skip." % (sql_dict['ORA_SQL_TEXT']), "DEBUG") continue sqlStmtTidb_str = sql_dict['TIDB_SQLSTMT'] sqlDB_str = sql_dict.get('ORA_PARSING_SCHEMA_NAME', tidb_conn_dict.get('database', '')) f_PrintfLog(sqlStmtTidb_str, "DEBUG") ##SQL Type sqlType = sqlStmtTidb_str.split()[0].upper() sql_dict['TIDB_SQLTYPE'] = sqlType f_dict_plus1(sqlTypes_dict, sqlType) if sqlType not in dmltypes_lst: f_PrintfLog("Other sql, not need run. %s" % (sqlStmtTidb_str), "DEBUG") sqlother_lst.append(sql_dict) f_dict_plus1(notRun_dict, sqlType) else: t0 = time.time() sql_dict['TIDB_EXEC_TIME'] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) (a, b, c, d, e) = f_runsql_tidb(sql_expln, sqlStmtTidb_str, sqlType, sqlDB_str) sql_dict['TIDB_EXEC_STATUS'] = b sql_dict['TIDB_PLAN'] = c sql_dict['TIDB_ELAPSED_TIME_MS'] = d sql_dict['TIDB_ENGINE_TIME_MS'] = e if a is True: sqlsucc_lst.append(sql_dict) else: sqlfail_lst.append(sql_dict) f_PrintfLog("%s" % ('*' * 20)) f_PrintfLog("total sqls %s" % (len(file_json['sqls']))) f_PrintfLog("total sqltype and count: %s" % (sqlTypes_dict)) f_PrintfLog("not run other sqls %s" %(len(sqlother_lst))) f_PrintfLog("run seccussful sqls count: %s" % (len(sqlsucc_lst))) f_PrintfLog("succ sqltype and count: %s" % (succSqltype_dict)) f_PrintfLog("run failed sqls count: %s" % (len(sqlfail_lst))) f_PrintfLog("fail sqltype and count: %s" % (failSqltype_dict)) f_PrintfLog("fail errCode and count: %s" % (errCode_dict)) f_PrintfLog("%s" % ('*' * 20)) cursor.close() cnx.close() global sqlFile_succ, sqlFile_fail, sqlFile_other sqlFile_succ = '%s_succ' % (filename) sqlFile_fail = '%s_fail' % (filename) sqlFile_other = '%s_other' % (filename) # write files with open(sqlFile_succ, 'w') as f: f.write(json.dumps({"sqls": sqlsucc_lst}, indent=12)) with open(sqlFile_fail, 'w') as f: f.write(json.dumps({"sqls": sqlfail_lst}, indent=12)) with open(sqlFile_other, 'w') as f: f.write(json.dumps({"sqls": sqlother_lst}, indent=11)) f_PrintfLog("script %s tidb run finish!" % (filename)) global sqls_summary sqls_summary = {"total": len(file_json['sqls']), "succ": len(sqlsucc_lst), "fail": len(sqlfail_lst), "other": len(sqlother_lst) } # del file_json # del sqlsucc_lst # del sqlfail_lst # del sqlother_lst def f_runresult2table_main(filename_lst=[]): '''读取SQL文本文件,在TiDB执行''' f_PrintfLog("start to run f_runresult2table_main") import mysql.connector # read config config = ConfigParser() config.read(configfile) tidb_conn_str = config.get('tidb', 'connstr') tidb_conn_dict = ast.literal_eval(tidb_conn_str) f_PrintfLog('tidb_conn_dict:%s' % (tidb_conn_dict), "DEBUG") global cursor, cnx cnx = mysql.connector.connect(**tidb_conn_dict) cursor = cnx.cursor() # create table f_create_config_tab(dbName=DBNAME) time.sleep(5) # get table columns col_lst = f_getcols(tableName=TABLENAME, dbName=DBNAME) for filename in filename_lst: # read json file with open(filename, mode='r') as f: f_PrintfLog("start to read file: %s" % (f.name)) file_json = json.loads(f.read()) srcFile = os.path.basename(filename) f_PrintfLog("total rows in %s are %s" % (srcFile, len(file_json['sqls']))) for data_spa_dict in file_json['sqls']: data_spa_dict["JSON_FILE_NAME"] = srcFile if "TIDB_USE_NOT_EXISTS" in data_spa_dict.keys() and type(data_spa_dict["TIDB_USE_NOT_EXISTS"]) != type(""): data_spa_dict["TIDB_USE_NOT_EXISTS"] = str(data_spa_dict["TIDB_USE_NOT_EXISTS"]) if "TIDB_USE_WHERE_FUNC" in data_spa_dict.keys() and type(data_spa_dict["TIDB_USE_WHERE_FUNC"]) != type(""): data_spa_dict["TIDB_USE_WHERE_FUNC"] = str(data_spa_dict["TIDB_USE_WHERE_FUNC"]) f_insert_data(tableName=TABLENAME, colName=col_lst, data=data_spa_dict) f_PrintfLog("json file %s insert into %s finished, total rows count: %s!!!" % (f.name, TABLENAME, len(file_json['sqls']))) cursor.close() cnx.close() f_PrintfLog("All exec result file insert into table finished!" ) del file_json def f_format_html_compatibility(html_name=''): '''生成SQL兼容性html报告''' from jinja2 import Environment, FileSystemLoader f_PrintfLog("start to run f_format_html_compatibility") # f_PrintfLog("total sqltype and count: %s" % (sqlTypes_dict)) # f_PrintfLog("not run other sqls %s" %(notRun_dict)) # f_PrintfLog("succ sqltype and count: %s" % (succSqltype_dict)) # f_PrintfLog("fail sqltype and count: %s" % (failSqltype_dict)) # f_PrintfLog("fail errCode and count: %s" % (errCode_dict)) f_PrintfLog("Summary table") comp_summary_lst = [] for k in succSqltype_dict.keys(): k_succ = succSqltype_dict.get(k, 0) k_fail = failSqltype_dict.get(k, 0) k_total = k_succ + k_fail k_pct = round(k_succ*100/k_total, 2) k_dict = {"TYPE": k, "TOTAL": k_total, "SUCC": k_succ, "FAIL": k_fail, "PCT": k_pct} comp_summary_lst.append(k_dict) for k in failSqltype_dict.keys(): if k in succSqltype_dict.keys(): continue k_fail = failSqltype_dict.get(k, 0) k_dict = {"TYPE": k, "TOTAL": k_fail, "SUCC": 0, "FAIL": k_fail, "PCT": 0} comp_summary_lst.append(k_dict) for k in notRun_dict.keys(): k_total = notRun_dict.get(k, 0) k_dict = {"TYPE": k, "TOTAL": k_total, "SUCC": "N/A", "FAIL": "N/A", "PCT": ""} comp_summary_lst.append(k_dict) # total k_succ = len(sqlsucc_lst) k_fail = len(sqlfail_lst) k_total = k_succ + k_fail k_pct = round(k_succ*100/k_total, 2) k_dict = {"TYPE": "TOTAL:", "TOTAL": k_total, "SUCC": k_succ, "FAIL": k_fail, "PCT": k_pct} comp_summary_lst.append(k_dict) # print(type(comp_summary_lst), len(comp_summary_lst), comp_summary_lst) f_PrintfLog("Error table") comp_err_lst = [] for k in errCode_dict.keys(): k_cnt = errCode_dict.get(k, 0) k_msg = errCodeInfo_dict.get(k, '') k_dict = {"CODE": k, "COUNT": k_cnt, "MSG": k_msg} comp_err_lst.append(k_dict) # print(type(comp_err_lst), len(comp_err_lst), comp_err_lst) # html_update_dict = {} # html_update_dict.update({'RPT_SUMMARY': comp_summary_lst, # 'ERR_SUMMARY': comp_err_lst, # 'FAILSQLS': sqlfail_lst, # 'SUCCSQLS': sqlsucc_lst, # 'OTHERSQLS': sqlother_lst, # }) f_PrintfLog("Html report") env = Environment(loader=FileSystemLoader('./')) template = env.get_template('compatibility_template.html') with open(html_name, 'w+', encoding='utf-8') as f: out = template.render(comp_summary_lst=comp_summary_lst, comp_err_lst=comp_err_lst, failsqls=sqlfail_lst, succsqls=sqlsucc_lst, othersqls=sqlother_lst ) f.write(out) def f_get_oracle_explain(sqlsetname="", sqlid="", planhashvalue="", sqlsetowner=""): '''获取Oralce执行计划''' stmt1 = """SELECT * FROM table (DBMS_XPLAN.DISPLAY_SQLSET('%s','%s','%s','TYPICAL','%s'))""" % (sqlsetname, sqlid, planhashvalue, sqlsetowner) f_PrintfLog('stmt1:%s' % (stmt1), "DEBUG") ora_cursor.execute(stmt1) row = ora_cursor.fetchall() # print(type(row), len(row)) plan_str = "" i = 0 for row_plan in row: # print(row_plan) i = i + 1 if i > 5 and re.search(r'----|\|', row_plan[0]): plan_str = plan_str + row_plan[0] + '\n' # print(plan_str) return plan_str def f_format_html_performance(html_name=''): '''生成SQL性能差异html报告''' if explainonly: f_PrintfLog("Option: explainonly, can not compare performance,quit...") return if sqlfiletype and sqlfiletype.lower() == "txt": f_PrintfLog("sqlfiletype: txt, can not compare performance,quit...") return # import cx_Oracle from jinja2 import Environment, FileSystemLoader f_PrintfLog("start to run f_format_html_performance") f_PrintfLog("succ sqltype and count: %s" % (succSqltype_dict)) f_PrintfLog("succ sql count: %s" % (len(sqlsucc_lst))) # connect oradb connection = cx_Oracle.connect(user=ora_conn_dict['user'], password=ora_conn_dict['password'], dsn="%s:%s/%s" % ( ora_conn_dict['host'], ora_conn_dict['port'], ora_conn_dict['sid']), encoding="UTF-8") global ora_cursor ora_cursor = connection.cursor() global sqlsucc_fast_lst, sqlsucc_slow_lst sqlsucc_fast_lst = [] sqlsucc_slow_lst = [] perf_summary_lst = [] for sql_dict in sqlsucc_lst: t_es = sql_dict["TIDB_ENGINE_TIME_MS"] o_es = sql_dict["ORA_ELAPSED_TIME_MS"] sql_dict["O2T_SLOW_MS"] = round(t_es - o_es, 2) sql_dict["O2T_SLOW_PCT"] = round((t_es - o_es)*100/o_es, 2) # print("TS:", t_es, o_es) if t_es <= o_es: sql_dict["PERF_WIN"] = "TIDB" sqlsucc_fast_lst.append(sql_dict) else: sql_dict["PERF_WIN"] = "ORACLE" o_plan = f_get_oracle_explain(sql_dict["ORA_SQLSET_NAME"], sql_dict["ORA_SQL_ID"], sql_dict["ORA_PLAN_HASH_VALUE"], sql_dict["ORA_SQLSET_OWNER"]) sql_dict["ORA_PLAN"] = o_plan sqlsucc_slow_lst.append(sql_dict) # print(sql_dict.keys()) # print(sql_dict) # close oradb ora_cursor.close() connection.close() # performacne summary perf_summary_lst = [{"TYPE": "TOTAL:", "TIDB_WIN": len(sqlsucc_fast_lst), "ORA_WIN": len(sqlsucc_slow_lst) },] # print(perf_summary_lst) # print(sqlsucc_slow_lst) f_PrintfLog("Html report") env = Environment(loader=FileSystemLoader('./')) template = env.get_template('performance_template.html') with open(html_name, 'w+', encoding='utf-8') as f: out = template.render(perf_summary_lst=perf_summary_lst, tidb_slow_sqls=sqlsucc_slow_lst, tidb_fast_sqls=sqlsucc_fast_lst, ) f.write(out) def f_format_html_main(): '''生成SQL兼容性和性能差异html报告''' f_PrintfLog("start to run f_format_html_main") compatibility_html_name = '%s_compatibility.html' % (sqlfile_new) f_format_html_compatibility(html_name=compatibility_html_name) performance_html_name = '%s_performance.html' % (sqlfile_new) f_format_html_performance(html_name=performance_html_name) if __name__ == '__main__': #input parameter (options, args) = parser.parse_args() sqlfile = options.sqlfile sqlfiletype = options.filetype configfile = options.config rewritesql = options.rewritesql ignoresqlfile = options.ignoresqlfile loglevel = options.loglevel sqlfromoraspa = options.sqlfromoraspa dmltypes = options.dmltypes explainonly = options.explainonly f_PrintfLog("Start to run script %s" % (SCRIPTNAME)) f_PrintfLog("the input parameters: %s" % (options)) if loglevel.upper() in loglevel_dict: loglevel_int = loglevel_dict[loglevel.upper()] f_PrintfLog("The default log level is %s:%s" % (loglevel, loglevel_int)) if not configfile: f_PrintfLog("Config file is null, pls provide the parameter:config!", "ERROR") elif not os.path.exists('%s' % (configfile)): f_PrintfLog("Config file %s not exists, pls check!!!" % (configfile), "ERROR") if sqlfromoraspa and sqlfromoraspa.lower() == "no" and not sqlfile: f_PrintfLog("Sqlfile is null, pls provide it by the parameter:sqlfile!", "ERROR") elif sqlfromoraspa and sqlfile and not os.path.exists('%s' % (sqlfile)): f_PrintfLog("Sqlfile %s not exists, pls check!!!" % (sqlfile), "ERROR") if not sqlfromoraspa and not sqlfile: f_PrintfLog("Sqlfile is null, start to get sql stmt from oracle database!", "WARN") sqlfromoraspa = "yes" ############################# ### main ############################# ignoresql_lst = [] if ignoresqlfile: f_PrintfLog("start to read sql file %s." % (ignoresqlfile)) f_readignoresql(ignoresqlfile) if sqlfromoraspa and sqlfromoraspa.lower() == "yes": # 默认选项,从Oracle测试库读取SPA SQL语句 f_PrintfLog("Start to get sql stmt from oracle db!", "INFO") sqlfile = 'oracle_sqls_%s.json' % (RUNTS) f_getoraspasql_main(sqlfile) rewritesql = "yes" sqlfiletype = "json" if sqlfiletype.lower() == "txt": # txt文本的SQL语句不做改写,直接去TiDB执行 f_PrintfLog("start to read sql file %s." % (sqlfile)) sqlfile_new = "%s.%s.json" % (sqlfile.replace('.txt', ''), RUNTS) f_txt2json_main(file_txt=sqlfile, file_json=sqlfile_new) sqlfile = sqlfile_new rewritesql = "no" if sqlfiletype.lower() == "json" and rewritesql.lower() == "no": # json文本的SQL语句可以不做改写,直接去TiDB执行 f_PrintfLog("start to read sql file %s." % (sqlfile)) sqlfile_new = sqlfile if rewritesql and rewritesql.lower() == "yes": # 处理SQL语句的参数变量,如果不需要替换参数变量,可以跳过此步骤,直接在TiDB执行SQL语句 sqlfile_new = re.sub(r'(202[\d\.]{5,})', '%s.' % (RUNTS), sqlfile) sqlfile_new = "tidb_%s" % (sqlfile_new.replace('oracle_', '')) f_PrintfLog("start rewrite sql file %s to %s." % (sqlfile, sqlfile_new)) # sqlfile = "tidb_sqls_20220617.140640.json" # sqlfile_new = "tidb_sqls_20220617.140640.json_new" f_rewriteorasql_main(sqlfile, sqlfile_new) # SQL文件在TiDB执行 f_trytidbsql_main(sqlfile_new) f_PrintfLog("%s" % ('#' * 20)) # 生成html报告 f_format_html_main() f_PrintfLog("%s" % ('#' * 20)) # SQL文件的执行结果入库 file_lst = [sqlFile_succ, sqlFile_fail, ] f_PrintfLog("start to load files content(%s) into table %s.%s" % (file_lst, DBNAME, TABLENAME)) f_runresult2table_main(file_lst)