test_sys_pull_load_hdfs.py

#!/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_sys_pull_load.py
#   @date 2015/05/13 15:26:21
#   @brief This file is a test file for palo data loading and verifying.
#
#############################################################################

"""
复合分区，导入操作测试
"""

import pytest
from data import pull_load_apache as DATA
from data import special as SPECIAL_DATA
from data import partition as PARTITION_DATA
from lib import palo_config
from lib import palo_client
from lib import util

LOG = palo_client.LOG
L = palo_client.L
config = palo_config.config
broker_info = palo_config.broker_info


def setup_module():
    """
    setUp
    """
    global client
    client = palo_client.get_client(config.fe_host, config.fe_query_port, user=config.fe_user,
                                    password=config.fe_password, http_port=config.fe_http_port)


def test_all_types():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types",
    "describe": "所有所有类型的随机生成数据的正确性测试,largeint类型测试,largeint是一种比较特殊的类型,4种聚合方式测试,所有支持类型的sum、replace、max、min聚合方式测试,key、value测试,支持的类型做key列和所有类型做value列的情况测试",
    "tag": "function,p1"
    }
    """
    """
    所有所有类型的随机生成数据的正确性测试
    largeint类型测试：largeint是一种比较特殊的类型
    4种聚合方式测试：所有支持类型的sum、replace、max、min聚合方式测试
    key、value测试：支持的类型做key列和所有类型做value列的情况测试
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1, table_name)
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_1, table_name)
    client.clean(database_name)


def test_all_noexsit():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_noexsit",
    "describe": "测试当指定不存在的文件时，导入失败",
    "tag": "function,p1,fuzz"
    }
    """
    """
    测试当指定不存在的文件时，导入失败
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_noexsit, table_name)
    assert not client.batch_load(util.get_label(), data_desc_list_1,
                                 is_wait=True, broker=broker_info)
    client.clean(database_name)


def test_all_types_gz():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types_gz",
    "describe": "所有类型的随机生成数据的正确性测试，导入文件格式为gz",
    "tag": "function,p1"
    }
    """
    """
    所有类型的随机生成数据的正确性测试，导入文件格式为gz
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1_gz, table_name)
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_1, table_name)
    client.clean(database_name)


def test_all_types_bz2():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types_bz2",
    "describe": "所有类型的随机生成数据的正确性测试，导入文件格式为bz2",
    "tag": "function,p1"
    }
    """
    """
    所有类型的随机生成数据的正确性测试，导入文件格式为bz2
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1_bz2, table_name)
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_1, table_name)
    client.clean(database_name)


def test_all_types_lzo():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types_lzo",
    "describe": "所有类型的随机生成数据的正确性测试，导入文件格式为lzo",
    "tag": "function,p1"
    }
    """
    """
    所有类型的随机生成数据的正确性测试，导入文件格式为lzo
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1_lzo, table_name)
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_1, table_name)
    client.clean(database_name)


def test_all_types_lz4():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types_lz4",
    "describe": "所有类型的随机生成数据的正确性测试，导入文件格式为lz4",
    "tag": "function,p1"
    }
    """
    """
    所有类型的随机生成数据的正确性测试，导入文件格式为lz4
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1_lz4, table_name)
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_1, table_name)
    client.clean(database_name)


def test_all_types_parquet():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types_parquet",
    "describe": "test parquet file load",
    "tag": "function,p1"
    }
    """
    """
    test parquet file loading
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1,
                        partition_info, distribution_info, set_null=True)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1_parquet, table_name, format_as='parquet')
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_6, table_name)
    client.clean(database_name)


