1316 lines
56 KiB
Python
1316 lines
56 KiB
Python
#!/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_storage_medium.py
|
|
# @date 2020-05-07 14:48:30
|
|
# @brief 存储介质基本case。需要使用env_config.py中的be数据目录及机器登录信息
|
|
#
|
|
#############################################################################
|
|
|
|
"""
|
|
分级存储测试
|
|
"""
|
|
import time
|
|
import datetime
|
|
import pytest
|
|
|
|
from data import partition as DATA
|
|
from data import schema as SCHEMA
|
|
from lib import palo_config
|
|
from lib import palo_client
|
|
from lib import palo_job
|
|
from lib import util
|
|
|
|
config = palo_config.config
|
|
LOG = palo_client.LOG
|
|
L = palo_client.L
|
|
broker_info = palo_config.broker_info
|
|
BE_DATA_PATH_EXISTS=False
|
|
|
|
|
|
def setup_module():
|
|
"""
|
|
setUp
|
|
"""
|
|
global client, BE_DATA_PATH_EXISTS
|
|
client = palo_client.get_client(config.fe_host, config.fe_query_port, user=config.fe_user, \
|
|
password=config.fe_password)
|
|
be_host = client.get_backend_host_list()
|
|
for host in be_host:
|
|
cmd = 'ls %s' % ' '.join(config.be_data_path)
|
|
status, output = util.exec_cmd(cmd, config.host_username, config.host_password, host)
|
|
if status == 0:
|
|
BE_DATA_PATH_EXISTS = True
|
|
|
|
|
|
def check_tablet(table_tablet_info, storage_medium):
|
|
"""
|
|
检查tablet是否在be数据目录的ssd中
|
|
table_tablet_info:show tablet result
|
|
"""
|
|
if not BE_DATA_PATH_EXISTS:
|
|
return 0
|
|
storage_path_list = list()
|
|
for data_path in config.be_data_path:
|
|
if data_path.endswith(storage_medium.upper()):
|
|
storage_path_list.append(data_path + '/data/')
|
|
for tablet in table_tablet_info:
|
|
tablet_id = tablet[0]
|
|
host = client.get_backend(tablet[2])[palo_job.BackendProcInfo.Host]
|
|
for data_path in storage_path_list:
|
|
cmd = 'find %s -type d -name %s ' % (data_path, tablet_id)
|
|
status, output = util.exec_cmd(cmd, config.host_username, config.host_password, host)
|
|
print(cmd)
|
|
print(status, output)
|
|
LOG.info(L('exec cmd on remote host', cmd=cmd, username=config.host_username, host=host))
|
|
LOG.info(L('exec cmd on remote host ret', status=status, output=output))
|
|
assert status == 0
|
|
assert output.find(tablet_id) != -1
|
|
|
|
|
|
def test_ssd_all():
|
|
"""
|
|
{
|
|
"title": "存储介质SSD测试",
|
|
"describe": "复合分区表建表时指定全部分区到SSD,验证BE在SSD上建立了存储分片,导入数据后, 验证数据正确分布到了SSD，并对数据进行正确性验证",
|
|
"tag": "p1,system"
|
|
}
|
|
"""
|
|
"""
|
|
复合分区表建表时指定全部分区到SSD
|
|
验证BE在SSD上建立了存储分片
|
|
导入数据后, 验证数据正确分布到了SSD，并对数据进行正确性验证
|
|
"""
|
|
database_name, table_name, index_name = util.gen_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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='SSD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='SSD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='SSD')
|
|
partition_d = palo_client.Partition('partition_d', 'MAXVALUE', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1, partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
ret = client.show_tablet(table_name)
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_hdd_all():
|
|
"""
|
|
{
|
|
"title": "建表指定HDD",
|
|
"describe": "复合分区表建表时指定全部分区到HDD，验证BE在HDD正确建立了存储分片，导入数据后，验证相应分区的数据正确分布到了HDD，并对数据进行正确性验证",
|
|
"tag": "p1,system"
|
|
}
|
|
"""
|
|
"""
|
|
复合分区表建表时指定全部分区到HDD
|
|
验证BE在HDD正确建立了存储分片
|
|
导入数据后，验证相应分区的数据正确分布到了HDD，并对数据进行正确性验证
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', 'MAXVALUE', storage_medium='HDD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k2)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
ret = client.show_tablet(table_name)
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_ssd_hdd():
|
|
"""
|
|
{
|
|
"title": "建分区表，ssd和hdd混合",
|
|
"describe": "复合分区表建表时指定部分分区到SSD，指定其余分区到HDD",
|
|
"tag": "p1,system"
|
|
}
|
|
"""
|
|
"""
|
|
复合分区表建表时指定部分分区到SSD，指定其余分区到HDD
|
|
并且保证所有BE的SSD和HDD空间充足
|
|
验证BE在SSD和HDD建立了存储分片
|
|
导入数据后，验证相应分区的数据正确分布到了SSD和HDD，并对数据进行正确性验证
|
|
"""
|
|
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.use(database_name)
|
|
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='SSD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='SSD')
|
|
partition_d = palo_client.Partition('partition_d', 'MAXVALUE', storage_medium='HDD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k3)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
ret = client.show_tablet(table_name, partition_list=['partition_a', 'partition_c'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
|
|
ret = client.show_tablet(table_name, partition_list=['partition_b', 'partition_d'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_ssd_table():
|
|
"""
|
|
{
|
|
"title": "建立ssd单分区表",
|
|
"describe": "单分区表建表时指定到SSD，验证BE在SSD上建立了存储分片，导入数据后，验证数据正确分布到了SSD，并对数据进行正确性验证",
|
|
"tag": "p1,system"
|
|
}
|
|
"""
|
|
"""
|
|
单分区表建表时指定到SSD，
|
|
验证BE在SSD上建立了存储分片
|
|
导入数据后，验证数据正确分布到了SSD，并对数据进行正确性验证
|
|
"""
|
|
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.use(database_name)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
distribution_info=distribution_info, storage_medium='SSD')
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, table_name) == 'SSD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
ret = client.show_tablet(table_name)
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_hdd_table():
|
|
"""
|
|
{
|
|
"title": "建立hdd单分区表",
|
|
"describe": "单分区表建表时指定到HDD，验证BE在HDD上建立了存储分片，导入数据后，验证数据正确分布到了HDD，并对数据进行正确性验证",
|
|
"tag": "p1,system"
|
|
}
|
|
"""
|
|
"""
|
|
单分区表建表时指定到HDD，
|
|
验证BE在HDD上建立了存储分片
|
|
导入数据后，验证数据正确分布到了HDD，并对数据进行正确性验证
|
|
"""
|
|
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.use(database_name)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
distribution_info=distribution_info, storage_medium='HDD')
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, table_name) == 'HDD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
ret = client.show_tablet(table_name)
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_add_ssd_partition():
|
|
"""
|
|
{
|
|
"title": "增加ssd分区",
|
|
"describe": "复合分区表，增加分区指定到SSD，同时修改BUCKETS个数，验证BE在SSD上建立了新的分区的存储分片，导入新增分区的数据后，验证数据分布到SSD，正确性验证",
|
|
"tag": "p1,system"
|
|
}
|
|
"""
|
|
"""
|
|
复合分区表，增加分区指定到SSD，同时修改BUCKETS个数
|
|
验证BE在SSD上建立了新的分区的存储分片
|
|
导入新增分区的数据后，验证数据正确分布到了SSD，并对数据进行正确性验证
|
|
"""
|
|
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.use(database_name)
|
|
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='HDD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k2)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
|
|
ret = client.show_tablet(table_name)
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
|
|
|
|
assert client.add_partition(table_name, 'partition_add', 'MAXVALUE',
|
|
storage_medium='SSD', distribute_type='Hash(k2)',
|
|
bucket_num=13)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_add') == 'SSD'
|
|
|
|
ret = client.show_tablet(table_name, partition_list=['partition_add'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_add_hdd_partition():
|
|
"""
|
|
{
|
|
"title": "增加hdd分区",
|
|
"describe": "复合分区表，增加分区指定到HDD，同时修改BUCKETS个数,验证BE在SSD上建立了新的分区的存储分片,导入新增分区的数据后，验证数据正确分布到了HDD，并对数据进行正确性验证",
|
|
"tag": "autotest"
|
|
}
|
|
"""
|
|
"""
|
|
复合分区表，增加分区指定到HDD，同时修改BUCKETS个数
|
|
验证BE在SSD上建立了新的分区的存储分片
|
|
导入新增分区的数据后，验证数据正确分布到了HDD，并对数据进行正确性验证
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='SSD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='SSD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='SSD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 1)
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
ret = client.show_tablet(table_name)
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
assert client.add_partition(table_name, 'partition_add', 'MAXVALUE',
|
|
storage_medium='HDD', distribute_type='Hash(k1)', bucket_num=13)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_add') == 'HDD'
|
|
ret = client.show_tablet(table_name, partition_list=['partition_add'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
time.sleep(10)
|
|
print(e)
|
|
if retry_times == 0:
|
|
assert 0 == 1, 'storage medium check error'
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_to_hdd_partition():
|
|
"""
|
|
{
|
|
"title": "修改ssd分区为hdd",
|
|
"describe": "主动修改分区存储属性，SSD改为HDD，HDD充足，验证BE汇报后迁移正确，数据正确",
|
|
"tag": "p1,system"
|
|
}
|
|
"""
|
|
"""
|
|
主动修改分区存储属性，SSD改为HDD，HDD充足，验证BE汇报后迁移正确，数据正确
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k3)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
assert client.modify_partition(table_name, 'partition_d', storage_medium='HDD')
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
ret = client.show_tablet(table_name, partition_list=['partition_d'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_to_ssd_partition():
|
|
"""
|
|
{
|
|
"title": "修改hdd分区为ssd",
|
|
"describe": "主动修改分区存储属性，HDD改为SSD，SSD充足，验证BE汇报后迁移正确，数据正确",
|
|
"tag": "p1,system"
|
|
}
|
|
"""
|
|
"""
|
|
主动修改分区存储属性，HDD改为SSD，SSD充足，验证BE汇报后迁移正确，数据正确
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='SSD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='SSD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='SSD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='HDD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k2)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
ret = client.show_tablet(table_name, partition_list=['partition_d'])
|
|
check_tablet(ret, 'hdd')
|
|
assert client.modify_partition(table_name, 'partition_d', storage_medium='SSD')
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
ret = client.show_tablet(table_name, partition_list=['partition_d'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
retry_times -= 1
|
|
time.sleep(30)
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_ssd_to_ssd_partition():
|
|
"""
|
|
{
|
|
"title": "将ssd分区改为ssd",
|
|
"describe": "主动修改分区存储属性，SSD改为SSD，验证不迁移",
|
|
"tag": "p1,system,fuzz"
|
|
}
|
|
"""
|
|
"""
|
|
主动修改分区存储属性，SSD改为SSD，验证不迁移
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='SSD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='SSD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='SSD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'SSD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
|
|
assert client.modify_partition(table_name, 'partition_d', storage_medium='SSD')
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
ret = client.show_tablet(table_name, partition_list=['partition_d'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from ssd')
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_hdd_to_hdd_partition():
|
|
"""
|
|
{
|
|
"title": "修改hdd分区为hdd",
|
|
"describe": "主动修改分区存储属性，HDD改为HDD，验证不迁移",
|
|
"tag": "p1,system,fuzz"
|
|
}
|
|
"""
|
|
"""
|
|
主动修改分区存储属性，HDD改为HDD，验证不迁移
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='HDD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 2)
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
|
|
assert client.modify_partition(table_name, 'partition_d', storage_medium='HDD')
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
ret = client.show_tablet(table_name, partition_list=['partition_d'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_to_hdd_partition_at_cooldown_time():
|
|
"""
|
|
{
|
|
"title": "修改hdd分区冷却时间",
|
|
"describe": "主动修改分区storage_cooldown_time属性，HDD充足，验证属性修改正确，验证过期时间达到后BE汇报后迁移正确",
|
|
"tag": "autotest"
|
|
}
|
|
"""
|
|
"""
|
|
主动修改分区storage_cooldown_time属性，HDD充足
|
|
验证属性修改正确，验证过期时间达到后BE汇报后迁移正确
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k3)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
|
|
storage_cooldown_delta_time = 150
|
|
delta = datetime.timedelta(seconds=storage_cooldown_delta_time)
|
|
storage_cooldown_time = (datetime.datetime.now() + delta).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
assert client.modify_partition(table_name, 'partition_d', storage_medium='SSD',
|
|
storage_cooldown_time=storage_cooldown_time)
|
|
time.sleep(storage_cooldown_delta_time + 30)
|
|
ret = client.show_tablet(table_name, partition_list=['partition_d'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_medium_cooldown_time():
|
|
"""
|
|
{
|
|
"title": "修改hdd分区为ssd，增加冷却时间",
|
|
"describe": "单分区表同时修改分区storage_medium、storage_cooldown_time属性，HDD改为SSD，SSD充足，验证属性修改正确，验证迁移正确",
|
|
"tag": "autotest"
|
|
}
|
|
"""
|
|
"""
|
|
单分区表同时修改分区storage_medium、storage_cooldown_time属性
|
|
HDD改为SSD，SSD充足，验证属性修改正确，验证迁移正确
|
|
"""
|
|
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.use(database_name)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
distribution_info=distribution_info, storage_medium='HDD')
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, table_name) == 'HDD'
|
|
|
|
storage_cooldown_delta_time = 150
|
|
delta = datetime.timedelta(seconds=storage_cooldown_delta_time)
|
|
storage_cooldown_time = (datetime.datetime.now() + delta).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
assert client.modify_partition(table_name, table_name, storage_medium='SSD',
|
|
storage_cooldown_time=storage_cooldown_time)
|
|
|
|
time.sleep(storage_cooldown_delta_time + 30)
|
|
ret = client.show_tablet(table_name)
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
assert client.get_partition_storage_medium(table_name, table_name) == 'HDD'
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_medium_cooldown_time_for_partitions():
|
|
"""
|
|
{
|
|
"title": "修改分区为ssd，修改冷却时间",
|
|
"describe": "复合分区表对多个分区同时修改分区storage_medium、storage_cooldown_time属性,HDD改为SSD，SSD充足，验证属性修改正确，验证迁移正确",
|
|
"tag": "p1,system"
|
|
}
|
|
"""
|
|
"""
|
|
复合分区表对多个分区同时修改分区storage_medium、storage_cooldown_time属性
|
|
HDD改为SSD，SSD充足，验证属性修改正确，验证迁移正确
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k3)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
|
|
storage_cooldown_delta_time = 150
|
|
delta = datetime.timedelta(seconds=storage_cooldown_delta_time)
|
|
storage_cooldown_time = (datetime.datetime.now() + delta).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
assert client.modify_partition(table_name, 'partition_c', storage_medium='SSD',
|
|
storage_cooldown_time=storage_cooldown_time)
|
|
assert client.modify_partition(table_name, 'partition_d', storage_medium='SSD',
|
|
storage_cooldown_time=storage_cooldown_time)
|
|
time.sleep(storage_cooldown_delta_time + 30)
|
|
ret = client.show_tablet(table_name)
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_cooldown_time():
|
|
"""
|
|
{
|
|
"title": "单分区表增加冷却时间",
|
|
"describe": "单分区表SSD指定过期时间，验证过期时间达到后BE汇报后数据迁移到HDD，验证数据正确",
|
|
"tag": "autotest"
|
|
}
|
|
"""
|
|
"""
|
|
单分区表SSD指定过期时间，验证过期时间达到后BE汇报后数据迁移到HDD，验证数据正确
|
|
"""
|
|
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.use(database_name)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
distribution_info=distribution_info, storage_medium='SSD')
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, table_name) == 'SSD'
|
|
|
|
storage_cooldown_delta_time = 150
|
|
delta = datetime.timedelta(seconds=storage_cooldown_delta_time)
|
|
storage_cooldown_time = (datetime.datetime.now() + delta).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
assert client.modify_partition(table_name, table_name, storage_medium='SSD',
|
|
storage_cooldown_time=storage_cooldown_time)
|
|
|
|
time.sleep(storage_cooldown_delta_time + 30)
|
|
ret = client.show_tablet(table_name)
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'hdd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
assert client.get_partition_storage_medium(table_name, table_name) == 'HDD'
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_invalid_cooldown_time_a():
|
|
"""
|
|
{
|
|
"title": "无效的冷却时间",
|
|
"describe": "单分区表SSD指定过去时间为过期时间，验证不迁移，报错",
|
|
"tag": "p1,system,fuzz"
|
|
}
|
|
"""
|
|
"""
|
|
单分区表SSD指定过去时间为过期时间，验证不迁移，报错
|
|
"""
|
|
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.use(database_name)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
distribution_info=distribution_info, storage_medium='SSD')
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, table_name) == 'SSD'
|
|
|
|
storage_cooldown_delta_time = 150
|
|
delta = datetime.timedelta(seconds=storage_cooldown_delta_time)
|
|
storage_cooldown_time = (datetime.datetime.now() - delta).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
ret = False
|
|
try:
|
|
ret = client.modify_partition(table_name, table_name, storage_medium='SSD',
|
|
storage_cooldown_time=storage_cooldown_time)
|
|
except:
|
|
pass
|
|
assert not ret
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_modify_invalid_cooldown_time_b():
|
|
"""
|
|
{
|
|
"title": "无效的冷却时间",
|
|
"describe": "单分区表SSD指定当前时间为过期时间，验证报错",
|
|
"tag": "p1,system,fuzz"
|
|
}
|
|
"""
|
|
"""
|
|
单分区表SSD指定当前时间为过期时间，验证报错
|
|
"""
|
|
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.use(database_name)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
distribution_info=distribution_info, storage_medium='SSD')
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, table_name) == 'SSD'
|
|
|
|
storage_cooldown_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
ret = False
|
|
try:
|
|
ret = client.modify_partition(table_name, table_name, storage_medium='SSD',
|
|
storage_cooldown_time=storage_cooldown_time)
|
|
except:
|
|
pass
|
|
assert not ret
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_cooldown_while_select():
|
|
"""
|
|
{
|
|
"title": "test_cooldown_while_select",
|
|
"describe": "复合分区表全部分区指定到SSD，部分分区指定过期时间，部分分区不指定过期时间过期时间达到前后1分钟内，持续进行查询操作",
|
|
"tag": "p1,system,stability"
|
|
}
|
|
"""
|
|
"""
|
|
复合分区表全部分区指定到SSD，SSD充足
|
|
部分分区指定过期时间，部分分区不指定过期时间
|
|
过期时间达到前后1分钟内，持续进行查询操作
|
|
验证查询结果一直正确，验证指定过期时间的分区的数据迁移到HDD
|
|
"""
|
|
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.use(database_name)
|
|
storage_cooldown_delta_time = 5 * 60
|
|
delta = datetime.timedelta(seconds=storage_cooldown_delta_time)
|
|
storage_cooldown_time = (datetime.datetime.now() + delta).strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
|
|
storage_cooldown_delta_time = 150
|
|
delta = datetime.timedelta(seconds=storage_cooldown_delta_time)
|
|
storage_cooldown = (datetime.datetime.now() + delta)
|
|
storage_cooldown_time = storage_cooldown.strftime('%Y-%m-%d %H:%M:%S')
|
|
|
|
assert client.modify_partition(table_name, 'partition_a', storage_medium='SSD',
|
|
storage_cooldown_time=storage_cooldown_time)
|
|
assert client.modify_partition(table_name, 'partition_c', storage_medium='SSD',
|
|
storage_cooldown_time=storage_cooldown_time)
|
|
|
|
while datetime.datetime.now() < storage_cooldown + delta:
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
time.sleep(5)
|
|
|
|
ret = client.show_tablet(table_name, partition_list=['partition_a', 'partition_c'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_migrate_while_loading():
|
|
"""
|
|
{
|
|
"title": "test_migrate_while_loading",
|
|
"describe": "迁移过程的同时导入，验证迁移正确，验证导入正确",
|
|
"tag": "p1,system,stability"
|
|
}
|
|
"""
|
|
"""
|
|
迁移过程的同时导入，验证迁移正确，验证导入正确
|
|
"""
|
|
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.use(database_name)
|
|
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k1)', 3)
|
|
assert client.create_table(table_name, DATA.schema_1,
|
|
partition_info, distribution_info)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
label = util.get_label()
|
|
assert client.batch_load(label, data_desc_list, broker=broker_info)
|
|
state = client.get_load_job_state(label)
|
|
while state != 'LOADING':
|
|
if state == 'FINISHED' or state == 'CANCELLED':
|
|
raise pytest.skip('can not get loading state')
|
|
time.sleep(1)
|
|
state = client.get_load_job_state(label)
|
|
|
|
assert client.modify_partition(table_name, 'partition_d', storage_medium='HDD')
|
|
time.sleep(30 * 2)
|
|
|
|
assert client.verify(DATA.expected_data_file_list_1, table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_migrate_before_schema_change():
|
|
"""
|
|
{
|
|
"title": "test_migrate_before_schema_change",
|
|
"describe": "迁移过程的同时schema change，验证迁移正确，验证schema change正确",
|
|
"tag": "p1,system,stability"
|
|
}
|
|
"""
|
|
"""
|
|
迁移过程的同时schema change，验证迁移正确，验证schema change正确
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k2)', 3)
|
|
keys_desc = 'DUPLICATE KEY(k1,k2,k3,k4,k5)'
|
|
assert client.create_table(table_name, SCHEMA.partition_column_no_agg_list,
|
|
partition_info, distribution_info, keys_desc=keys_desc)
|
|
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
|
|
assert client.modify_partition(table_name, 'partition_d', storage_medium='HDD')
|
|
|
|
column_name_list = ['k1',]
|
|
assert client.schema_change_drop_column(table_name, column_name_list, is_wait_job=True)
|
|
|
|
assert client.verify(DATA.expected_data_file_list_3, table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
ret = client.show_tablet(table_name, partition_list=['partition_d'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
client.clean(database_name)
|
|
|
|
|
|
def test_migrate_after_schema_change():
|
|
"""
|
|
{
|
|
"title": "test_migrate_after_schema_change",
|
|
"describe": "schema change的同时迁移，验证迁移正确，验证schema change正确",
|
|
"tag": "p1,system,stability"
|
|
}
|
|
"""
|
|
"""
|
|
schema change的同时迁移，验证迁移正确，验证schema change正确
|
|
"""
|
|
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.use(database_name)
|
|
partition_a = palo_client.Partition('partition_a', '100', storage_medium='HDD')
|
|
partition_b = palo_client.Partition('partition_b', '200', storage_medium='HDD')
|
|
partition_c = palo_client.Partition('partition_c', '300', storage_medium='HDD')
|
|
partition_d = palo_client.Partition('partition_d', '50000', storage_medium='SSD')
|
|
partition_list = [partition_a, partition_b, partition_c, partition_d]
|
|
partition_info = palo_client.PartitionInfo('k3', partition_list=partition_list)
|
|
distribution_info = palo_client.DistributionInfo('Hash(k2)', 3)
|
|
keys_desc = 'DUPLICATE KEY(k1,k2,k3,k4,k5)'
|
|
assert client.create_table(table_name, SCHEMA.partition_column_no_agg_list,
|
|
partition_info, distribution_info, keys_desc=keys_desc)
|
|
assert client.show_tables(table_name)
|
|
assert client.get_index(table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_a') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_b') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_c') == 'HDD'
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'SSD'
|
|
|
|
data_desc_list = palo_client.LoadDataInfo(DATA.file_path_1, table_name)
|
|
assert client.batch_load(util.get_label(), data_desc_list, is_wait=True, broker=broker_info)
|
|
|
|
column_name_list = ['k1',]
|
|
assert client.schema_change_drop_column(table_name, column_name_list,
|
|
is_wait_job=True, is_wait_delete_old_schema=True)
|
|
|
|
assert client.modify_partition(table_name, 'partition_d', storage_medium='HDD')
|
|
|
|
time.sleep(60)
|
|
assert client.verify(DATA.expected_data_file_list_3, table_name)
|
|
assert client.get_partition_storage_medium(table_name, 'partition_d') == 'HDD'
|
|
ret = client.show_tablet(table_name, partition_list=['partition_d'])
|
|
retry_times = 10
|
|
while retry_times > 0:
|
|
try:
|
|
time.sleep(30)
|
|
retry_times -= 1
|
|
check_tablet(ret, 'ssd')
|
|
break
|
|
except Exception as e:
|
|
print(e)
|
|
if retry_times == 0:
|
|
raise pytest.skip('timeout, cannot get tablet from hdd')
|
|
client.clean(database_name)
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
setup_module()
|
|
# test_ssd_all()
|
|
# test_hdd_all()
|
|
# test_ssd_table()
|
|
# test_hdd_table()
|
|
# test_ssd_hdd()
|
|
# test_add_ssd_partition()
|
|
# test_add_hdd_partition()
|
|
# test_modify_to_hdd_partition()
|
|
# test_modify_to_ssd_partition()
|
|
# test_modify_ssd_to_ssd_partition()
|
|
# test_modify_hdd_to_hdd_partition()
|
|
# test_modify_to_hdd_partition_at_cooldown_time()
|
|
# test_modify_medium_cooldown_time()
|
|
# test_modify_medium_cooldown_time_for_partitions()
|
|
# test_modify_cooldown_time()
|
|
# test_modify_invalid_cooldown_time_a()
|
|
# test_modify_invalid_cooldown_time_b()
|
|
# test_cooldown_while_select()
|
|
# test_migrate_while_loading()
|
|
test_migrate_before_schema_change()
|
|
test_migrate_after_schema_change()
|