| #!/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. |
| |
| ############################################################################ |
| # |
| # @file test_array_alter.py |
| # @date 2022-08-15 11:09:53 |
| # @brief This file is a test file for array type. |
| # |
| ############################################################################# |
| """ |
| test_array_alter.py |
| """ |
| import sys |
| import os |
| import time |
| import pytest |
| file_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")) |
| sys.path.append(file_dir) |
| file_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) |
| sys.path.append(file_dir) |
| |
| from lib import palo_config |
| from lib import palo_client |
| from lib import palo_job |
| from lib import util |
| from lib import common |
| from data import schema as SCHEMA |
| from data import load_file as FILE |
| |
| config = palo_config.config |
| broker_info = palo_config.broker_info |
| |
| check_db = 'array_check_db' |
| check_tb = 'array_tb' |
| |
| |
| def setup_module(): |
| """setup""" |
| client = common.get_client() |
| ret = client.show_variables('enable_vectorized_engine') |
| if len(ret) == 1 and ret[0][1] == 'false': |
| raise pytest.skip('skip if enable_vectorized_engine is false') |
| |
| ret = client.admin_show_config('enable_array_type') |
| assert len(ret) == 1, 'get enable_array_type config error' |
| value = palo_job.AdminShowConfig(ret[0]).get_value() |
| if value != 'true': |
| client.set_frontend_config('enable_array_type', 'true') |
| if len(client.show_databases(check_db)) == 0: |
| init_check() |
| assert client.verify(FILE.expe_array_table_file, check_tb, database_name=check_db) |
| |
| |
| def init_check(): |
| """init check db & tb""" |
| client = common.create_workspace(check_db) |
| ret = client.create_table(check_tb, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(check_tb, FILE.test_array_table_local_file, column_separator='|') |
| assert ret, 'load chech table failed' |
| |
| |
| def teardown_module(): |
| """teardown""" |
| pass |
| |
| |
| def test_add_array_column(): |
| """ |
| { |
| "title": "test_add_array_column", |
| "describe": "array类型，增加array类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| ret = client.create_table(table_name, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_table_local_file, column_separator='|') |
| assert ret, 'stream load failed' |
| column_name_list = util.get_attr(SCHEMA.array_table_list, 0) |
| |
| column_list = [('add_arr', 'array<int>', None, "[]")] |
| ret = client.schema_change_add_column(table_name, column_list, is_wait_job=True, set_null=True) |
| assert ret, 'add column failed' |
| |
| column_list = [('add_arr', 'array<int>', None, None)] |
| msg = 'Can not set null default value to non nullable column: add_arr' |
| ret = util.assert_return(False, msg, client.schema_change_add_column, table_name, column_list, set_null=False) |
| |
| column_list = [('add_arr', 'array<int>', 'key', None)] |
| msg = 'Array can only be used in the non-key column of the duplicate table at present.' |
| ret = util.assert_return(False, msg, client.schema_change_add_column, table_name, column_list, set_null=True) |
| |
| column_list = [('add_arr1', 'array<int>', None, None)] |
| ret = client.schema_change_add_column(table_name, column_list, is_wait_job=True, set_null=True) |
| assert ret, 'add column failed' |
| |
| ret = client.desc_table(table_name) |
| assert 'ARRAY<INT(11)>' == util.get_attr_condition_value(ret, palo_job.DescInfo.Field, 'add_arr1', |
| palo_job.DescInfo.Type) |
| sql1 = 'select * from %s.%s order by k1' % (database_name, table_name) |
| sql2 = 'select *, [], null from %s.%s order by k1' % (check_db, check_tb) |
| common.check2(client, sql1, sql2=sql2) |
| |
| column_name_list.append('add_arr=a4') |
| ret = client.stream_load(table_name, FILE.test_array_table_local_file, column_separator='|', |
| column_name_list=column_name_list) |
| assert ret, 'stream load failed' |
| time.sleep(5) |
| sql1 = 'select count(*) from %s.%s' % (database_name, table_name) |
| sql2 = 'select count(*) * 2 from %s.%s' % (check_db, check_tb) |
| common.check2(client, sql1, sql2=sql2) |
| |
| client.clean(database_name) |
| |
| |
| def test_drop_array_column(): |
| """ |
| { |
| "title": "test_drop_array_column", |
| "describe": "array类型，删除array类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| ret = client.create_table(table_name, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_table_local_file, column_separator='|') |
| assert ret, 'stream load failed' |
| column_name_list = util.get_attr(SCHEMA.array_table_list, 0) |
| ret = client.schema_change_drop_column(table_name, ['a1', 'a2', 'a7', 'a11', 'a13'], is_wait_job=True) |
| assert ret, 'drop column failed' |
| # check |
| sql1 = 'select * from %s.%s order by k1' % (database_name, table_name) |
| sql2 = 'select k1, a3, a4, a5, a6, a8, a9, a10, a12, a14 from %s.%s order by k1' % (check_db, check_tb) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.desc_table(table_name) |
| assert util.get_attr_condition_value(ret, palo_job.DescInfo.Field, 'a1', palo_job.DescInfo.Type) is None |
| assert util.get_attr_condition_value(ret, palo_job.DescInfo.Field, 'a2', palo_job.DescInfo.Type) is None |
| assert util.get_attr_condition_value(ret, palo_job.DescInfo.Field, 'a7', palo_job.DescInfo.Type) is None |
| assert util.get_attr_condition_value(ret, palo_job.DescInfo.Field, 'a11', palo_job.DescInfo.Type) is None |
| assert util.get_attr_condition_value(ret, palo_job.DescInfo.Field, 'a13', palo_job.DescInfo.Type) is None |
| |
| ret = client.stream_load(table_name, FILE.test_array_table_local_file, column_separator='|', |
| column_name_list=column_name_list) |
| assert ret, 'stream load failed' |
| time.sleep(5) |
| sql1 = 'select count(*) from %s.%s' % (database_name, table_name) |
| sql2 = 'select count(*) * 2 from %s.%s' % (check_db, check_tb) |
| common.check2(client, sql1, sql2=sql2) |
| |
| client.clean(database_name) |
| |
| |
| def test_modify_array_column_not_support(): |
| """ |
| { |
| "title": "test_modify_array_column_not_support", |
| "describe": "不支持修改array类型", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| ret = client.create_table(table_name, SCHEMA.array_boolean_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| msg = 'Can not change ARRAY to STRING' |
| util.assert_return(False, msg, client.schema_change_modify_column, table_name, 'k2', 'string') |
| msg = 'Can not change ARRAY to ARRAY' |
| util.assert_return(False, msg, client.schema_change_modify_column, table_name, 'k2', 'array<string>') |
| client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_boolean(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_boolean", |
| "describe": "array类型，修改array<boolean>类型列为其他array子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['tinyint', 'smallint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_boolean_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_boolean_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_boolean_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_boolean_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_boolean_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_tinyint(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_tinyint", |
| "describe": "array类型，修改array<tinyint>类型列为其他array子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'smallint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_tinyint_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_tinyint_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_tinyint_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_tinyint_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_tinyint_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_smallint(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_smallint", |
| "describe": "array类型，修改array<smallint>类型列为array其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| column_type_list = ['int', 'bigint'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_smallint_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_smallint_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_smallint_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_smallint_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_smallint_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_int(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_int", |
| "describe": "array类型，修改array<int>类型列为array其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_int_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_int_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_int_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_int_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_int_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_bigint(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_bigint", |
| "describe": "array类型，修改array<bigint>类型列为其他array子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'largeint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_bigint_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_bigint_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_bigint_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_bigint_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_bigint_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_largeint(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_largeint", |
| "describe": "array类型，修改array<largeint>类型列为其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'bigint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_largeint_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_largeint_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_largeint_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_largeint_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_largeint_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_decimal(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_decimal", |
| "describe": "array类型，修改array<decimal>类型列为array其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'bigint', 'largeint', |
| 'double', 'float', 'date', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_decimal_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_decimal_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_decimal_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_decimal_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_decimal_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_double(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_double", |
| "describe": "array类型，修改array<double>类型列为array其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'float', 'date', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_double_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_double_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_double_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_double_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_double_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_float(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_float", |
| "describe": "array类型，修改array<double>类型列为其他array子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'date', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_float_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_float_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_float_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_float_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_float_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_date(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_date", |
| "describe": "array类型，修改array<date>类型列为array其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'float', |
| 'datetime', 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_date_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_date_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_date_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_date_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_date_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_datetime(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_datetime", |
| "describe": "array类型，修改array<datetime>类型列为array其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'char(10)', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_datetime_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_datetime_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_datetime_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_datetime_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_datetime_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_char(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_char", |
| "describe": "array类型，修改array<char>类型列为array其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'datetime', 'varchar(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_char_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_char_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_char_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_char_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_char_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_varchar(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_varchar", |
| "describe": "array类型，修改array<varchar>类型列为array其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'datetime', 'char(10)', 'string'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_varchar_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_varchar_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_varchar_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_varchar_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_varchar_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| @pytest.mark.skip() |
| def test_modify_array_column_type_array_string(): |
| """ |
| { |
| "title": "test_modify_array_column_type_array_string", |
| "describe": "array类型，修改array<string>类型列为array其他子类型列", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_type_list = ['boolean', 'tinyint', 'smallint', 'int', 'bigint', 'largeint', |
| 'deical(9, 3)', 'double', 'float', 'date', |
| 'datetime', 'char(10)', 'varchar(10)'] |
| |
| ret = client.create_table(table_name, SCHEMA.array_string_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_string_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| c_type = 'array<%s>' |
| k = 'k2' |
| for sub_type in column_type_list: |
| tb = table_name[-50:] + '_' + sub_type.split('(')[0] |
| ret = client.create_table(tb, SCHEMA.array_string_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(tb, FILE.test_array_string_local_file, max_filter_ratio=0.01) |
| assert ret, 'stream load failed' |
| ret = client.schema_change_modify_column(tb, k, c_type % sub_type, is_wait_job=True) |
| assert ret, 'modify table failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, tb) |
| sql2 = 'select k1, cast(%s as %s) from %s.%s order by k1' % (k, sub_type, database_name, table_name) |
| common.check2(client, sql1, sql2=sql2) |
| ret = client.stream_load(tb, FILE.test_array_string_local_file, max_filter_ratio=0.01) |
| assert ret, 'after modify stream load failed' |
| sql1 = 'select count(*) from %s.%s' % (database_name, tb) |
| sql2 = 'select count(*) * 2 from %s.%s' % (database_name, table_name) |
| print(client.execute(sql1)) |
| # common.check2(client, sql1, sql2=sql2) |
| # client.clean(database_name) |
| |
| |
| def test_modify_array_column_order(): |
| """ |
| { |
| "title": "test_modify_array_column_order", |
| "describe": "array类型，修改表的array类型列的顺序", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| ret = client.create_table(table_name, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_table_local_file, column_separator='|') |
| assert ret, 'stream load failed' |
| column_name_list = ['k1', 'a14', 'a13', 'a12', 'a11', 'a10', 'a9', |
| 'a8', 'a7', 'a6', 'a5', 'a4', 'a3', 'a2', 'a1'] |
| ret = client.schema_change_order_column(table_name, column_name_list, is_wait_job=True) |
| assert ret, 'modify column order failed' |
| ret = client.desc_table(table_name) |
| actual_column = util.get_attr(ret, palo_job.DescInfo.Field) |
| assert actual_column == column_name_list, 'column order error, expect %s, actual %s' \ |
| % (column_name_list, actual_column) |
| sql1 = 'select * from %s.%s order by k1' % (database_name, table_name) |
| sql2 = 'select %s from %s.%s order by k1' % (', '.join(column_name_list), check_db, check_tb) |
| common.check2(client, sql1, sql2=sql2) |
| |
| column_name_list = util.get_attr(SCHEMA.array_table_list, 0) |
| ret = client.stream_load(table_name, FILE.test_array_table_local_file, column_separator='|', |
| column_name_list=column_name_list) |
| assert ret, 'stream load failed' |
| time.sleep(5) |
| sql1 = 'select count(*) from %s.%s' % (database_name, table_name) |
| sql2 = 'select count(*) * 2 from %s.%s' % (check_db, check_tb) |
| common.check2(client, sql1, sql2=sql2) |
| client.clean(database_name) |
| |
| |
| def test_array_bloom_filter(): |
| """ |
| { |
| "title": "test_array_bloom_filter", |
| "describe": "array类型列作为bloom filter, 报错", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| bf_column = ['k1', 'a2'] |
| msg = 'ARRAY is not supported in bloom filter index. invalid column: a2' |
| util.assert_return(False, msg, client.create_table, table_name, SCHEMA.array_table_list, |
| keys_desc=SCHEMA.duplicate_key, bloom_filter_column_list=bf_column) |
| ret = client.create_table(table_name, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_table_local_file, column_separator='|') |
| assert ret, 'stream load failed' |
| util.assert_return(False, msg, client.schema_change, table_name, bloom_filter_column_list=bf_column) |
| client.clean(database_name) |
| |
| |
| def test_array_mv(): |
| """ |
| { |
| "title": "test_array_mv", |
| "describe": "array类型不支持创建物化视图, k1, a2, group by, order by, sum/min/max", |
| "tag": "function,p0,fuzz" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column_list = [('k0', 'int')] + SCHEMA.array_table_list |
| column_name_list = util.get_attr(SCHEMA.array_table_list, 0) + ['k0=k1+1'] |
| ret = client.create_table(table_name, column_list, keys_desc='DUPLICATE KEY(k0, k1)') |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_table_local_file, column_separator='|', |
| column_name_list=column_name_list) |
| assert ret, 'stream load failed' |
| # array, without order by/group by |
| index_name = index_name + '1' |
| sql = 'select k1, a2 from %s order by k1' % table_name |
| msg = ' The ARRAY column[`mv_a2` array<tinyint(4)> NOT NULL] not support to create materialized view' |
| util.assert_return(False, msg, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| index_name = index_name + '2' |
| sql = 'select k1, a3 from %s' % table_name |
| msg = 'The ARRAY column[`mv_a3` array<smallint(6)> NOT NULL] not support to create materialized view' |
| util.assert_return(False, msg, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| |
| index_name = 'err_idx' |
| # order by array |
| sql = 'select k1, a4 from %s order by k1, a4' % table_name |
| msg = "must use with specific function, and don't support filter or group by." |
| util.assert_return(False, msg, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| # group by array |
| sql = 'select k1, a5 from %s group by k1, a5' % table_name |
| util.assert_return(False, msg, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| # count array |
| sql = 'select k1, count(a6) from %s group by k1' % table_name |
| util.assert_return(False, msg, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| # sum array |
| msg_s = 'sum requires a numeric parameter: sum(`a7`)' |
| sql = 'select k1, sum(a7) from %s group by k1' % table_name |
| util.assert_return(False, msg_s, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| # max array |
| sql = 'select k1, max(a8) from %s group by k1' % table_name |
| util.assert_return(False, msg, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| # min array |
| sql = 'select k1, min(a9) from %s group by k1' % table_name |
| util.assert_return(False, msg, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| # bitmap_union |
| msg_b = 'No matching function with signature: to_bitmap_with_check(array<date>).' |
| sql = 'select k1, bitmap_union(to_bitmap(a10)) from %s group by k1' % table_name |
| util.assert_return(False, msg_b, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| # hll_union |
| msg_h = 'No matching function with signature: hll_hash(array<datetime>).' |
| sql = 'select k1, hll_union_agg(hll_hash(a11)) from %s group by k1' % table_name |
| util.assert_return(False, msg_h, client.create_materialized_view, table_name, index_name, sql, is_wait=True) |
| client.clean(database_name) |
| |
| |
| def test_array_export(): |
| """ |
| { |
| "title": "test_array_export", |
| "describe": "array类型，导出成功", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| ret = client.create_table(table_name, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| ret = client.stream_load(table_name, FILE.test_array_table_local_file, column_separator='|') |
| assert ret, 'stream load failed' |
| property_dict = {'column_separator': '|'} |
| ret = client.export(table_name, FILE.export_to_hdfs_path, broker_info=broker_info, property_dict=property_dict) |
| assert ret, 'export failed' |
| client.wait_export() |
| ret = client.show_export(state="FINISHED") |
| assert len(ret) == 1 |
| client.clean(database_name) |
| |
| |
| def test_array_outfile_csv(): |
| """ |
| { |
| "title": "test_array_outfile_csv", |
| "describe": "array类型，查询结果导出为csv", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| ret = client.create_table(table_name, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| load_data_info = palo_client.LoadDataInfo(FILE.test_array_table_remote_file, table_name, column_terminator='|') |
| ret = client.batch_load(util.get_label(), load_data_info, broker=broker_info, max_filter_ratio=0.01, is_wait=True) |
| assert ret, 'broker load failed' |
| |
| sql = 'select * from %s' % table_name |
| csv_output_path = palo_config.gen_remote_file_path('export/%s/%s/%s') % (database_name, util.get_label(), |
| util.get_label()) |
| ret = client.select_into(sql, csv_output_path, broker_info, format_as='csv') |
| print(ret) |
| assert ret, 'select into csv failed' |
| |
| csv_check_table = table_name + '_csv' |
| csv_load_file = str(palo_job.SelectIntoInfo(ret[0]).get_url() + '*') |
| ret = client.create_table(csv_check_table, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| load_data_list = palo_client.LoadDataInfo(csv_load_file, csv_check_table) |
| ret = client.batch_load(util.get_label(), load_data_list, broker=broker_info, is_wait=True) |
| assert ret, 'csv outfile load failed' |
| |
| for k in (1, 2, 3, 4, 5, 6, 7, 10, 11, 12, 13, 14): |
| sql1 = 'select a%s from %s order by k1' % (k, table_name) |
| sql2 = 'select a%s from %s order by k1' % (k, csv_check_table) |
| common.check2(client, sql1=sql1, sql2=sql2, forced=True) |
| # array double & float 精度与原数据不一致，使用array_sum用作校验 |
| for k in (8, 9): |
| sql1 = 'select size(a%s), array_sum(a%s) from %s order by k1' % (k, k, table_name) |
| sql2 = 'select size(a%s), array_sum(a%s) from %s order by k1' % (k, k, csv_check_table) |
| common.check2(client, sql1=sql1, sql2=sql2, forced=True) |
| client.clean(database_name) |
| |
| |
| def test_array_outfile_format(): |
| """ |
| { |
| "title": "test_array_outfile_format", |
| "describe": "array类型，导出格式测试，array不支持导出为parquet格式", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| ret = client.create_table(table_name, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| load_data_info = palo_client.LoadDataInfo(FILE.test_array_table_remote_file, table_name, column_terminator='|') |
| ret = client.batch_load(util.get_label(), load_data_info, broker=broker_info, max_filter_ratio=0.01, is_wait=True) |
| assert ret, 'broker load failed' |
| |
| sql = 'select * from %s' % table_name |
| csv_output_path = palo_config.gen_remote_file_path('export/%s/%s/%s') % (database_name, util.get_label(), |
| util.get_label()) |
| ret = client.select_into(sql, csv_output_path, broker_info, format_as='csv_with_names') |
| print(ret) |
| assert ret, 'select into csv failed' |
| |
| csv_output_path = palo_config.gen_remote_file_path('export/%s/%s/%s') % (database_name, util.get_label(), |
| util.get_label()) |
| ret = client.select_into(sql, csv_output_path, broker_info, format_as='csv_with_names_and_types') |
| print(ret) |
| assert ret, 'select into csv failed' |
| |
| # not support outfile parquet |
| parquet_output_path = palo_config.gen_remote_file_path('export/%s/%s/%s') % (database_name, util.get_label(), |
| util.get_label()) |
| msg = 'currently parquet do not support column type: ARRAY' |
| util.assert_return(False, msg, client.select_into, sql, parquet_output_path, broker_info, format_as='parquet') |
| client.clean(database_name) |
| |
| |
| def test_array_delete(): |
| """ |
| { |
| "title": "test_array_delete", |
| "describe": "array类型删除，删除后数据正确", |
| "tag": "function,p0" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| ret = client.create_table(table_name, SCHEMA.array_table_list, keys_desc=SCHEMA.duplicate_key) |
| assert ret, 'create table failed' |
| load_data_info = palo_client.LoadDataInfo(FILE.test_array_table_remote_file, table_name, column_terminator='|') |
| ret = client.batch_load(util.get_label(), load_data_info, broker=broker_info, max_filter_ratio=0.01, is_wait=True) |
| assert ret, 'broker load failed' |
| ret = client.delete(table_name, 'k1 > 0') |
| assert ret, 'delete failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, table_name) |
| sql2 = 'select * from %s.%s where k1<= 0 order by k1' % (check_db, check_tb) |
| common.check2(client, sql1=sql1, sql2=sql2) |
| ret = client.delete(table_name, 'a1 is null') |
| assert ret, 'delete failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, table_name) |
| sql2 = 'select * from %s.%s where k1<= 0 order by k1' % (check_db, check_tb) |
| common.check2(client, sql1=sql1, sql2=sql2) |
| ret = client.delete(table_name, 'a1 is not null and k1 > -10') |
| assert ret, 'delete failed' |
| sql1 = 'select * from %s.%s order by k1' % (database_name, table_name) |
| sql2 = 'select * from %s.%s where k1<= -10 order by k1' % (check_db, check_tb) |
| common.check2(client, sql1=sql1, sql2=sql2) |
| client.truncate(table_name) |
| ret = client.select_all(table_name) |
| assert ret == (), 'expect empty table' |
| client.clean(database_name) |
| |
| |
| if __name__ == '__main__': |
| setup_module() |
| # todo modify |