def test_all_types_parquet_multi():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types_parquet_multi",
    "describe": "test parquet multi load",
    "tag": "function,p1"
    }
    """
    """test parquet multi load"""
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1,
                        partition_info, distribution_info, set_null=True)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1_parquet, table_name, format_as='parquet')
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_6 * 2, table_name)
    client.clean(database_name)


def test_all_types_parquet_format():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types_parquet_format",
    "describe": "test parquet file loading",
    "tag": "function,p1,fuzz"
    }
    """
    """
    test parquet file loading
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1,
                        partition_info, distribution_info, set_null=True)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)
    # error format
    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1_parquet, table_name, format_as='parqut')
    try:
        client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
        assert 0 == 1
    except Exception as e:
        print(str(e))
    # default .parquet
    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1_parquet, table_name)
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_6, table_name)
    client.clean(database_name)


def test_all_types_parquet_null():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types_parquet_null",
    "describe": "test parquet file load",
    "tag": "function,p1"
    }
    """
    """
    test parquet file loading
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1,
                        partition_info, distribution_info, set_null=True)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_parquet_null, table_name, format_as='parquet')
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_7, table_name)
    client.clean(database_name)


def test_all_types_orc():
    """
    {
    "title": "test_all_types_orc",
    "describe": "test orc file load",
    "tag": "function,p1"
    }
    """
    """
    test orc file loading
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1,
                        partition_info, distribution_info, set_null=True)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_1_orc, table_name, format_as='orc')
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_6, table_name)
    client.clean(database_name)


def test_all_types_orc_null():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_types_orc_null",
    "describe": "test orc file load null",
    "tag": "function,p1"
    }
    """
    """
    test orc file load null
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1,
                        partition_info, distribution_info, set_null=True)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_orc_null, table_name, format_as='orc')
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_7, table_name)
    client.clean(database_name)


def test_all_column_list_duo():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_all_column_list_duo",
    "describe": "所有类型的随机生成数据的正确性测试，导入指定列名",
    "tag": "function,p1"
    }
    """
    """
    所有类型的随机生成数据的正确性测试，导入指定列名
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    column_name_list = ['tinyint_key', 'smallint_key', 'int_key',
                        'bigint_key', 'largeint_key', 'char_key', 'varchar_key',
                        'decimal_key', 'date_key', 'datetime_key', 'tinyint_value_max',
                        'smallint_value_min', 'int_value_sum', 'bigint_value_sum',
                        'largeint_value_sum', 'largeint_value_replace', 'char_value_replace',
                        'varchar_value_replace', 'duo', 'decimal_value_replace',
                        'date_value_replace', 'datetime_value_replace',
                        'float_value_sum', 'double_value_sum']

    data_desc_list_1 = palo_client.LoadDataInfo(DATA.data_3, table_name,
                                                column_name_list=column_name_list)
    assert client.batch_load(util.get_label(), data_desc_list_1, is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_3, table_name)
    client.clean(database_name)


def test_more_files():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_more_files",
    "describe": "多数据文件导入一张表测试,多数据文件导入多张表测试,分隔符测试：测试分隔符为x01的情况,column list与数据文件比较少列、表设置默认值的情况测试",
    "tag": "function,p1"
    }
    """
    """
    多数据文件导入一张表测试
    多数据文件导入多张表测试
    分隔符测试：测试分隔符为\x01的情况
    column list与数据文件比较少列、表设置默认值的情况测试
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    table_name_a = table_name + '_a'
    table_name_b = table_name + '_b'
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name_a, DATA.schema_2, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name_a)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name_a, partition_name)
    client.create_table(table_name_b, DATA.schema_2, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name_b)

    column_name_list = ['tinyint_key', 'smallint_key', 'int_key',
                        'bigint_key', 'largeint_key', 'char_key', 'varchar_key',
                        'decimal_key', 'date_key', 'datetime_key', 'tinyint_value_max',
                        'smallint_value_min', 'int_value_sum', 'largeint_value_sum',
                        'float_value_sum', 'double_value_sum']

    data_desc_list_1 = palo_client.LoadDataInfo([DATA.data_4, DATA.data_5],
                                                table_name_a, column_name_list=column_name_list)
    data_desc_list_2 = palo_client.LoadDataInfo([DATA.data_6, DATA.data_7],
                                                table_name_b, column_name_list=column_name_list, 
                                                column_terminator='\x01')
    assert client.batch_load(util.get_label(), [data_desc_list_1,
                                                data_desc_list_2], is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_4, table_name_a)
    vf_a = palo_client.VerifyFile(DATA.verify_5[0], '\x01')
    vf_b = palo_client.VerifyFile(DATA.verify_5[1], '\x01')
    assert client.verify([vf_a, vf_b], table_name_b)
    client.clean(database_name)


