| #!/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. |
| |
| """ |
| test delete on palo |
| Date: 2015/03/23 15:08:29 |
| """ |
| |
| import time |
| from decimal import Decimal |
| import datetime |
| from data import delete as DATA |
| from lib import palo_client |
| from lib import palo_config |
| from lib import util |
| from lib import palo_task |
| |
| client = None |
| config = palo_config.config |
| column_list = DATA.column_list |
| column_name_list = DATA.column_name_list |
| partition_info = DATA.random_partition_info |
| load_data_list = DATA.load_data_list |
| broker_info = palo_config.broker_info |
| |
| |
| def setup_module(): |
| """ |
| set up |
| """ |
| # 需要修改配置文件中的max_tablet_data_size_bytes |
| global client |
| client = palo_client.PaloClient(config.fe_host, config.fe_query_port, user=config.fe_user, \ |
| password=config.fe_password) |
| client.init() |
| |
| |
| def init(database_name, table_family_name, is_wait=True): |
| """ |
| 初始化 |
| """ |
| client.clean(database_name) |
| ret = client.create_database(database_name) |
| assert ret |
| client.use(database_name) |
| ret = client.create_table(table_family_name, column_list, partition_info=partition_info) |
| assert ret |
| load_label = "%s_1" % database_name |
| for data in load_data_list: |
| data.table_name = table_family_name |
| ret = client.batch_load(load_label, load_data_list, max_filter_ratio="0.5", is_wait=is_wait, broker=broker_info) |
| assert ret |
| |
| |
| def test_bigint(): |
| """ |
| { |
| "title": "test_sys_delete.test_bigint", |
| "describe": "big int类型列上的删除条件", |
| "tag": "system,p1" |
| } |
| """ |
| """ |
| 功能点：big int类型列上的删除条件 |
| """ |
| database_name, table_family_name, rollup_table_name = util.gen_name_list() |
| print(database_name, table_family_name, rollup_table_name) |
| init(database_name, table_family_name) |
| sql = "SELECT %s FROM %s.%s" % (",".join(column_name_list), database_name, table_family_name) |
| data_1 = client.execute(sql) |
| delete_condition_list = [("bigint_key", "<=", "2")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_1) |
| |
| data_2 = client.execute(sql) |
| delete_condition_list = [("bigint_key", "<", "20"), ("bigint_key", "!=", "8")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_2) |
| |
| data_3 = client.execute(sql) |
| delete_condition_list = [("bigint_key", "<", "200"), ("bigint_key", "!=", "8")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_3) |
| |
| data_4 = client.execute(sql) |
| delete_condition_list = [("bigint_key", "<", "21225033234"), ("bigint_key", "!=", "8")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_4) |
| |
| data_5 = client.execute(sql) |
| delete_condition_list = [("bigint_key", "=", "21225033234")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_5) |
| |
| sql = "SELECT %s FROM %s.%s" % (",".join(column_name_list), database_name, table_family_name) |
| data_6 = client.execute(sql) |
| delete_condition_list = [("bigint_key", ">", "94465395517212"), \ |
| ("bigint_key", ">=", "92080498116224")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_6) |
| client.clean(database_name) |
| |
| |
| def test_date(): |
| """ |
| { |
| "title": "test_sys_delete.test_date", |
| "describe": "date类型列上的删除条件", |
| "tag": "system,p1" |
| } |
| """ |
| """ |
| 功能点：date类型列上的删除条件 |
| """ |
| database_name, table_family_name, rollup_table_name = util.gen_name_list() |
| init(database_name, table_family_name) |
| sql = "SELECT %s FROM %s.%s" % (",".join(column_name_list), database_name, table_family_name) |
| data_1 = client.execute(sql) |
| delete_condition_list = [("date_key", "!=", "1982-04-02")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_1) |
| client.clean(database_name) |
| |
| |
| def test_datetime(): |
| """ |
| { |
| "title": "test_sys_delete.test_datetime", |
| "describe": "date类型列上的删除条件", |
| "tag": "system,p1" |
| } |
| """ |
| """ |
| 功能点：date类型列上的删除条件 |
| """ |
| database_name, table_family_name, rollup_table_name = util.gen_name_list() |
| init(database_name, table_family_name) |
| sql = "SELECT %s FROM %s.%s" % (",".join(column_name_list), database_name, table_family_name) |
| data_1 = client.execute(sql) |
| delete_condition_list = [("datetime_key", "<=", "2000-12-03 16:50:21")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_1) |
| |
| data_2 = client.execute(sql) |
| delete_condition_list = [("datetime_key", "=", "2002-10-31 18:11:03")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_2) |
| |
| data_3 = client.execute(sql) |
| delete_condition_list = [("datetime_key", ">", "2013-09-05 11:16:04")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_3) |
| client.clean(database_name) |
| |
| |
| def test_char(): |
| """ |
| { |
| "title": "test_sys_delete.test_char", |
| "describe": "char类型列上的删除条件", |
| "tag": "system,p1" |
| } |
| """ |
| """ |
| 功能点：char类型列上的删除条件 |
| """ |
| database_name, table_family_name, rollup_table_name = util.gen_name_list() |
| init(database_name, table_family_name) |
| sql = "SELECT %s FROM %s.%s" % (",".join(column_name_list), database_name, table_family_name) |
| data_1 = client.execute(sql) |
| delete_condition_list = [("char_key", ">=", "y")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_1) |
| |
| data_2 = client.execute(sql) |
| delete_condition_list = [("char_key", "<=", "0"), ("char_key", ">=", "0")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_2) |
| |
| data_3 = client.execute(sql) |
| delete_condition_list = [("char_50_key", "!=", "AA6NZMc"), ("char_50_key", "<=", "m8CjCS")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_3) |
| client.clean(database_name) |
| |
| |
| def test_varchar(): |
| """ |
| { |
| "title": "test_sys_delete.test_varchar", |
| "describe": "varchar类型列上的删除条件", |
| "tag": "system,p1" |
| } |
| """ |
| """ |
| 功能点：varchar类型列上的删除条件 |
| """ |
| database_name, table_family_name, rollup_table_name = util.gen_name_list() |
| init(database_name, table_family_name) |
| sql = "SELECT %s FROM %s.%s" % (",".join(column_name_list), database_name, table_family_name) |
| data_1 = client.execute(sql) |
| delete_condition_list = [("varchar_key", "<=", "sk6S0")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_1) |
| |
| data_2 = client.execute(sql) |
| delete_condition_list = [("varchar_most_key", ">=", "z"), ("varchar_key", "=", "9C6md")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_2) |
| |
| data_3 = client.execute(sql) |
| delete_condition_list = [("varchar_key", "=", "9C6md")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_3) |
| client.clean(database_name) |
| |
| |
| def test_decimal(): |
| """ |
| { |
| "title": "test_sys_delete.test_decimal", |
| "describe": "decimal类型列上的删除条件", |
| "tag": "system,p1" |
| } |
| """ |
| """ |
| 功能点：decimal类型列上的删除条件 |
| """ |
| database_name, table_family_name, rollup_table_name = util.gen_name_list() |
| init(database_name, table_family_name) |
| sql = "SELECT %s FROM %s.%s" % (",".join(column_name_list), database_name, table_family_name) |
| data_1 = client.execute(sql) |
| delete_condition_list = [("decimal_key", "<=", "9035.16739")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_1) |
| |
| data_2 = client.execute(sql) |
| delete_condition_list = [("decimal_key", ">", "9926614.32657"), \ |
| ("decimal_most_key", "=", "123456789012345678.123456789")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_2) |
| |
| data_3 = client.execute(sql) |
| delete_condition_list = [("decimal_most_key", "=", "999999999999999999.999999999")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| check(database_name, table_family_name, delete_condition_list, data_3) |
| client.clean(database_name) |
| |
| |
| def test_selecting(): |
| """ |
| { |
| "title": "test_sys_delete.test_selecting", |
| "describe": "删除不影响查询", |
| "tag": "system,p1" |
| } |
| """ |
| """ |
| 功能点：删除不影响查询 |
| """ |
| database_name, table_family_name, rollup_table_name = util.gen_name_list() |
| init(database_name, table_family_name) |
| sql = "SELECT distinct(bigint_key) FROM %s.%s \ |
| WHERE bigint_key >= 92036854775807 \ |
| order by bigint_key" % \ |
| (database_name, table_family_name) |
| select_task = palo_task.SelectTask(config.fe_host, config.fe_query_port, sql, |
| database_name=database_name) |
| select_thread = palo_task.TaskThread(select_task) |
| select_thread.start() |
| sql = "SELECT %s FROM %s.%s" % (",".join(column_name_list), database_name, table_family_name) |
| data_1 = client.execute(sql) |
| delete_conditions_list = list() |
| sql = "SELECT bigint_key FROM %s.%s order by bigint_key desc limit 100" % \ |
| (database_name, table_family_name) |
| bigint_key_list = client.execute(sql) |
| for value in bigint_key_list: |
| delete_conditions_list.append([("bigint_key", ">", "%d" % value)]) |
| delete_task = palo_task.DeleteTask(config.fe_host, config.fe_query_port, database_name, |
| table_family_name, delete_conditions_list) |
| delete_thread = palo_task.TaskThread(delete_task) |
| delete_thread.start() |
| time.sleep(30) |
| select_thread.stop() |
| delete_thread.stop() |
| delete_thread.join() |
| print(delete_task.delete_conditions_index) |
| if delete_task.delete_conditions_index > 100: |
| idx = -1 |
| else: |
| idx = delete_task.delete_conditions_index - 1 |
| delete_condition_list = delete_conditions_list[idx] |
| check(database_name, table_family_name, delete_condition_list, data_1) |
| client.clean(database_name) |
| |
| |
| def test_two_condition(): |
| """ |
| { |
| "title": "test_sys_delete.test_two_condition", |
| "describe": "一次删除中在不同列上(k1, k2)带有条件", |
| "tag": "system,p1" |
| } |
| """ |
| """ |
| 功能点：一次删除中在不同列上(k1, k2)带有条件 |
| 验证： |
| 1. 查询中不包含条件 |
| 2. 查询中在k1或k2上带有条件 |
| 3. 查询中在k1和k2上都带有条件 |
| """ |
| database_name, table_family_name, rollup_table_name = util.gen_name_list() |
| init(database_name, table_family_name) |
| delete_condition_list = [("tinyint_key", "=", "1"), ("smallint_key", ">", "0")] |
| ret = client.delete(table_family_name, delete_condition_list) |
| assert ret |
| sql = "SELECT DISTINCT(tinyint_key) FROM \ |
| (SELECT tinyint_key, smallint_key FROM %s.%s) t ORDER BY tinyint_key" % \ |
| (database_name, table_family_name) |
| expected_data = client.execute(sql) |
| sql = "SELECT DISTINCT(tinyint_key) FROM %s.%s ORDER BY tinyint_key" % \ |
| (database_name, table_family_name) |
| actual_data = client.execute(sql) |
| assert actual_data == expected_data, "query: %s\nexpected: %s\nactual: %s" % \ |
| (sql, str(expected_data), str(actual_data)) |
| client.clean(database_name) |
| |
| |
| def test_largeint_column(): |
| """ |
| { |
| "title": "test_sys_delete.test_largeint_column", |
| "describe": "列存表上largeint类型列上的删除", |
| "tag": "system,p1" |
| } |
| """ |
| """ |
| 列存表上largeint类型列上的删除 |
| """ |
| largeint("column") |
| |
| |
| def largeint(storage_type): |
| """ |
| largeint类型列上使用delete条件测试 |
| """ |
| database_name, table_name, rollup_name = util.gen_name_list() |
| database_name = "%s_%s" % (database_name, storage_type) |
| client.clean(database_name) |
| ret = client.create_database(database_name) |
| assert ret |
| |
| ret = client.create_table(table_name, DATA.largeint_column_list, storage_type=storage_type, |
| distribution_info=DATA.largeint_distribution_info) |
| assert ret |
| |
| ret = client.stream_load(table_name, DATA.local_largeint_file_path, |
| database_name=database_name, max_filter_ratio=0.05) |
| assert ret |
| |
| ret = client.verify("./data/delete/largeint.expected", table_name) |
| assert ret |
| |
| result = client.execute("SELECT * from %s.%s" % (database_name, table_name)) |
| delete_condition_list = [("k2", "=", "-102208859176565441383255581740494103204")] |
| ret = client.delete(table_name, delete_condition_list) |
| assert ret |
| check(database_name, table_name, delete_condition_list, result) |
| |
| result = client.execute("SELECT * from %s.%s" % (database_name, table_name)) |
| delete_condition_list = [("k2", "=", "-18446744073709551616")] |
| ret = client.delete(table_name, delete_condition_list) |
| assert ret |
| check(database_name, table_name, delete_condition_list, result) |
| |
| result = client.execute("SELECT * from %s.%s" % (database_name, table_name)) |
| delete_condition_list = [("k2", "<", "-18446744073709551616")] |
| ret = client.delete(table_name, delete_condition_list) |
| assert ret |
| check(database_name, table_name, delete_condition_list, result) |
| |
| result = client.execute("SELECT * from %s.%s" % (database_name, table_name)) |
| ret = client.delete(table_name, delete_condition_list) |
| assert ret |
| check(database_name, table_name, delete_condition_list, result) |
| |
| result = client.execute("SELECT * from %s.%s" % (database_name, table_name)) |
| delete_condition_list = [("k2", "=", "18446744073709551616")] |
| ret = client.delete(table_name, delete_condition_list) |
| assert ret |
| check(database_name, table_name, delete_condition_list, result) |
| |
| result = client.execute("SELECT * from %s.%s" % (database_name, table_name)) |
| delete_condition_list = [("k2", ">=", "82824766624071150396653061164852204443")] |
| ret = client.delete(table_name, delete_condition_list) |
| assert ret |
| check(database_name, table_name, delete_condition_list, result) |
| |
| result = client.execute("SELECT * from %s.%s" % (database_name, table_name)) |
| delete_condition_list = [("k2", ">=", "31875034539506183418166553393355242463"), \ |
| ("k2", "<=", "31875034539506183418166553393355242463"), ("k1", "=", "1")] |
| ret = client.delete(table_name, delete_condition_list) |
| assert ret |
| check(database_name, table_name, delete_condition_list, result) |
| |
| result = client.execute("SELECT * from %s.%s" % (database_name, table_name)) |
| delete_condition_list = [("k2", "!=", "0")] |
| ret = client.delete(table_name, delete_condition_list) |
| assert ret |
| check(database_name, table_name, delete_condition_list, result) |
| client.clean(database_name) |
| |
| |
| def check(database_name, table_family_name, delete_condition_list, data_before=None): |
| """ |
| 校验删除正确性 |
| 只能有一个delete条件 |
| """ |
| table_schema = client.get_index_schema(table_family_name, database_name=database_name) |
| column_name_list = [column[0] for column in table_schema] |
| sql = "SELECT %s FROM %s.%s" % (",".join(column_name_list), database_name, table_family_name) |
| if data_before: |
| #用于比较 |
| op = {"<": _lt, ">": _gt, "<=": _le, ">=": _ge, "=": _eq, "!=": _ne} |
| #用于类型转化 |
| type_info = {'tinyint': int, 'smallint': int, 'int': int, 'bigint': int, 'largeint': int, \ |
| 'float': float, 'double': float, 'decimal': Decimal, 'char': str, 'varchar': str, \ |
| "character": str, 'datetime': _datetime, 'date': _date, 'decimalv3': Decimal} |
| checker_list = [] |
| for c in delete_condition_list: |
| column_index = column_name_list.index(c[0]) |
| column_type = table_schema[column_index][1] |
| if column_type.find("(") == -1: |
| column_type = type_info.get(column_type.lower()) |
| else: |
| column_type = type_info.get(column_type[:column_type.find("(")].lower()) |
| #第1个元素为比较运算符，第2元素为删除条件所在的列，第3个元素为类型转化后的value |
| checker_list.append([c[1], column_index, column_type(c[2])]) |
| |
| data_after = client.execute(sql) |
| #验证删除的数据都是符合删除条件的 |
| for data in data_before: |
| if data not in data_after: |
| for c in checker_list: |
| d = data[c[1]] |
| if table_schema[c[1]][1].lower().startswith("largeint"): |
| d = int(d) |
| assert op.get(c[0])(d, c[2]), "column %s: %s not %s %s" % (column_name_list[c[1]], d, |
| str(c[0]), c[2]) |
| |
| #验证剩下的数据正确性 |
| for data in data_after: |
| assert data in data_before, "error data: %s" % str(data) |
| |
| #验证符合删除条件的数据都被删除 |
| delete_string_list = list() |
| for c in delete_condition_list: |
| column_index = column_name_list.index(c[0]) |
| column_type = table_schema[column_index][1] |
| if column_type.find("(") != -1: |
| column_type = column_type[:column_type.find("(")] |
| c = list(c) |
| if column_type.lower() in ["char", "varchar", "datetime", "character", "date"]: |
| c[2] = "\"%s\"" % c[2] |
| delete_string_list.append(" ".join(c)) |
| sql = "%s WHERE %s" % (sql, " and ".join(delete_string_list)) |
| result = client.execute(sql) |
| assert result == (), result |
| |
| |
| def _lt(v1, v2): |
| return v1 < v2 |
| |
| |
| def _le(v1, v2): |
| return v1 <= v2 |
| |
| |
| def _gt(v1, v2): |
| return v1 > v2 |
| |
| |
| def _ge(v1, v2): |
| return v1 >= v2 |
| |
| |
| def _eq(v1, v2): |
| return v1 == v2 |
| |
| |
| def _ne(v1, v2): |
| return v1 != v2 |
| |
| |
| def _datetime(d): |
| return datetime.datetime.strptime(d, "%Y-%m-%d %H:%M:%S") |
| |
| |
| def _date(d): |
| d = datetime.datetime.strptime(d, "%Y-%m-%d") |
| return datetime.date(d.year, d.month, d.day) |
| |
| |
| |
| def teardown_module(): |
| """ |
| tear down |
| """ |
| pass |
| |
| |
| if __name__ == '__main__': |
| setup_module() |
| test_varchar() |
| # test_largeint_column() |
