mirror of
https://github.com/apache/doris.git
synced 2026-11-04 18:29:59 +00:00
1480 lines
67 KiB
Python
1480 lines
67 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_partition_multi_col.py
|
||
* @brief Test for partition by multi column
|
||
*
|
||
**************************************************************************/
|
||
"""
|
||
import time
|
||
import pytest
|
||
|
||
from data import schema as DATA
|
||
from data import load_file as FILE
|
||
from lib import palo_client
|
||
from lib import util
|
||
from lib import common
|
||
from lib import palo_job
|
||
from lib import palo_task
|
||
|
||
LOG = palo_client.LOG
|
||
L = palo_client.L
|
||
|
||
|
||
def setup_module():
|
||
"""
|
||
setUp
|
||
"""
|
||
pass
|
||
|
||
|
||
def teardown_module():
|
||
"""
|
||
tearDown
|
||
"""
|
||
pass
|
||
|
||
|
||
def test_update_tinyint():
|
||
"""
|
||
{
|
||
"title": "test_update_tinyint",
|
||
"describe": "update数据类型tinyint, 数据溢出，Null，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.tinyint_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
util.assert_return(False, '', client.update, table_name, ['v1=null'], 'k1=1')
|
||
# 溢出
|
||
msg = 'Number out of range[-129]. type: TINYINT'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=-129'], 'k1=1')
|
||
msg = 'Number out of range[128]. type: TINYINT'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=128'], 'k1=1')
|
||
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_tinyint_unique_file)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, -1, -1, -1, -1 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1=-1', 'v2=-1', 'v3=-1', 'v4=-1'], 'k1 is not null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3, v4 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_smallint():
|
||
"""
|
||
{
|
||
"title": "test_update_smallint",
|
||
"describe": "update数据类型smallint，数据溢出，Null，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.smallint_column_no_agg_list, keys_desc=DATA.unique_key, set_null=True)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
ret = client.update(table_name, ['v1=null'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
common.check2(client, sql1='select * from %s' % table_name, sql2='select 1, null, 3, 4, 5')
|
||
# 溢出
|
||
msg = 'Number out of range[-32769]. type: SMALLINT'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=-32769'], 'k1=1')
|
||
msg = 'Number out of range[32768]. type: SMALLINT'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=32768'], 'k1=1')
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_smallint_file, max_filter_ratio=0.1)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, -1, -1, -1, -1 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1=-1, v2=-1, v3=-1, v4=-1'], 'k1 is not null or k1 is null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3, v4 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_int():
|
||
"""
|
||
{
|
||
"title": "test_update_int",
|
||
"describe": "update数据类型int，数据溢出，Null，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.int_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null, not null列 update failed
|
||
util.assert_return(False, '', client.update, table_name, ['v1=NULL'], 'k1=1')
|
||
# 溢出
|
||
msg = 'Number out of range[-2147483649]. type: INT'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=-2147483649'], 'k1=1')
|
||
msg = 'Number out of range[2147483648]. type: INT'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=2147483648'], 'k1=1')
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_int_file, max_filter_ratio=0.01)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, -1, -1, -1, -1 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1=-1, v2=-1, v3=-1, v4=-1'], 'k1 is not null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3, v4 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_bigint():
|
||
"""
|
||
{
|
||
"title": "test_update_bigint",
|
||
"describe": "update数据类型bigint，数据溢出，Null，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.bigint_column_no_agg_list, keys_desc=DATA.unique_key, set_null=True)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null todo
|
||
ret = client.update(table_name, ['v1=null'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
common.check2(client, sql1='select * from %s' % table_name, sql2='select 1, null, 3, 4, 5')
|
||
# 溢出
|
||
msg = 'Number out of range[-9223372036854775809]. type: BIGINT'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=-9223372036854775809'], 'k1=1')
|
||
msg = 'Number out of range[9223372036854775808]. type: BIGINT'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=9223372036854775808'], 'k1=1')
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_bigint_file, max_filter_ratio=0.01)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, -1, -1, -1, -1 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1=-1, v2=-1, v3=-1, v4=-1'], 'k1 is not null or k1 is null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3, v4 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_largeint():
|
||
"""
|
||
{
|
||
"title": "test_update_largeint",
|
||
"describe": "update数据类型largeint，数据溢出，Null，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.largeint_column_no_agg_list, keys_desc=DATA.unique_key, set_null=True)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
ret = client.update(table_name, ['v1=null'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
common.check2(client, sql1='select * from %s' % table_name, sql2='select 1, null, 3, 4, 5')
|
||
# 溢出
|
||
msg = 'Numeric overflow'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=-170141183460469231731687303715884105729'], 'k1=1')
|
||
msg = 'Number Overflow. literal: 170141183460469231731687303715884105728'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1=170141183460469231731687303715884105728'], 'k1=1')
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_largeint_file, max_filter_ratio=0.01)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, "-1", "-1", "-1", "-1" from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1=-1, v2=-1, v3=-1, v4=-1'], 'k1 is not null or k1 is null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3, v4 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_char():
|
||
"""
|
||
{
|
||
"title": "test_update_char",
|
||
"describe": "update数据类型char，长度溢出，Null，空串，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.char_normal_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values("1", "2")' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
util.assert_return(False, '', client.update, table_name, ['v1=null'], 'k1=1')
|
||
# 溢出
|
||
util.assert_return(False, '', client.update, table_name,
|
||
['v1="-170141183460469231731687303715884105729"'], 'k1=1')
|
||
# 空串
|
||
ret = client.update(table_name, ['v1=""'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
common.check2(client, sql1='select * from %s' % table_name, sql2='select "1", ""')
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_char_normal_file, max_filter_ratio=0.01)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, "-1" from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1="-1"'], 'k1 is not null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_varchar():
|
||
"""
|
||
{
|
||
"title": "test_update_varchar",
|
||
"describe": "update数据类型varchar，长度溢出，Null，空串，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.varchar_least_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values("1", "2")' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
util.assert_return(False, '', client.update, table_name, ['v1=null'], 'k1=1')
|
||
# 溢出
|
||
util.assert_return(False, '', client.update, table_name,
|
||
['v1="-170141183460469231731687303715884105729"'], 'k1=1')
|
||
# 空串
|
||
ret = client.update(table_name, ['v1=""'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
common.check2(client, sql1='select * from %s' % table_name, sql2='select "1", ""')
|
||
# 多个数据导入
|
||
assert client.drop_table(table_name)
|
||
ret = client.create_table(table_name, DATA.varchar_normal_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_varchar_normal_file, max_filter_ratio=0.02)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, "-1" from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1="-1"'], 'k1 is not null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_date():
|
||
"""
|
||
{
|
||
"title": "test_update_date",
|
||
"describe": "update数据类型date，各种日期格式，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.date_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values("2000-01-01", "2000-01-02", "2000-01-03", "2000-01-04")' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
util.assert_return(False, '', client.update, table_name, ['v1=null'], 'k1="2000-01-01"')
|
||
# 正常数据
|
||
ret = client.update(table_name, ['v1="2020-01-01"'], 'k1="2000-01-01"')
|
||
assert ret, 'update table failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select cast("2000-01-01" as date), cast("2020-01-01" as date), ' \
|
||
'cast("2000-01-03" as date), cast("2000-01-04" as date)'
|
||
common.check2(client, sql1, sql2=sql2)
|
||
# int数据
|
||
ret = client.update(table_name, ['v1="20200102"'], 'k1="2000-01-01"')
|
||
assert ret, 'update table failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select cast("2000-01-01" as date), cast("2020-01-02" as date), ' \
|
||
'cast("2000-01-03" as date), cast("2000-01-04" as date)'
|
||
common.check2(client, sql1, sql2=sql2)
|
||
# 00数据
|
||
util.assert_return(False, '', client.update, table_name, ['v1="0000-00-00"'], 'k1="2000-01-01"')
|
||
# 不支持数据
|
||
util.assert_return(False, '', client.update, table_name, ['v1="2000-13-01"'], 'k1="2000-01-01"')
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_date_file, max_filter_ratio=0.015)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, "2021-01-01", "2021-01-03", "2021-01-02" from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1="2021-01-01"', 'v2="2021-01-03"', 'v3="2021-01-02"'], 'k1 is not null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_datetime():
|
||
"""
|
||
{
|
||
"title": "test_update_datetime",
|
||
"describe": "update数据类型datetime，各种日期格式，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.datetime_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values("2000-01-01 00:00:00", "2000-01-02 01:00:00", ' \
|
||
'"2000-01-03 00:00:00", "2000-01-04 00:00:01")' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
util.assert_return(False, '', client.update, table_name, ['v1=null'], 'k1="2000-01-01 00:00:00"')
|
||
# 正常
|
||
ret = client.update(table_name, ['v1="2020-01-01 00:00:00"'], 'k1="2000-01-01 00:00:00"')
|
||
assert ret, 'update failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select cast("2000-01-01 00:00:00" as datetime), cast("2020-01-01 00:00:00" as datetime), ' \
|
||
'cast("2000-01-03 00:00:00" as datetime), cast("2000-01-04 00:00:01" as datetime)'
|
||
common.check2(client, sql1, sql2=sql2)
|
||
# 溢出
|
||
util.assert_return(False, '', client.update, table_name, ['v1="10020-01-01 00:00:00"'], 'k1="2000-01-01 00:00:00"')
|
||
# 00数据
|
||
util.assert_return(False, '', client.update, table_name, ['v1="2020-01-00 00:00:00"'], 'k1="2000-01-01 00:00:00"')
|
||
# 不支持datetime
|
||
util.assert_return(False, '', client.update, table_name, ['v1="2020-13-01 00:00:00"'], 'k1="2000-01-01 00:00:00"')
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_datetime_file, max_filter_ratio=0.015)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, "2021-01-01 08:00:00", "2021-01-03 08:00:00", "2021-01-02 08:00:00" ' \
|
||
'from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1="2021-01-01 08:00:00"',
|
||
'v2="2021-01-03 08:00:00"',
|
||
'v3="2021-01-02 08:00:00"'], 'k1 is not null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_decimal():
|
||
"""
|
||
{
|
||
"title": "test_update_decimal",
|
||
"describe": "update数据类型decimal，精度溢出，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.decimal_normal_column_no_agg_list,
|
||
keys_desc=DATA.unique_key, set_null=True)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1.001, 2.002, 3.003, 4.004, 5.006)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
ret = client.update(table_name, ['v1=null'], 'k1 > 0')
|
||
assert ret, 'update failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1.001, null, 3.003, 4.004, 5.006'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
# 溢出
|
||
ret = client.update(table_name, ['v1="0.00000001"'], 'k1 = 1.001')
|
||
assert ret, 'update failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1.001, 0, 3.003, 4.004, 5.006'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
util.assert_return(False, '', client.update, table_name, ['v1=100000000000.001'], 'k1 = 1.001')
|
||
# 格式不正确
|
||
util.assert_return(False, '', client.update, table_name, ['v1=123456'], 'k1 = 1.001')
|
||
msg = 'Unexpected exception: Character a is neither a decimal digit number, ' \
|
||
'decimal point, nor "e" notation exponential mark'
|
||
util.assert_return(False, msg, client.update, table_name, ['v1="aaa"'], 'k1 = 1.001')
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_decimal_normal_file, max_filter_ratio=0.01)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, 2.002, 3.003, 4.004, 5.006 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1=2.002', 'v2=3.003', 'v3=4.004', 'v4=5.006'], 'k1 is not null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3, v4 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_double():
|
||
"""
|
||
{
|
||
"title": "test_update_double",
|
||
"describe": "update数据类型double，精度，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.double_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2.002, 3.003, 4.004, 5.006)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
util.assert_return(False, '', client.update, table_name, ['v1=null'], 'k1=1')
|
||
# 溢出
|
||
ret = client.update(table_name, ['v2="1.99999999999999999"'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1, 2.002, 2.0, 4.004, 5.006'
|
||
common.check2(client, sql1, sql2=sql2)
|
||
# 格式
|
||
ret = client.update(table_name, ['v2="1.009e2"'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
sql2 = 'select 1, 2.002, 100.9, 4.004, 5.006'
|
||
common.check2(client, sql1, sql2=sql2)
|
||
ret = client.update(table_name, ['v2=2.0003e-2'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
sql2 = 'select 1, 2.002, 0.020003, 4.004, 5.006'
|
||
common.check2(client, sql1, sql2=sql2)
|
||
# msg = "'a3' is not a number"
|
||
msg = 'error totally whack'
|
||
util.assert_return(False, msg, client.update, table_name, ['v2="a3"'], 'k1=1')
|
||
print(client.select_all(table_name))
|
||
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_double_int_file, max_filter_ratio=0.01)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, 2.002, 3.003, 4.004, 5.006 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1=2.002', 'v2=3.003', 'v3=4.004', 'v4=5.006'], 'k1 is not null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3, v4 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_float():
|
||
"""
|
||
{
|
||
"title": "test_update_float",
|
||
"describe": "update数据类型float，精度，数据校验正确",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
ret = client.create_table(table_name, DATA.float_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2.002, 3.003, 4.004, 5.006)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# null
|
||
util.assert_return(False, '', client.update, table_name, ['v1=null'], 'k1 = 1')
|
||
# 溢出, todo
|
||
ret = client.update(table_name, ['v2="1.99999999999999999"'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1, 2.002, 2.0, 4.004, 5.006'
|
||
common.check2(client, sql1, sql2=sql2)
|
||
# 格式不正确
|
||
ret = client.update(table_name, ['v2="1.009e2"'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1, 2.002, 100.9, 4.004, 5.006'
|
||
common.check2(client, sql1, sql2=sql2)
|
||
ret = client.update(table_name, ['v2=2.0003e-2'], 'k1=1')
|
||
assert ret, 'update table failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1, 2.002, 0.020003, 4.004, 5.006'
|
||
common.check2(client, sql1, sql2=sql2)
|
||
# msg = "'a3' is not a number"
|
||
msg = 'error totally whack'
|
||
util.assert_return(False, msg, client.update, table_name, ['v2="a3"'], 'k1=1')
|
||
print(client.select_all(table_name))
|
||
# 多个数据导入
|
||
assert client.truncate(table_name)
|
||
assert len(client.select_all(table_name)) == 0
|
||
ret = client.stream_load(table_name, FILE.test_float_int_file, max_filter_ratio=0.01)
|
||
assert ret, 'stream load failed'
|
||
sql = 'select k1, 2.002, 3.003, 4.004, 5.006 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql)
|
||
ret = client.update(table_name, ['v1=2.002', 'v2=3.003', 'v3=4.004', 'v4=5.006'], 'k1 is not null')
|
||
assert ret, 'update table failed'
|
||
sql = 'select k1, v1, v2, v3, v4 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_other_table():
|
||
"""
|
||
{
|
||
"title": "test_update_other_table",
|
||
"describe": "update只支持unique表，不支持duplicate表和aggregate表",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
# 创建不同类型的表，并导入数据
|
||
agg_tb = table_name + '_agg'
|
||
dup_tb = table_name + '_dup'
|
||
uniq_tb = table_name + '_uniq'
|
||
ret = client.create_table(agg_tb, DATA.partition_column_list)
|
||
assert ret, 'create table failed'
|
||
ret = client.create_table(dup_tb, DATA.partition_column_no_agg_list)
|
||
assert ret, 'create table failed'
|
||
ret = client.create_table(uniq_tb, DATA.partition_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
for tb in [agg_tb, dup_tb, uniq_tb]:
|
||
ret = client.stream_load(tb, FILE.partition_local_file)
|
||
assert ret, 'stream load failed'
|
||
# update不同类型的表
|
||
ret = client.update(uniq_tb, ['v2="2"'], 'k1=1')
|
||
assert ret, 'update failed'
|
||
msg = 'Only unique table could be updated'
|
||
util.assert_return(False, msg, client.update, agg_tb, ['v2="2"'], 'k1=1')
|
||
msg = 'Only unique table could be updated'
|
||
util.assert_return(False, msg, client.update, dup_tb, ['v2="2"'], 'k1=1')
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_key():
|
||
"""
|
||
{
|
||
"title": "test_update_key",
|
||
"describe": "update只支持value列，不支持key列，包括分区列，分桶列",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["-10", "0", "10", "100", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3, k4, k5)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'load data failed'
|
||
# update key列/分桶列/分区列
|
||
msg = 'Only value columns of unique table could be updated'
|
||
util.assert_return(False, msg, client.update, table_name, ['k3=0'], 'k1=1')
|
||
msg = 'Only value columns of unique table could be updated'
|
||
util.assert_return(False, msg, client.update, table_name, ['k1=-10'], 'k4=1000')
|
||
msg = 'Only value columns of unique table could be updated'
|
||
util.assert_return(False, msg, client.update, table_name, ['k2=-10'], 'k4=1000')
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_multi_row():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_row",
|
||
"describe": "update多行数据，多个分区的多条数据",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3, k4, k5)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'load data failed'
|
||
sql1 = 'select k1, k2, k3, k4, k5, v1, v2, "after update" v3, v4, v5, v6 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql1)
|
||
# update多个分区的多个值
|
||
ret = client.update(table_name, ['v3="after update"'], 'k1 > 0')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select * from %s order by k1' % table_name
|
||
ret2 = client.execute(sql2)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_multi_col():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_col",
|
||
"describe": "update多个列",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3, k4, k5)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'load data failed'
|
||
sql1 = 'select k1, k2, k3, k4, k5, "2020-01-20", "-", "after update" v3, 0, 0, 0 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql1)
|
||
# update多个分区的多个值
|
||
ret = client.update(table_name, ['v3="after update"',
|
||
'v2="-"',
|
||
'v1="2020-01-20"',
|
||
'v4=0',
|
||
'v5=0',
|
||
'v6=0'], 'k1 > 0')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select * from %s order by k1' % table_name
|
||
ret2 = client.execute(sql2)
|
||
util.check(ret1, ret2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_empty():
|
||
"""
|
||
{
|
||
"title": "test_update_empty",
|
||
"describe": "update空表",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3, k4, k5)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
# update多个分区的多个值
|
||
ret = client.update(table_name, ['v3="after update"',
|
||
'v2="-"',
|
||
'v5=0',
|
||
'v6=0'], 'k1 > 0')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select * from %s order by k1' % table_name
|
||
ret2 = client.execute(sql2)
|
||
assert len(ret2) == 0, 'check data error'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'load data failed'
|
||
assert client.verify(FILE.partition_local_file, table_name)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_condition():
|
||
"""
|
||
{
|
||
"title": "test_update_condition",
|
||
"describe": "update中where条件，in，like，< > =，between...and，not，is null，is not null，函数，子查询，and，or, empty",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3, k4, k5)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file, partition_list=["p5"], max_filter_ratio=1)
|
||
assert ret, 'load data failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 2579.668636'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
|
||
# in
|
||
ret = client.update(table_name, ['v6=0'], 'v2 in ("d", "a", "b")')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 0'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
|
||
# like
|
||
ret = client.update(table_name, ['v6=1'], 'v3 like "%%wnrk"')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 0'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
ret = client.update(table_name, ['v6=1'], 'v3 like "%%wnrk%%"')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 1'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
|
||
# 比较
|
||
ret = client.update(table_name, ['v6=2'], 'k1=30')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 2'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
ret = client.update(table_name, ['v6=3'], 'k1 > 30')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 2'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
ret = client.update(table_name, ['v6=3'], 'k1<=20')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 2'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
|
||
# and
|
||
ret = client.update(table_name, ['v6=3'], 'k1 >= 30 and k2 < 300')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 2'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
|
||
ret = client.update(table_name, ['v6=3'], 'k1 in (3, 30) and k5="2040-01-01 00:00:00"')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 3'
|
||
# or
|
||
ret = client.update(table_name, ['v6=4'], 'k1 >= 30 or k2 < 300')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 4'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
# is not null
|
||
ret = client.update(table_name, ['v6=5'], 'v6 is not null')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 5'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
ret = client.update(table_name, ['v6=6'], 'v6 is null')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 5'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
# function
|
||
ret = client.update(table_name, ['v6=6'], 'abs(v5) > 900')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 6'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
# 常量表达式
|
||
ret = client.update(table_name, ['v6=7'], '1=3')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 6'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
ret = client.update(table_name, ['v6=7'], '1=1')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 7'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
ret = client.update(table_name, ['v6=8'], 'true')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 8'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
ret = client.update(table_name, ['v6=9'], 'null is not null')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 8'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_set():
|
||
"""
|
||
{
|
||
"title": "test_update_set",
|
||
"describe": "update中的set设置，包含自增，函数，表达式等，",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3, k4, k5)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file, partition_list=["p5"], max_filter_ratio=1)
|
||
assert ret, 'load data failed'
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 2579.668636'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
# set计算表达式
|
||
ret = client.update(table_name, 'v6 = v6 + v5', 'k1=30')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 976.26649, 3555.935126'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
# set自增
|
||
ret = client.update(table_name, 'v5 = v5 + 1', 'k1=30')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-01-30" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 977.26649, 3555.935126'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
# set自增 & 函数
|
||
ret = client.update(table_name, 'v1 = date_add(v1, interval 2 day)', 'k1=30')
|
||
assert ret, 'update failed'
|
||
sql2 = 'select 30, 300, 3000, 30000, cast("2040-01-01 00:00:00" as datetime), cast("2010-02-1" as date), "d", ' \
|
||
'"tbuvpobzluhbwkljlhwnrkrhowybk", 949.57562, 977.26649, 3555.935126'
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_multi_table():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_table",
|
||
"describe": "同时对不同表执行update操作，update操作均可执行成功",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
# init table
|
||
tb1 = table_name + '_1'
|
||
tb2 = table_name + '_2'
|
||
ret = client.create_table(tb1, DATA.double_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5)' % tb1
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
ret = client.create_table(tb2, DATA.double_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5)' % tb2
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# update
|
||
client1 = common.get_client()
|
||
client2 = common.get_client()
|
||
update_task_1 = palo_task.SyncTask(client1.update, tb1, ['v1=v1+1'], 'k1=1', database_name)
|
||
update_thread_1 = palo_task.TaskThread(update_task_1)
|
||
update_thread_1.start()
|
||
update_task_2 = palo_task.SyncTask(client2.update, tb2, ['v2=v2+1'], 'k1=1', database_name)
|
||
update_thread_2 = palo_task.TaskThread(update_task_2)
|
||
update_thread_2.start()
|
||
time.sleep(10)
|
||
update_thread_1.stop()
|
||
update_thread_1.join()
|
||
update_thread_2.stop()
|
||
update_thread_2.join()
|
||
print(update_task_1.succ_count)
|
||
print(update_task_2.succ_count)
|
||
sql1 = 'select * from %s' % tb1
|
||
sql2 = 'select 1, 2 + %s, 3, 4, 5' % update_task_1.succ_count
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
sql1 = 'select * from %s' % tb2
|
||
sql2 = 'select 1, 2, 3 + %s, 4, 5' % update_task_2.succ_count
|
||
common.check2(client, sql1=sql1, sql2=sql2)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_concurrent_update_table_1():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_table_1",
|
||
"describe": "enable_concurrent_update=false，对同一个表执行update操作，偶发报错：There is an update operation in progress for the current table. Please try again later, or set enable_concurrent_update in fe.conf to true",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
if client.user in ('root', 'admin'):
|
||
ret = client.admin_show_config('enable_concurrent_update')
|
||
fe_config = palo_job.AdminShowConfig(ret[0])
|
||
if fe_config.get_value() == 'true':
|
||
raise pytest.skip("need to set fe config: enable_concurrent_update=false")
|
||
# init table
|
||
ret = client.create_table(table_name, DATA.double_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
print(client.select_all(table_name))
|
||
# update
|
||
client1 = common.get_client()
|
||
client2 = common.get_client()
|
||
update_task_1 = palo_task.SyncTask(client1.update, table_name, ['v1=v1+1'], 'k1=1', database_name)
|
||
update_thread_1 = palo_task.TaskThread(update_task_1)
|
||
update_thread_1.start()
|
||
update_task_2 = palo_task.SyncTask(client2.update, table_name, ['v2=v2+1'], 'k1=1', database_name)
|
||
update_thread_2 = palo_task.TaskThread(update_task_2)
|
||
update_thread_2.start()
|
||
time.sleep(10)
|
||
update_thread_1.stop()
|
||
update_thread_1.join()
|
||
time.sleep(1)
|
||
update_thread_2.stop()
|
||
update_thread_2.join()
|
||
time.sleep(1)
|
||
LOG.info(L('there are %s update v1=v1+1 succ' % update_task_1.succ_count))
|
||
LOG.info(L('there are %s update v2=v2+1 succ' % update_task_2.succ_count))
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1, 2 + %s, 3 + %s, 4, 5' % (update_task_1.succ_count, update_task_2.succ_count)
|
||
#common.check2(client, sql1=sql1, sql2=sql2)
|
||
# msg = 'There is an update operation in progress for the current table. Please try again later, or set enable_concurrent_update in fe.conf to true'
|
||
|
||
|
||
def test_concurrent_update_table_2():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_table_2",
|
||
"describe": "enable_concurrent_update=fasle，对同一个表执行update操作，update的数据不冲突，偶发报错：There is an update operation in progress for the current table. Please try again later, or set enable_concurrent_update in fe.conf to true",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
if client.user in ('root', 'admin'):
|
||
ret = client.admin_show_config('enable_concurrent_update')
|
||
fe_config = palo_job.AdminShowConfig(ret[0])
|
||
if fe_config.get_value() == 'true':
|
||
raise pytest.skip("need to set fe config: enable_concurrent_update=false")
|
||
# init table
|
||
ret = client.create_table(table_name, DATA.double_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5), (2, 3, 4, 5, 6)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# update
|
||
client1 = common.get_client()
|
||
client2 = common.get_client()
|
||
update_task_1 = palo_task.SyncTask(client1.update, table_name, ['v1=v1+1'], 'k1=1', database_name)
|
||
update_thread_1 = palo_task.TaskThread(update_task_1)
|
||
update_thread_1.start()
|
||
update_task_2 = palo_task.SyncTask(client2.update, table_name, ['v2=v2+1'], 'k1=2', database_name)
|
||
update_thread_2 = palo_task.TaskThread(update_task_2)
|
||
update_thread_2.start()
|
||
time.sleep(10)
|
||
update_thread_1.stop()
|
||
update_thread_1.join()
|
||
update_thread_2.stop()
|
||
update_thread_2.join()
|
||
LOG.info(L('there are %s update k1=1 succ' % update_task_1.succ_count))
|
||
LOG.info(L('there are %s update k1=2 succ' % update_task_2.succ_count))
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1, 2 + %s, 3, 4, 5 union select 2, 3, 4 + %s, 5, 6' \
|
||
% (update_task_1.succ_count, update_task_2.succ_count)
|
||
common.check2(client, sql1=sql1, sql2=sql2, forced=True)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_concurrent_update_table_3():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_table_2",
|
||
"describe": "enable_concurrent_update=true，对同一个表执行update操作，update的数据不冲突"
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
if client.user in ('root', 'admin'):
|
||
ret = client.admin_show_config('enable_concurrent_update')
|
||
fe_config = palo_job.AdminShowConfig(ret[0])
|
||
if fe_config.get_value() == 'false':
|
||
raise pytest.skip("need to set fe config: enable_concurrent_update=true")
|
||
# init table
|
||
ret = client.create_table(table_name, DATA.double_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5), (2,3,4,5,6)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
# update
|
||
client1 = common.get_client()
|
||
client2 = common.get_client()
|
||
update_task_1 = palo_task.SyncTask(client1.update, table_name, ['v1=v1+1'], 'k1=1', database_name)
|
||
update_thread_1 = palo_task.TaskThread(update_task_1)
|
||
update_thread_1.start()
|
||
update_task_2 = palo_task.SyncTask(client2.update, table_name, ['v2=v2+1'], 'k1=2', database_name)
|
||
update_thread_2 = palo_task.TaskThread(update_task_2)
|
||
update_thread_2.start()
|
||
time.sleep(10)
|
||
update_thread_1.stop()
|
||
update_thread_1.join()
|
||
update_thread_2.stop()
|
||
update_thread_2.join()
|
||
LOG.info(L('there are %s update k1=1 succ' % update_task_1.succ_count))
|
||
LOG.info(L('there are %s update k1=2 succ' % update_task_2.succ_count))
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1, 2 + %s, 3, 4, 5 union select 2, 3, 4 + %s, 5, 6' \
|
||
% (update_task_1.succ_count, update_task_2.succ_count)
|
||
common.check2(client, sql1=sql1, sql2=sql2, forced=True)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_concurrent_update_table_4():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_table_4",
|
||
"describe": "enable_concurrent_update=true，对同一个表执行update操作，update的数据冲突"
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
if client.user in ('root', 'admin'):
|
||
ret = client.admin_show_config('enable_concurrent_update')
|
||
fe_config = palo_job.AdminShowConfig(ret[0])
|
||
if fe_config.get_value() == 'false':
|
||
raise pytest.skip("need to set fe config: enable_concurrent_update=true")
|
||
# init table
|
||
ret = client.create_table(table_name, DATA.double_column_no_agg_list, keys_desc=DATA.unique_key)
|
||
assert ret, 'create table failed'
|
||
sql = 'insert into %s values(1, 2, 3, 4, 5)' % table_name
|
||
ret = client.execute(sql)
|
||
assert ret == (), 'insert date failed'
|
||
print(client.select_all(table_name))
|
||
# update
|
||
client1 = common.get_client()
|
||
client2 = common.get_client()
|
||
update_task_1 = palo_task.SyncTask(client1.update, table_name, ['v1=v1+1'], 'k1=1', database_name)
|
||
update_thread_1 = palo_task.TaskThread(update_task_1)
|
||
update_thread_1.start()
|
||
update_task_2 = palo_task.SyncTask(client2.update, table_name, ['v2=v2+1'], 'k1=1', database_name)
|
||
update_thread_2 = palo_task.TaskThread(update_task_2)
|
||
update_thread_2.start()
|
||
time.sleep(10)
|
||
update_thread_1.stop()
|
||
update_thread_1.join()
|
||
time.sleep(1)
|
||
update_thread_2.stop()
|
||
update_thread_2.join()
|
||
time.sleep(1)
|
||
LOG.info(L('there are %s update v1=v1+1 succ' % update_task_1.succ_count))
|
||
LOG.info(L('there are %s update v2=v2+1 succ' % update_task_2.succ_count))
|
||
sql1 = 'select * from %s' % table_name
|
||
sql2 = 'select 1, 2 + %s, 3 + %s, 4, 5' % (update_task_1.succ_count, update_task_2.succ_count)
|
||
# common.check2(client, sql1=sql1, sql2=sql2, forced=True)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_table_with_idx():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_table_2",
|
||
"describe": "update带有索引的表，bloom filter，bitmap Index",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'load data failed'
|
||
sql1 = 'select k1, k2, k3, k4, k5, v1, "-", v3, v4, v5, 0.01 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql1)
|
||
sql2 = 'select k1, k2, k3, 0, k5, v1, "-", v3, v4, v5, 0.01 from %s order by k1' % table_name
|
||
ret2 = client.execute(sql2)
|
||
|
||
ret = client.schema_change(table_name, bloom_filter_column_list=['v2', 'v6'], is_wait=True)
|
||
assert ret, 'alter add bloom filter failed'
|
||
ret = client.update(table_name, ['v2="-"', 'v6=0.01'], 'k1 > 0')
|
||
assert ret, 'update failed'
|
||
ret1_sut = client.select_all(table_name)
|
||
util.check(ret1, ret1_sut, True)
|
||
time.sleep(10)
|
||
ret = client.create_bitmap_index_table(table_name, index_name, 'k4', is_wait=True)
|
||
assert ret
|
||
ret = client.update(table_name, 'k4=0', 'k1 > 0')
|
||
ret2_sut = client.select_all(table_name)
|
||
util.check(ret2, ret2_sut, True)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_table_with_batch_delete():
|
||
"""
|
||
{
|
||
"title": "test_update_table_with_batch_delete",
|
||
"describe": "update开启batch delte的表 update成功",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
try:
|
||
client.enable_feature_batch_delete(table_name)
|
||
except Exception as e:
|
||
pass
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'stream load failed'
|
||
sql1 = 'select k1, k2, k3, k4, k5, v1, "-", v3, v4, v5, 0.01 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql1)
|
||
ret = client.update(table_name, ['v2="-"', 'v6=0.01'], 'k1 > 0')
|
||
assert ret, 'update failed'
|
||
ret1_sut = client.select_all(table_name)
|
||
util.check(ret1, ret1_sut, True)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_table_while_load():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_table_2",
|
||
"describe": "在update的同时，导入数据",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
client1 = common.get_client()
|
||
update_task = palo_task.SyncTask(client1.update, table_name, ['v2="0"'], 'k1 > 0', database_name)
|
||
update_thread = palo_task.TaskThread(update_task)
|
||
update_thread.start()
|
||
time.sleep(1)
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'stream load failed'
|
||
update_thread.stop()
|
||
update_thread.join()
|
||
ret = client.select_all(table_name)
|
||
assert len(ret) == 30, 'check data error'
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_table_while_delete():
|
||
"""
|
||
{
|
||
"title": "test_update_multi_table_2",
|
||
"describe": "在update的同时，删除数据，偶发：doris-1130",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'stream load failed'
|
||
client1 = common.get_client()
|
||
update_task = palo_task.SyncTask(client1.update, table_name, ['v2="0"'], 'k1 > 0', database_name)
|
||
update_task.interval = 0
|
||
update_thread = palo_task.TaskThread(update_task)
|
||
update_thread.start()
|
||
ret = client.delete(table_name, [("k1", ">", "20")])
|
||
assert ret, 'delete failed'
|
||
time.sleep(1)
|
||
update_thread.stop()
|
||
update_thread.join()
|
||
ret = client.select_all(table_name)
|
||
print(len(ret))
|
||
# assert len(ret) == 20, 'check data error, there are %s rows' % len(ret)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_table_while_add_column():
|
||
"""
|
||
{
|
||
"title": "test_update_table_while_add_column",
|
||
"describe": "在update的同时，加列",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'stream load failed'
|
||
client1 = common.get_client()
|
||
update_task = palo_task.SyncTask(client1.update, table_name, ['v2="0"'], 'k1 > 0', database_name)
|
||
update_thread = palo_task.TaskThread(update_task)
|
||
update_thread.start()
|
||
# add column
|
||
ret = client.schema_change_add_column(table_name, [('v_add', 'INT', '', '0')], is_wait_job=True)
|
||
assert ret, 'add column failed'
|
||
update_thread.stop()
|
||
update_thread.join()
|
||
ret = client.select_all(table_name)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_table_while_drop_column():
|
||
"""
|
||
{
|
||
"title": "test_update_table_while_drop_column",
|
||
"describe": "在update的同时，减列",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'stream load failed'
|
||
sql1 = 'select k1, k2, k3, k4, k5, v1, "0", v3, v4, v5 from %s order by k1' % table_name
|
||
ret1 = client.execute(sql1)
|
||
|
||
client1 = common.get_client()
|
||
update_task = palo_task.SyncTask(client1.update, table_name, ['v2="0"'], 'k1 > 0', database_name)
|
||
update_thread = palo_task.TaskThread(update_task)
|
||
update_thread.start()
|
||
# add column
|
||
ret = client.schema_change_drop_column(table_name, ['v6'], is_wait_job=True)
|
||
assert ret, 'add column failed'
|
||
update_thread.stop()
|
||
update_thread.join()
|
||
ret = client.select_all(table_name)
|
||
util.check(ret, ret1, True)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_table_while_modify_column():
|
||
"""
|
||
{
|
||
"title": "test_update_table_while_modify_column",
|
||
"describe": "在update的同时，修改列",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'stream load failed'
|
||
client1 = common.get_client()
|
||
update_task = palo_task.SyncTask(client1.update, table_name, ['v2="0"'], 'k1 > 0', database_name)
|
||
update_thread = palo_task.TaskThread(update_task)
|
||
update_thread.start()
|
||
# modify column
|
||
ret = client.schema_change_modify_column(table_name, 'v2', 'char(5)', is_wait_job=True, aggtype='-')
|
||
assert ret, 'add column failed'
|
||
update_thread.stop()
|
||
update_thread.join()
|
||
ret = client.select_all(table_name)
|
||
client.clean(database_name)
|
||
|
||
|
||
def test_update_table_while_rollup():
|
||
"""
|
||
{
|
||
"title": "test_update_table_while_rollup",
|
||
"describe": "在update的同时，创建rollup",
|
||
"tag": "function,p0,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 = common.create_workspace(database_name)
|
||
partition_info = palo_client.PartitionInfo("k2",
|
||
["p1", "p2", "p3", "p4", "p5"],
|
||
["0", "100", "200", "300", "MAXVALUE"])
|
||
distribution_info = palo_client.DistributionInfo('hash(k1)', 10)
|
||
keys_desc = "UNIQUE KEY(k1, k2, k3)"
|
||
ret = client.create_table(table_name, DATA.partition_column_no_agg_list, keys_desc=keys_desc,
|
||
partition_info=partition_info, distribution_info=distribution_info)
|
||
assert ret, 'create table failed'
|
||
ret = client.stream_load(table_name, FILE.partition_local_file)
|
||
assert ret, 'stream load failed'
|
||
client1 = common.get_client()
|
||
update_task = palo_task.SyncTask(client1.update, table_name, ['v2="0"'], 'k1 > 0', database_name)
|
||
update_thread = palo_task.TaskThread(update_task)
|
||
update_thread.start()
|
||
# create rollup
|
||
ret = client.create_rollup_table(table_name, index_name, ['k1', 'k3', 'k2', 'v2', 'v6'], is_wait=True)
|
||
assert ret, 'create rollup failed'
|
||
update_thread.stop()
|
||
update_thread.join()
|
||
ret = client.select_all(table_name)
|
||
client.clean(database_name)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
setup_module()
|
||
#test_update_condition()
|
||
test_update_table_while_add_column()
|