def test_automic():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_automic",
    "describe": "原子导入",
    "tag": "function,p1,fuzz"
    }
    """
    """
    原子导入
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    table_name_a = table_name + '_a'
    table_name_b = table_name + '_b'
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name_a, DATA.schema_2, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name_a)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name_a, partition_name)
    client.create_table(table_name_b, DATA.schema_2, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name_b)

    column_name_list = ['tinyint_key', 'smallint_key', 'int_key',
                        'bigint_key', 'largeint_key', 'char_key', 'varchar_key',
                        'decimal_key', 'date_key', 'datetime_key', 'tinyint_value_max',
                        'smallint_value_min', 'int_value_sum', 'largeint_value_sum',
                        'float_value_sum', 'double_value_sum']

    data_desc_list_1 = palo_client.LoadDataInfo([DATA.data_4, DATA.data_5],
                                                table_name_a, column_name_list=column_name_list)
    data_desc_list_2 = palo_client.LoadDataInfo([DATA.data_6, DATA.data_1],
                                                table_name_b, column_name_list=column_name_list, 
                                                column_terminator='\x01')
    assert not client.batch_load(util.get_label(), [data_desc_list_1,
                                                    data_desc_list_2], is_wait=True, 
                                                    broker=broker_info)
    client.clean(database_name)


def test_dir():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_dir",
    "describe": "测试导入指定目录，与通配符一起使用,多数据文件导入一张表测试,多数据文件导入多张表测试,分隔符测试：测试分隔符为x01的情况,column list与数据文件比较少列、表设置默认值的情况测试",
    "tag": "function,p1"
    }
    """
    """
    测试导入指定目录，与通配符一起使用
    多数据文件导入一张表测试
    多数据文件导入多张表测试
    分隔符测试：测试分隔符为\x01的情况
    column list与数据文件比较少列、表设置默认值的情况测试
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    table_name_a = table_name + '_a'
    table_name_b = table_name + '_b'
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name_a, DATA.schema_2, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name_a)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name_a, partition_name)
    client.create_table(table_name_b, DATA.schema_2, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name_b)

    column_name_list = ['tinyint_key', 'smallint_key', 'int_key',
                        'bigint_key', 'largeint_key', 'char_key', 'varchar_key',
                        'decimal_key', 'date_key', 'datetime_key', 'tinyint_value_max',
                        'smallint_value_min', 'int_value_sum', 'largeint_value_sum',
                        'float_value_sum', 'double_value_sum']

    data_desc_list_1 = palo_client.LoadDataInfo(
        DATA.data_8, table_name_a, column_name_list=column_name_list)
    data_desc_list_2 = palo_client.LoadDataInfo(
        DATA.data_9, table_name_b, column_name_list=column_name_list, column_terminator='\x01')
    assert client.batch_load(
        util.get_label(), [data_desc_list_1, data_desc_list_2], is_wait=True, broker=broker_info)
#   assert client.batch_load(util.get_label(), [data_desc_list_1], is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_4, table_name_a)
    vf_a = palo_client.VerifyFile(DATA.verify_5[0], '\x01')
    vf_b = palo_client.VerifyFile(DATA.verify_5[1], '\x01')
    assert client.verify([vf_a, vf_b], table_name_b)
    client.clean(database_name)


# 接下来的几个case是：特殊值、边界值测试：各种int类型的最大最小值、varchar的空串、最大长度串的测试
def test_special_int_valid_hash_column():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_special_int_valid_hash_column",
    "describe": "功能点：int类型特殊值, hash分区, column存储",
    "tag": "function,p1,fuzz"
    }
    """
    """
    功能点：int类型特殊值, hash分区, column存储
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.create_table(table_name, SPECIAL_DATA.schema_1,
                        keys_desc='AGGREGATE KEY (tinyint_key, smallint_key)')

    assert client.show_tables(table_name)
    assert client.get_index(table_name)

    data_desc_list = palo_client.LoadDataInfo(SPECIAL_DATA.file_path_1, table_name)
    ret = client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
    assert ret

    ret = client.verify(SPECIAL_DATA.expected_data_file_list_1, table_name)
    assert ret
    client.clean(database_name)


