| #!/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_ddl.py |
| # @date 2022-08-15 11:09:53 |
| # @brief This file is a test file for array type. |
| # |
| ############################################################################# |
| """ |
| test_array_ddl.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 util |
| from lib import common |
| from lib import palo_job |
| from data import schema as SCHEMA |
| from data import load_file as FILE |
| |
| config = palo_config.config |
| broker_info = palo_config.broker_info |
| |
| |
| 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') |
| |
| |
| def teardown_module(): |
| """teardown""" |
| pass |
| |
| |
| def test_array_basic(): |
| """ |
| { |
| "title": "test_array_basic", |
| "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' |
| assert client.verify(FILE.expe_array_table_file, table_name), 'check data failed' |
| client.clean(database_name) |
| |
| |
| def test_array_key(): |
| """ |
| { |
| "title": "", |
| "describe": "array类型，不支持作为key列", |
| "tag": "function,p1" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column = [("k2", "ARRAY<INT>"), ("k1", "INT")] |
| msg = 'The olap table first column could not be' |
| # with duplicate key |
| util.assert_return(False, msg, client.create_table, table_name, column) |
| column = [("k1", "INT"), ("k2", "ARRAY<INT>")] |
| msg = 'Array can only be used in the non-key column of the duplicate table at present.' |
| # with duplicate key |
| util.assert_return(False, msg, client.create_table, table_name, column, keys_desc='DUPLICATE KEY(k1,k2)') |
| # default dupllicate key |
| util.assert_return(True, '', client.create_table, table_name, column) |
| ret = client.desc_table(table_name) |
| key = util.get_attr_condition_value(ret, palo_job.DescInfo.Field, 'k2', palo_job.DescInfo.Key) |
| assert key == 'false', 'array column can not be key' |
| client.clean(database_name) |
| |
| |
| def test_array_table_model(): |
| """ |
| { |
| "title": "", |
| "describe": "array类型，仅支持duplicate模型，不支持unique表和agg表，不支持agg中任一聚合方式", |
| "tag": "function,p1" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column = [("k1", "INT"), ("k2", "ARRAY<INT>")] |
| # unique table |
| msg = "ARRAY column can't support aggregation REPLACE" |
| util.assert_return(False, msg, client.create_table, table_name, column, keys_desc='UNIQUE KEY(k1)') |
| # duplicate table |
| util.assert_return(True, '', client.create_table, table_name, column, keys_desc='DUPLICATE KEY(k1)') |
| # aggregate table |
| agg_types = ['MAX', 'MIN', 'SUM', 'BITMAP_UNION', 'HLL_UNION'] |
| msg = 'is not compatible with primitive type array<int(11)>' |
| for agg in agg_types: |
| column = [("k1", "INT"), ("k2", "ARRAY<INT> %s" % agg)] |
| util.assert_return(False, msg, client.create_table, table_name, column) |
| agg_types = ['REPLACE', 'REPLACE_IF_NOT_NULL'] |
| msg = "ARRAY column can't support aggregation" |
| for agg in agg_types: |
| column = [("k1", "INT"), ("k2", "ARRAY<INT> %s" % agg)] |
| util.assert_return(False, msg, client.create_table, table_name, column) |
| client.clean(database_name) |
| |
| |
| def test_array_tb_partition(): |
| """ |
| { |
| "title": "", |
| "describe": "array类型，分区表，导入、查询成功", |
| "tag": "function,p1" |
| } |
| """ |
| 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, |
| partition_info=SCHEMA.baseall_tinyint_partition_info, |
| 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' |
| assert client.verify(FILE.expe_array_table_file, table_name), 'check data failed' |
| client.clean(database_name) |
| |
| |
| def test_array_default(): |
| """ |
| { |
| "title": "", |
| "describe": "array类型，设置default支持，null的default值，not null的default值", |
| "tag": "function,p1" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| # array default not null |
| column = [("k1", "INT", "", "88"), ("k2", "ARRAY<INT>", "", "[1,2,3]")] |
| msg = 'Array type column default value only support null' |
| util.assert_return(False, msg, client.create_table, table_name, column, |
| keys_desc="DUPLICATE KEY(k1)", set_null=True) |
| # not null array default null |
| column = [("k1", "INT", "", "99"), ("k2", "ARRAY<INT>", "", None)] |
| msg = 'Can not set null default value to non nullable column: k2' |
| util.assert_return(False, msg, client.create_table, table_name, column, |
| keys_desc="DUPLICATE KEY(k1)", set_null=False) |
| # null array defaylt null |
| ret = client.create_table(table_name, column, keys_desc="DUPLICATE KEY(k1)", |
| set_null=True) |
| assert ret, 'create table failed' |
| sql = 'insert into %s values(1, [1,2,3])' % table_name |
| client.execute(sql) |
| sql = 'insert into %s(k1) values(2), (3), (4)' % table_name |
| client.execute(sql) |
| ret1 = client.select_all(table_name) |
| ret2 = ((1, '[1, 2, 3]'), (2, None), (3, None), (4, None)) |
| util.check(ret1, ret2, True) |
| client.clean(database_name) |
| |
| |
| def test_array_create_like(): |
| """ |
| { |
| "title": "", |
| "describe": "含有array类型的表，使用create table like创建一个空表，导入查询成功", |
| "tag": "function,p1" |
| } |
| """ |
| 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, |
| partition_info=SCHEMA.baseall_tinyint_partition_info, |
| 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' |
| new_tb = 'new_%s' % table_name |
| ret = client.create_table_like(new_tb, table_name) |
| assert ret, 'create table failed' |
| ret = client.select_all(new_tb) |
| assert ret == (), 'expect empty table' |
| ret1 = client.desc_table(table_name) |
| ret2 = client.desc_table(new_tb) |
| util.check(ret1, ret2) |
| |
| ret = client.stream_load(new_tb, FILE.test_array_table_local_file, column_separator='|') |
| assert ret, 'stream load failed' |
| time.sleep(10) |
| assert client.verify(FILE.expe_array_table_file, new_tb), 'check data failed' |
| client.clean(database_name) |
| |
| |
| def test_array_ctas(): |
| """ |
| { |
| "title": "", |
| "describe": "含有array类型的查询结果，创建表，建表成功，数据导入成功", |
| "tag": "function,p1" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| sql1 = 'select 1, 2, 3, array_sort(collect_list(k6)), array_sort(collect_list(k7)), ' \ |
| 'array_sort(collect_list(k5)) from test_query_qa.test' |
| sql = 'create table %s as select 1 k1, 2 k2, 3 k3, collect_list(k6) a1, collect_list(k7) a2, ' \ |
| 'collect_list(k5) a3 from test_query_qa.test' % table_name |
| ret = client.execute(sql) |
| sql2 = 'select k1, k2, k3, array_sort(a1), array_sort(a2), array_sort(a3) from %s.%s' % (database_name, table_name) |
| common.check2(client, sql1, sql2=sql2, forced=True) |
| ret = client.desc_table(table_name) |
| assert "ARRAY<CHAR(5)>" == util.get_attr_condition_value(ret, palo_job.DescInfo.Field, |
| 'a1', palo_job.DescInfo.Type), 'column type error' |
| assert "ARRAY<VARCHAR(20)>" == util.get_attr_condition_value(ret, palo_job.DescInfo.Field, |
| 'a2', palo_job.DescInfo.Type), 'column type error' |
| assert "ARRAY<DECIMALV3(9, 3)>" == util.get_attr_condition_value(ret, palo_job.DescInfo.Field, |
| 'a3', palo_job.DescInfo.Type), 'column type error' |
| |
| client.clean(database_name) |
| |
| |
| def test_array_bucket_column(): |
| """ |
| { |
| "title": "", |
| "describe": "使用array作为bucket列，建表失败", |
| "tag": "function,p1" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| distribute = palo_client.DistributionInfo('hash(a2)', 1) |
| msg = 'Array Type should not be used in distribution column[a2]' |
| util.assert_return(False, msg, client.create_table, table_name, SCHEMA.array_table_list, |
| partition_info=SCHEMA.baseall_tinyint_partition_info, |
| keys_desc=SCHEMA.duplicate_key, distribution_info=distribute) |
| client.clean(database_name) |
| |
| |
| def test_array_nest(): |
| """ |
| { |
| "title": "", |
| "describe": "array嵌套，仅验证建表，导入查询", |
| "tag": "function,p2" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| column = [("k1", "INT"), ("k2", "ARRAY<ARRAY<INT>>")] |
| ret = client.create_table(table_name, column, keys_desc='DUPLICATE KEY(k1)') |
| assert ret, 'create table failed' |
| sql = 'insert into %s values(1, [[1, 2, 3], [-1, -2, -3]])' % table_name |
| client.execute(sql) |
| sql = 'insert into %s values(2, [[3, 4, 5], [-1, 2, 3]])' % table_name |
| client.execute(sql) |
| ret1 = client.select_all(table_name) |
| ret2 = ((1, '[[1, 2, 3], [-1, -2, -3]]'), (2, '[[3, 4, 5], [-1, 2, 3]]')) |
| util.check(ret1, ret2, True) |
| ret = client.desc_table(table_name) |
| assert "ARRAY<ARRAY<INT(11)>>" == \ |
| util.get_attr_condition_value(ret, palo_job.DescInfo.Field, 'k2', palo_job.DescInfo.Type) |
| client.clean(database_name) |
| |
| |
| def test_array_external_table(): |
| """ |
| { |
| "title": "", |
| "describe": "创建含有array类型的外部表（hdfs csv），创建查询成功", |
| "tag": "function,p1" |
| } |
| """ |
| database_name, table_name, index_name = util.gen_name_list() |
| client = common.create_workspace(database_name) |
| property = {"broker_name": config.broker_name, |
| "path": FILE.test_array_table_remote_file, |
| "column_separator": "|", |
| "format": "csv"} |
| msg = 'Array can only be used in the non-key column of the duplicate table at present' |
| util.assert_return(False, msg, client.create_external_table, table_name, SCHEMA.array_table_list, |
| engine='broker', property=property, broker_property=config.broker_property) |
| # ret = client.create_external_table(table_name, SCHEMA.array_table_list, engine='broker', |
| # property=property, broker_property=config.broker_property) |
| # assert ret, 'create array external table failed' |
| # assert client.verify(FILE.expe_array_table_file, table_name), 'check data failed' |
| client.clean(database_name) |
| |
| |
| if __name__ == '__main__': |
| setup_module() |
