mirror of
https://github.com/apache/impala.git
synced 2025-12-19 18:12:08 -05:00
125df65322
test_rename_drop continues to be flaky in non-HDFS environment. HMS seems to be slower to response in non-HDFS environment. There are also possibillity of table lock contention with background TableLoader and CatalogDelta thread that cause DROP query happen after RENAME instead of BEFORE. This patch limit running test_rename_drop in HDFS and exhaustive mode in the meantime. Change-Id: Ie55e6d6093367c454cf3e31ed8a409b6e091193d Reviewed-on: http://gerrit.cloudera.org:8080/22933 Reviewed-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com> Tested-by: Michael Smith <michael.smith@cloudera.com>
109 lines
4.6 KiB
Python
109 lines
4.6 KiB
Python
# 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.
|
|
|
|
from __future__ import absolute_import, division, print_function
|
|
from copy import deepcopy
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
|
|
from tests.util.filesystem_utils import IS_HDFS
|
|
|
|
|
|
@CustomClusterTestSuite.with_args(
|
|
catalogd_args="--hms_event_polling_interval_s=0",
|
|
# Requires longer update frequency so invalidate->drop can happen within an update.
|
|
# Still shorter than default so tests are quicker.
|
|
statestored_args="--statestore_update_frequency_ms=1000",
|
|
cluster_size=1)
|
|
class TestConcurrentRename(CustomClusterTestSuite):
|
|
"""Test concurrent rename with invalidate and other DDLs."""
|
|
|
|
@pytest.mark.skipif(
|
|
not IS_HDFS, reason="Test is not deterministic in non-HDFS environment")
|
|
def test_rename_drop(self, vector, unique_database):
|
|
if self.exploration_strategy() != 'exhaustive':
|
|
pytest.skip('runs only in exhaustive')
|
|
catalogd = self.cluster.catalogd
|
|
name = "{}.tbl".format(unique_database)
|
|
self.execute_query("create table {} (s string)".format(name))
|
|
self.execute_query("describe {}".format(name))
|
|
|
|
new_vector = deepcopy(vector)
|
|
new_vector.get_value('exec_option')['debug_action'] = \
|
|
"catalogd_table_rename_delay:SLEEP@6000"
|
|
with self.create_impala_client_from_vector(new_vector) as alter_client, \
|
|
self.create_impala_client() as reset_client:
|
|
version_after_create = catalogd.service.get_catalog_version()
|
|
alter_handle = alter_client.execute_async(
|
|
"alter table {0} rename to {0}2".format(name))
|
|
alter_client.wait_for_admission_control(alter_handle, timeout_s=10)
|
|
# Wait for at most 10 second until catalogd increase the version for rename
|
|
# operation.
|
|
start_time = time.time()
|
|
while (time.time() - start_time < 10.0
|
|
and catalogd.service.get_catalog_version() <= version_after_create):
|
|
time.sleep(0.05)
|
|
invalidate_1 = "invalidate metadata {}".format(name)
|
|
invalidate_2 = "invalidate metadata {}2".format(name)
|
|
drop_query = "drop table {}2".format(name)
|
|
reset_handle = reset_client.execute_async(invalidate_1)
|
|
inv_success = self.execute_query_until_success(invalidate_2)
|
|
drop_success = self.execute_query_until_success(drop_query)
|
|
|
|
try:
|
|
alter_client.wait_for_finished_timeout(alter_handle, timeout=10)
|
|
alter_client.close_query(alter_handle)
|
|
assert False, "Expected alter to fail"
|
|
except Exception as e:
|
|
assert "The new table/view {}2 was concurrently removed during rename."\
|
|
.format(name) in str(e)
|
|
finally:
|
|
reset_client.wait_for_finished_timeout(reset_handle, timeout=10)
|
|
reset_client.close_query(reset_handle)
|
|
assert inv_success, "Expect '{}' to succeed, but failed.".format(invalidate_2)
|
|
assert drop_success, "Expect '{}' to succeed, but failed.".format(drop_query)
|
|
|
|
def test_rename_invalidate(self, vector, unique_database):
|
|
name = "{}.tbl".format(unique_database)
|
|
self.execute_query("create table {} (s string)".format(name))
|
|
self.execute_query("describe {}".format(name))
|
|
|
|
new_vector = deepcopy(vector)
|
|
new_vector.get_value('exec_option')['debug_action'] = \
|
|
"catalogd_table_rename_delay:SLEEP@5000"
|
|
with self.create_impala_client_from_vector(new_vector) as alter_client:
|
|
alter_handle = alter_client.execute_async(
|
|
"alter table {0} rename to {0}2".format(name))
|
|
time.sleep(0.1)
|
|
self.execute_query("invalidate metadata {}".format(name))
|
|
|
|
alter_client.wait_for_finished_timeout(alter_handle, timeout=10)
|
|
alter_client.close_query(alter_handle)
|
|
|
|
def execute_query_until_success(self, query, max_retry=3):
|
|
success = False
|
|
while not success and max_retry > 0:
|
|
try:
|
|
result = self.execute_query(query)
|
|
success = result.success
|
|
except Exception:
|
|
# ignore
|
|
max_retry -= 1
|
|
return success
|