def test_special_decimal_valid_hash_column():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_special_decimal_valid_hash_column",
    "describe": "功能点：decimal类型特殊值, hash分区, column存储",
    "tag": "function,p1,fuzz"
    }
    """
    """
    功能点：decimal类型特殊值, hash分区, column存储
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.create_table(table_name, SPECIAL_DATA.schema_2, keys_desc='AGGREGATE KEY (tinyint_key)')

    assert client.show_tables(table_name)
    assert client.get_index(table_name)

    data_desc_list = palo_client.LoadDataInfo(SPECIAL_DATA.file_path_2, table_name)
    ret = client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
    assert ret

    ret = client.verify(SPECIAL_DATA.expected_data_file_list_2, table_name)
    assert ret
    client.clean(database_name)


def test_special_max_decimal_valid_hash_column():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_special_max_decimal_valid_hash_column",
    "describe": "功能点：decimal类型特殊值, 最大值, hash分区, column存储",
    "tag": "function,p1,fuzz"
    }
    """
    """
    功能点：decimal类型特殊值, 最大值, hash分区, column存储
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.create_table(table_name, SPECIAL_DATA.schema_3, keys_desc='AGGREGATE KEY (tinyint_key)')

    assert client.show_tables(table_name)
    assert client.get_index(table_name)

    data_desc_list = palo_client.LoadDataInfo(SPECIAL_DATA.file_path_3, table_name)
    ret = client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
    assert ret

    ret = client.verify(SPECIAL_DATA.expected_data_file_list_3, table_name)
    assert ret
    client.clean(database_name)


def test_special_char_valid_hash_column():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_special_char_valid_hash_column",
    "describe": "功能点：char类型特殊值, hash分区, column存储",
    "tag": "function,p1,fuzz"
    }
    """
    """
    功能点：char类型特殊值, hash分区, column存储
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)
    client.create_table(table_name, SPECIAL_DATA.schema_4, keys_desc='AGGREGATE KEY (tinyint_key)')

    assert client.show_tables(table_name)
    assert client.get_index(table_name)

    data_desc_list = palo_client.LoadDataInfo(SPECIAL_DATA.file_path_4, table_name)
    ret = client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
    assert ret

    ret = client.verify(SPECIAL_DATA.expected_data_file_list_4, table_name)
    assert ret
    client.clean(database_name)


def test_special_char_invalid_hash_column():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_special_char_invalid_hash_column",
    "describe": "功能点：char类型特殊值, invalid, hash分区, column存储",
    "tag": "function,p1,fuzz"
    }
    """
    """
    功能点：char类型特殊值, invalid, hash分区, column存储
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)
    client.create_table(table_name, SPECIAL_DATA.schema_5, keys_desc='AGGREGATE KEY (tinyint_key)')

    assert client.show_tables(table_name)
    assert client.get_index(table_name)

    data_desc_list = palo_client.LoadDataInfo(SPECIAL_DATA.file_path_5, table_name)
    assert not client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)

    assert not client.verify(SPECIAL_DATA.expected_data_file_list_5, table_name)

    assert client.batch_load(util.get_label(), data_desc_list, is_wait=True,
                             max_filter_ratio=0.5, broker=broker_info)
    assert client.batch_load(util.get_label(), data_desc_list, is_wait=True,
                             max_filter_ratio=0.51, broker=broker_info)
    assert not client.batch_load(util.get_label(), data_desc_list,
                                 is_wait=True, max_filter_ratio=0.49, broker=broker_info)
    client.clean(database_name)


