| #!/bin/env python | |
| # -*- coding: utf-8 -*- | |
| # Licensed to the Apache Software Foundation (ASF) under one | |
| # or more contributor license agreements. See the NOTICE file | |
| # distributed with this work for additional information | |
| # regarding copyright ownership. The ASF licenses this file | |
| # to you under the Apache License, Version 2.0 (the | |
| # "License"); you may not use this file except in compliance | |
| # with the License. You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, | |
| # software distributed under the License is distributed on an | |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
| # KIND, either express or implied. See the License for the | |
| # specific language governing permissions and limitations | |
| # under the License. | |
| """ | |
| util模块，提供一些辅助函数 | |
| Date: 2014/08/05 17:19:26 | |
| """ | |
| import datetime | |
| import os | |
| import sys | |
| import struct | |
| import random | |
| import inspect | |
| import time | |
| import subprocess | |
| import pexpect | |
| from decimal import Decimal | |
| import hashlib | |
| import functools | |
| import palo_logger | |
| LOG = palo_logger.Logger.getLogger() | |
| L = palo_logger.StructedLogMessage | |
| def pretty(data): | |
| """ | |
| 将data转成string方便打印，需要list中的元素实现__str__方法 | |
| """ | |
| result = "" | |
| if isinstance(data, dict): | |
| result += '{' | |
| for key, value in data.iteritems(): | |
| result += '%s:%s' % (str(key), pretty(value)) | |
| result += ', ' | |
| result = result.rstrip(", ") | |
| result += '}' | |
| elif isinstance(data, list): | |
| for val in data: | |
| result += '[' | |
| result += pretty(val) | |
| result += ', ' | |
| result = result.rstrip(", ") | |
| result += ']' | |
| else: | |
| result += str(data) | |
| return result | |
| def gen_name_list(prefix=""): | |
| """ | |
| 根据调用的文件名和函数名生成database_name, table_name, index_name | |
| """ | |
| file_name = "" | |
| for command in sys.argv: | |
| if command.endswith(".py"): | |
| file_name = command | |
| dir_name, file_name = os.path.split(os.path.abspath(file_name)) | |
| file_name = file_name[:-3] | |
| function_name = inspect.stack()[1][3] | |
| database_name = "%s_%s_%s_db" % (prefix, file_name, function_name) | |
| LOG.info(L('test case', file_name=file_name, case=function_name)) | |
| if len(database_name) > 60: | |
| md5_str = get_md5(database_name) | |
| database_name = 'd' + md5_str[-4:] + '_' + database_name[-50:] | |
| database_name = database_name.lstrip("_") | |
| table_name = "%s_%s_%s_tb" % (prefix, file_name, function_name) | |
| if len(table_name) > 60: | |
| md5_str = get_md5(table_name) | |
| table_name = 't' + md5_str[-4:] + '_' + table_name[-50:] | |
| table_name = table_name.lstrip("_") | |
| index_name = "%s_%s_%s_index" % (prefix, file_name, function_name) | |
| if len(index_name) > 60: | |
| md5_str = get_md5(index_name) | |
| index_name = 'i' + md5_str[-4:] + '_' + index_name[-50:] | |
| index_name = index_name.lstrip("_") | |
| return database_name, table_name, index_name | |
| def gen_num_format_name_list(prefix=""): | |
| """ | |
| 根据调用的文件名和函数名生成database_name, table_name, index_name | |
| 在文件名和函数名生成的字符串过长时，取其md5值的一部分，尽可能减少case之间重名的问题 | |
| """ | |
| file_name = "" | |
| for command in sys.argv: | |
| if command.endswith(".py"): | |
| file_name = command | |
| dir_name, file_name = os.path.split(os.path.abspath(file_name)) | |
| file_name = file_name[:-3] | |
| function_name = inspect.stack()[1][3] | |
| # 分别获取database_name, table_name, index_name | |
| suffixes = ['db', 'tb', 'index'] | |
| names = [] | |
| for suffix in suffixes: | |
| name = "%s_%s_%s_%s" % (prefix, file_name, function_name, suffix) | |
| if len(name) > 60: | |
| # palo里的数据库和表名必须以字符串开始 | |
| md5_str = get_md5(name) | |
| name = 's' + md5_str[-4:] + '_' + name[-50:] | |
| name = name.lstrip("_") | |
| names.append(name) | |
| return names | |
| def get_label(): | |
| """ | |
| 生成label字符串 | |
| """ | |
| fmt = '%d_%H_%M_%S_%f' | |
| return "label_%s_%d" % (datetime.datetime.now().strftime(fmt), random.randint(0, 2 ** 31 - 1)) | |
| def get_snapshot_label(prefix=None): | |
| """生成snapshot label""" | |
| if prefix is None: | |
| prefix = 'random' | |
| fmt = '%d_%H_%M_%S_%f' | |
| return "%s_snapshot_%s_%d" % (prefix, datetime.datetime.now().strftime(fmt), | |
| random.randint(0, 2 ** 31 - 1)) | |
| def column_to_sql(column, set_null=False): | |
| """ | |
| 将column 4元组转成palo格式的sql字符串 | |
| (column_name, column_type, aggregation_type, default_value) | |
| """ | |
| sql = "%s %s" % (column[0], column[1]) | |
| #value列有聚合方法 | |
| #key列有默认值时指定此项为None | |
| if len(column) > 2: | |
| if column[2]: | |
| sql = "%s %s" % (sql, column[2]) | |
| if set_null is False: | |
| sql = sql + ' NOT NULL' | |
| elif set_null is True: | |
| sql = sql + ' NULL' | |
| else: | |
| pass | |
| #有默认值的列 | |
| if len(column) > 3: | |
| if column[3] is None: | |
| sql = '%s DEFAULT NULL' % sql | |
| else: | |
| sql = '%s DEFAULT "%s"' % (sql, column[3]) | |
| return sql | |
| def column_to_no_agg_sql(column, set_null=False): | |
| """ | |
| 将column 4元组转成SQL字符串 | |
| (column_name, column_type, aggregation_type, default_value) | |
| """ | |
| sql = "%s %s" % (column[0], column[1]) | |
| if set_null is False: | |
| sql = sql + ' NOT NULL' | |
| elif set_null is True: | |
| sql = sql + ' NULL' | |
| else: | |
| pass | |
| #有默认值的列 | |
| if len(column) > 3: | |
| sql = '%s DEFAULT "%s"' % (sql, column[3]) | |
| return sql | |
| def convert_agg_column_to_no_agg_column(column_list): | |
| """ | |
| 将agg column 4元组转成no agg column 4元组 | |
| (column_name, column_type, aggregation_type, default_value) | |
| """ | |
| no_agg_column = [] | |
| for i in column_list: | |
| if len(i) == 2: | |
| no_agg_column.append(i) | |
| elif len(i) == 3: | |
| no_agg_column.append((i[0], i[1])) | |
| elif len(i) == 4: | |
| no_agg_column.append((i[0], i[1], "", i[3])) | |
| return no_agg_column | |
| def file_to_insert_sql_value(file_name, to_str=False): | |
| """ | |
| 将文件中的列转为insert into类型sql中value字段 | |
| 会将文件中的N转为NULL（broker load和insert中空值的差别） | |
| :param file_name: | |
| :param to_str: 默认是false，这样的话对数字不会加双引号 | |
| :return: | |
| """ | |
| fp = open(file_name, 'r') | |
| total_sql_list = [] | |
| for line in fp.readlines(): | |
| items = line.split('\n')[0].split('\t') | |
| str = '' | |
| for i in range(len(items)): | |
| item = items[i] | |
| if not to_str and is_number(item): | |
| str += item | |
| elif item == '\\N': | |
| str += 'NULL' | |
| else: | |
| str += '"' + item + '"' | |
| if i < len(items) - 1: | |
| str += ',' | |
| total_sql_list.append('(' + str + ')') | |
| return ','.join(total_sql_list) | |
| def is_number(s): | |
| """ | |
| 验证字符串是否为数字/浮点数 | |
| :param s: | |
| :return: | |
| """ | |
| try: | |
| if s == 'NaN': | |
| return False | |
| float(s) | |
| return True | |
| except ValueError: | |
| return False | |
| def exec_cmd(cmd, user=None, password=None, host=None, timeout=30): | |
| """ | |
| 执行shell命令 | |
| """ | |
| if user is not None and password is not None and host is not None: | |
| if cmd.find("'") >= 0: | |
| raise Exception("We do not support quote ' now!") | |
| output, status = pexpect.run("ssh %s@%s '%s'" % (user, host, cmd), | |
| timeout=timeout, withexitstatus=True, | |
| events = {"continue connecting":"yes\n", "password:":"%s\n" % password}) | |
| LOG.info(L('exec remote cmd', cmd=cmd, output=output, status=status)) | |
| else: | |
| status, output = subprocess.getstatusoutput(cmd) | |
| return status, output | |
| def compare(a, b): | |
| """compare data to None""" | |
| assert isinstance(a, (tuple, list)) | |
| assert isinstance(b, (tuple, list)) | |
| for i in range(0, len(a)): | |
| if a[i] is not None and b[i] is not None: | |
| if a[i] > b[i]: | |
| return 1 | |
| elif a[i] < b[i]: | |
| return -1 | |
| else: | |
| continue | |
| elif a[i] is None and b[i] is None: | |
| continue | |
| elif a[i] is None: | |
| return -1 | |
| else: | |
| return 1 | |
| return 0 | |
| def check(palo_result, mysql_result, force_order=False): | |
| """ | |
| check the palo result and mysql result | |
| 1. data format | |
| 2. order | |
| """ | |
| if force_order and palo_result != (): | |
| try: | |
| palo_result = list(palo_result) | |
| mysql_result = list(mysql_result) | |
| palo_result.sort() | |
| mysql_result.sort() | |
| except Exception as e: | |
| if isinstance(e, TypeError): | |
| palo_result.sort(key=functools.cmp_to_key(compare)) | |
| mysql_result.sort(key=functools.cmp_to_key(compare)) | |
| if mysql_result != palo_result: | |
| if len(palo_result) != len(mysql_result): | |
| LOG.error(L('check error', palo_length=len(palo_result), | |
| expect_length=len(mysql_result))) | |
| assert 0 == 1, "\npalo_result length: %d\nmysql_result length:%d" % \ | |
| (len(palo_result), len(mysql_result)) | |
| for palo_line, mysql_line in zip(palo_result, mysql_result): | |
| if len(palo_line) != len(mysql_line): | |
| LOG.error(L('check error, number of columns not match', | |
| palo_colums_number=len(palo_line), expect_columns_number=len(mysql_line))) | |
| LOG.error(L('check error', palo_line=str(palo_line))) | |
| LOG.error(L('check error', expect_line=str(mysql_line))) | |
| assert 0 == 1, "\npalo line: %s\nmysql line:%s" % (str(palo_line), str(mysql_line)) | |
| if palo_line != mysql_line: | |
| for palo_data, mysql_data in zip(palo_line, mysql_line): | |
| if palo_data != mysql_data: | |
| same = False | |
| # str vs bytes | |
| if isinstance(palo_data, (str, bytes)) and isinstance(mysql_data, (str, bytes)): | |
| if isinstance(palo_data, bytes): | |
| palo_data = str(palo_data, "utf8") | |
| if isinstance(mysql_data, bytes): | |
| mysql_data = str(mysql_data, "utf8") | |
| if palo_data == mysql_data: | |
| return True | |
| # chinese & palo largeint return unicode | |
| if isinstance(palo_data, str): | |
| if palo_data == str(mysql_data): | |
| same = True | |
| if palo_data == mysql_data: | |
| same = True | |
| # blank | |
| if isinstance(palo_data, str): | |
| if palo_data.strip() == "" and mysql_data == "": | |
| same = True | |
| # null string | |
| if palo_data is None and mysql_data == "": | |
| same = True | |
| #float | |
| elif isinstance(mysql_data, (Decimal, float)) \ | |
| and isinstance(palo_data, (Decimal, float)): | |
| same = check_float(float(palo_data), float(mysql_data)) | |
| # list | |
| elif isinstance(mysql_data, list): | |
| same = check_list(palo_data, mysql_data) | |
| if not same: | |
| LOG.error(L('check error', palo_data=palo_data)) | |
| LOG.error(L('check error', expect_data=mysql_data)) | |
| LOG.error(L('check error', palo_line=palo_line)) | |
| LOG.error(L('check error', expect_line=mysql_line)) | |
| assert 0 == 1, "\ndiff data: \npalo:%s; \nexpect:%s;\npalo line:%s\nexpect line: %s" \ | |
| % (palo_data, mysql_data, palo_line, mysql_line) | |
| def check_float(data1, data2): | |
| """ | |
| check float | |
| """ | |
| if float(data1) == float(data2): | |
| return True | |
| if abs(float(data1) - float(data2)) < 0.001: | |
| return True | |
| if data2 != 0 and abs(float(data1) / float(data2) - 1.0) < 0.001: | |
| return True | |
| return False | |
| def check_list(data1, data2): | |
| """ | |
| check list | |
| """ | |
| if data1 is None and data2 is None: | |
| return True | |
| elif data1 is None or data2 is None: | |
| return False | |
| elif len(data2) > 0 and isinstance(data2[0], float): | |
| for d1, d2 in zip(data1, data2): | |
| if not check_float(d1, d2): | |
| return False | |
| else: | |
| return data1 == data2 | |
| return True | |
| def convert_dict2property(properties): | |
| """convert mat to property str""" | |
| sql = '(' | |
| for k, v in properties.items(): | |
| if v is not None: | |
| sql += ' "%s" = "%s", ' % (k, v) | |
| sql = sql.rstrip(', ') | |
| sql += ')' | |
| return sql | |
| def get_timestamp(palo_data): | |
| """将datetime 转为时间戳""" | |
| timeArray = palo_data[0][0].timetuple() | |
| timestamp = time.mktime(timeArray) | |
| return timestamp | |
| def check2_time_zone(palo_re, mysql_res, gap=2): | |
| """比较两个datetime时间，转为时间戳，允许的差值是gap""" | |
| palo_data = get_timestamp(palo_re) | |
| mysql_data = get_timestamp(mysql_res) | |
| assert abs(palo_data - mysql_data) <= 2 + gap, "palo_res %s, mysql_res %s, \ | |
| excepted gap < %s" % (palo_data, mysql_data, gap) | |
| def assert_return(expected_flag, expected_msg, func, *args, **kwargs): | |
| """ | |
| 验证一个函数的执行结果 | |
| 验证是否正确；如果错误的话，验证错误信息 | |
| :param expected_flag:True/False | |
| :param expected_msg: | |
| :param func: | |
| :param args: | |
| :param kwargs: | |
| :return: | |
| """ | |
| try: | |
| func(*args, **kwargs) | |
| except Exception as e: | |
| print(str(e)) | |
| print(expected_msg) | |
| LOG.info(L('get an error', msg=str(e))) | |
| assert not expected_flag, "real sql status is False, expect %s" % expected_flag | |
| assert expected_msg in str(e), "expect:%s, doris:%s" % (expected_msg, str(e)) | |
| else: | |
| assert expected_flag, "real sql status is True, expect %s" % expected_flag | |
| def get_md5(target): | |
| """ | |
| 计算字符串的md5值 | |
| :param target: | |
| :return: | |
| """ | |
| obj = hashlib.md5(b"fkldsajlkfjlaksdjfkladsjfkladsjkldsjfklfjs") | |
| obj.update(target.encode("utf-8")) | |
| return obj.hexdigest() | |
| def assert_return_flag(expected_flag, func, *args, **kwargs): | |
| """ | |
| 验证一个函数的返回值 | |
| 验证是否正确 | |
| :param expected_flag: | |
| :param func: | |
| :param args: | |
| :param kwargs: | |
| :return: | |
| """ | |
| return_flag = func(*args, **kwargs) | |
| assert expected_flag == return_flag | |
| def bitmap_index_to_sql(bitmap_index, set_null=False): | |
| """ | |
| bitmap_index: 由3元组(index_name, column_name, index_type)组成 | |
| """ | |
| sql = "INDEX %s (%s) USING %s" % (bitmap_index[0], bitmap_index[1], bitmap_index[2]) | |
| return sql | |
| def get_attr(ret, column_idx): | |
| """ | |
| 获取返回结果ret的第n列 | |
| ret = client.show_backend() | |
| be_ip = get_attr(ret, palo_job.BackendShowInfo.IP) | |
| """ | |
| return_list = list() | |
| for record in ret: | |
| return_list.append(record[column_idx]) | |
| LOG.info(L('get column from ret', idx=column_idx, ret=return_list)) | |
| return return_list | |
| def get_attr_condition_value(ret, condition_col_idx, condition_value, retrun_col_idx=None): | |
| """寻找ret的第condition_col_idx列的值为condition_value的行，返回该行的第retrun_col_idx列，只返回一个符合条件的""" | |
| if retrun_col_idx is None: | |
| retrun_col_idx = condition_col_idx | |
| for record in ret: | |
| if record[condition_col_idx] == condition_value: | |
| LOG.info(L('get result from ret', searced_key=condition_value, value=record[retrun_col_idx])) | |
| return record[retrun_col_idx] | |
| LOG.info(L('can not get result from ret', searched_key=condition_value)) | |
| return None | |
| def get_attr_condition_list(ret, condition_col_idx, condition_value, retrun_col_idx=None): | |
| """寻找ret的第condition_col_idx列的值为condition_value的行，返回所有行的第retrun_col_idx列，返回说有符合条件的""" | |
| result_list = list() | |
| if retrun_col_idx is None: | |
| retrun_col_idx = condition_col_idx | |
| for record in ret: | |
| if record[condition_col_idx] == condition_value: | |
| result_list.append(record[retrun_col_idx]) | |
| if len(result_list) == 0: | |
| return None | |
| else: | |
| LOG.info(L('get all result from ret', searced_key=condition_value, value=result_list)) | |
| return result_list | |
| def gen_tuple_num_str(begin, end): | |
| """gen_tuple_num_str(1, 3) -> ('1', '2')""" | |
| return tuple(map(str, range(begin, end))) | |
| def get_string_md5(st): | |
| """get string md5""" | |
| hl = hashlib.md5() | |
| hl.update(st.encode(encoding='utf-8')) | |
| return hl.hexdigest() | |