def test_load_with_partition():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_load_with_partition",
    "describe": "指定partition：多分区表，测试指定partition进行导入",
    "tag": "function,p1"
    }
    """
    """
    指定partition：多分区表，测试指定partition进行导入
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '31', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('k1', partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(k1, k5)', 13)
    client.create_table(table_name, PARTITION_DATA.schema_1, keys_desc='AGGREGATE KEY (k1, k2, k3, k4, k5)',
                        partition_info=partition_info, distribution_info=distribution_info)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list = palo_client.LoadDataInfo(PARTITION_DATA.file_path_1,
                     table_name, ['partition_a', 'partition_b', 'partition_c'])
    assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
    assert client.verify(PARTITION_DATA.expected_data_file_list_1, table_name)
    assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
    assert client.verify(list(PARTITION_DATA.expected_data_file_list_1) * 2, table_name)
    client.clean(database_name)


def test_partition_timeout_and_cancel():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_partition_timeout_and_cancel",
    "describe": "test timeout and cancel",
    "tag": "function,p1,fuzz"
    }
    """
    """
    多分区表，大数据量
    并行导入
    数据量大于1G，并行ETL
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)
    client.use(database_name)

    partition_name_list = ['partition_a', 'partition_b',
                           'partition_c', 'partition_d', 'partition_e']
    partition_value_list = ['1000000', '2000000000', '3000000000', '500000000000000', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('largeint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name, DATA.schema_1, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name, partition_name)

    data_desc_list_1 = palo_client.LoadDataInfo(
        DATA.data_2, table_name, ['partition_a', 'partition_b', 'partition_c'])
    data_desc_list_2 = palo_client.LoadDataInfo(
        DATA.data_2, table_name, ['partition_d', 'partition_e'])
    load_label_1 = util.get_label()
    client.batch_load(load_label_1, data_desc_list_1,
                      broker=broker_info, max_filter_ratio=1, timeout=5)
    load_label_2 = util.get_label()
    client.batch_load(load_label_2, data_desc_list_2, broker=broker_info, max_filter_ratio=1)
    client.cancel_load(load_label_2, database_name)
    assert not client.wait_load_job(load_label_1, database_name, cluster_name='test_cluster')
    assert not client.wait_load_job(load_label_2, database_name, cluster_name='test_cluster')
    client.clean(database_name)


def test_hunhe():
    """
    {
    "title": "test_sys_pull_load_hdfs.test_hunhe",
    "describe": "d多数据文件导入一张表测试,多数据文件导入多张表测试,column list与数据文件比较少列、表设置默认值的情况测试",
    "tag": "function,p1"
    }
    """
    """
    多数据文件导入一张表测试
    多数据文件导入多张表测试
    分隔符测试：测试分隔符为\x01的情况
    column list与数据文件比较少列、表设置默认值的情况测试
    """
    database_name, table_name, index_name = util.gen_num_format_name_list()
    table_name_a = table_name + '_a'
    table_name_b = table_name + '_b'
    LOG.info(L('', database_name=database_name,
               table_name=table_name, index_name=index_name))
    client.use(database_name)
    client.clean(database_name)
    client.create_database(database_name)

    partition_name_list = ['partition_a', 'partition_b', 'partition_c', 'partition_d']
    partition_value_list = ['5', '20', '30', 'MAXVALUE']

    partition_info = palo_client.PartitionInfo('tinyint_key',
                                               partition_name_list, partition_value_list)
    distribution_info = palo_client.DistributionInfo('HASH(largeint_key, decimal_key)', 13)
    client.create_table(table_name_a, DATA.schema_2, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name_a)
    for partition_name in partition_name_list:
        assert client.get_partition(table_name_a, partition_name)
    client.create_table(table_name_b, DATA.schema_2, 
                        partition_info, distribution_info)
    assert client.show_tables(table_name_b)

    column_name_list = ['tinyint_key', 'smallint_key', 'int_key', \
            'bigint_key', 'largeint_key', 'char_key', 'varchar_key', \
            'decimal_key', 'date_key', 'datetime_key', 'tinyint_value_max', \
            'smallint_value_min', 'int_value_sum', 'largeint_value_sum', \
            'float_value_sum', 'double_value_sum']

    data_desc_list_1 = palo_client.LoadDataInfo(
        [DATA.data_4, DATA.data_5_gz], table_name_a, column_name_list=column_name_list)
    data_desc_list_2 = palo_client.LoadDataInfo(
        [DATA.data_6_lzo, DATA.data_7_bz2], table_name_b, 
        column_name_list=column_name_list, column_terminator='\x01')
    assert client.batch_load(
        util.get_label(), [data_desc_list_1, data_desc_list_2], is_wait=True, broker=broker_info)
    assert client.verify(DATA.verify_4, table_name_a)
    vf_a = palo_client.VerifyFile(DATA.verify_5[0], '\x01')
    vf_b = palo_client.VerifyFile(DATA.verify_5[1], '\x01')
    assert client.verify([vf_a, vf_b], table_name_b)
    client.clean(database_name)


def teardown_module():
    """
    tearDown
    """
    pass


if __name__ == '__main__':
    setup_module()
    # test_all_types_orc()
    test_all_types_orc_null()
    test_all_types_parquet_null()