blob: 8f5066e6d699566510ae359729fc4c03a04d6548 [file] [log] [blame]
# 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
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# The base class that should be used for almost all Impala tests
import glob
import grp
import json
import logging
import os
import pprint
import pwd
import pytest
import re
import requests
import shutil
import socket
import subprocess
import tempfile
import time
from functools import wraps
from getpass import getuser
from random import choice
from subprocess import check_call
from tests.common.base_test_suite import BaseTestSuite
from tests.common.environ import HIVE_MAJOR_VERSION
from tests.common.errors import Timeout
from tests.common.impala_connection import create_connection
from tests.common.impala_service import ImpaladService
from tests.common.test_dimensions import (
from tests.common.test_result_verifier import (
from tests.common.test_vector import ImpalaTestDimension
from tests.performance.query import Query
from tests.performance.query_exec_functions import execute_using_jdbc
from tests.performance.query_executor import JdbcQueryExecConfig
from tests.util.filesystem_utils import (
from tests.util.hdfs_util import (
from tests.util.test_file_parser import (
from tests.util.thrift_util import create_transport
# Imports required for Hive Metastore Client
from hive_metastore import ThriftHiveMetastore
from thrift.protocol import TBinaryProtocol
# Initializing the logger before conditional imports, since we will need it
# for them.
LOG = logging.getLogger('impala_test_suite')
# The ADLS python client isn't downloaded when ADLS isn't the target FS, so do a
# conditional import.
from tests.util.adls_util import ADLSClient
except ImportError:
LOG.error("Need the ADLSClient for testing with ADLS")
IMPALAD_HOST_PORT_LIST = pytest.config.option.impalad.split(',')
assert len(IMPALAD_HOST_PORT_LIST) > 0, 'Must specify at least 1 impalad to target'
IMPALAD_HOSTNAME + ":" + pytest.config.option.impalad_hs2_port
IMPALAD_HOSTNAME + ":" + pytest.config.option.impalad_hs2_http_port
HIVE_HS2_HOST_PORT = pytest.config.option.hive_server2
HDFS_CONF = HdfsConfig(pytest.config.option.minicluster_xml_conf)
# Some tests use the IP instead of the host.
INTERNAL_LISTEN_IP = socket.gethostbyname_ex(INTERNAL_LISTEN_HOST)[2][0]
# Match any SET statement. Assume that query options' names
# only contain alphabets, underscores and digits after position 1.
# The statement may include SQL line comments starting with --, which we need to
# strip out. The test file parser already strips out comments starting with #.
COMMENT_LINES_REGEX = r'(?:\s*--.*\n)*'
SET_PATTERN = re.compile(
COMMENT_LINES_REGEX + r'\s*set\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*=*', re.I)
METRICS_URL = 'http://{0}:25000/metrics?json'.format(IMPALAD_HOSTNAME)
VARZ_URL = 'http://{0}:25000/varz?json'.format(IMPALAD_HOSTNAME)
GROUP_NAME = grp.getgrgid(pwd.getpwnam(getuser()).pw_gid).gr_name
# Base class for Impala tests. All impala test cases should inherit from this class
class ImpalaTestSuite(BaseTestSuite):
def add_test_dimensions(cls):
A hook for adding additional dimensions.
By default load the table_info and exec_option dimensions, but if a test wants to
add more dimensions or different dimensions they can override this function.
super(ImpalaTestSuite, cls).add_test_dimensions()
# Execute tests through Beeswax by default. Individual tests that have been converted
# to work with the HS2 client can add HS2 in addition to or instead of beeswax.
cls.ImpalaTestMatrix.add_dimension(ImpalaTestDimension('protocol', 'beeswax'))
def setup_class(cls):
"""Setup section that runs before each test suite"""
cls.client = None
cls.hive_client = None
cls.hs2_client = None
cls.hs2_http_client = None
# Create a Hive Metastore Client (used for executing some test SETUP steps
metastore_host, metastore_port = pytest.config.option.metastore_server.split(':')
trans_type = 'buffered'
if pytest.config.option.use_kerberos:
trans_type = 'kerberos'
cls.hive_transport = create_transport(
protocol = TBinaryProtocol.TBinaryProtocol(cls.hive_transport)
cls.hive_client = ThriftHiveMetastore.Client(protocol)
# Default query options are populated on demand.
cls.default_query_options = {}
cls.impalad_test_service = cls.create_impala_service()
# There are multiple clients for interacting with the underlying storage service.
# There are two main types of clients: filesystem-specific clients and CLI clients.
# CLI clients all use the 'hdfs dfs' CLI to execute operations against a target
# filesystem.
# 'filesystem_client' is a generic interface for doing filesystem operations that
# works across all the filesystems that Impala supports. 'filesystem_client' uses
# either the HDFS command line (e.g. 'hdfs dfs'), a filesystem-specific library, or
# a wrapper around both, to implement common HDFS operations.
# *Test writers should always use 'filesystem_client' unless they are using filesystem
# specific functionality (e.g. HDFS ACLs).*
# The implementation of 'filesystem_client' for each filesystem is:
# HDFS: uses a mixture of pywebhdfs (which is faster than the HDFS CLI) and the
# S3: uses the HDFS CLI
# ABFS: uses the HDFS CLI
# ADLS: uses a mixture of azure-data-lake-store-python and the HDFS CLI (TODO:
# this should completely switch to the HDFS CLI once we test it)
# 'hdfs_client' is a HDFS-specific client library, and it only works when running on
# HDFS. When using 'hdfs_client', the test must be skipped on everything other than
# HDFS. This is only really useful for tests that do HDFS ACL operations. The
# 'hdfs_client' supports all the methods and functionality of the 'filesystem_client',
# with additional support for ACL operations such as chmod, chown, getacl, and setacl.
# 'hdfs_client' is set to None on non-HDFS systems.
cls.hdfs_client = cls.create_hdfs_client()
cls.filesystem_client = cls.hdfs_client
elif IS_S3:
# S3Guard needs filesystem operations to go through the s3 connector. Use the
# HDFS command line client.
cls.filesystem_client = HadoopFsCommandLineClient("S3")
elif IS_ABFS:
# ABFS is implemented via HDFS command line client
cls.filesystem_client = HadoopFsCommandLineClient("ABFS")
elif IS_ADLS:
cls.filesystem_client = ADLSClient(ADLS_STORE_NAME)
# Override the shell history path so that commands run by any tests
# don't write any history into the developer's file.
os.environ['IMPALA_HISTFILE'] = '/dev/null'
def teardown_class(cls):
"""Setup section that runs after each test suite"""
# Cleanup the Impala and Hive Metastore client connections
if cls.hive_transport:
def create_impala_client(cls, host_port=None, protocol='beeswax',
if host_port is None:
host_port = cls.__get_default_host_port(protocol)
client = create_connection(host_port=host_port,
use_kerberos=pytest.config.option.use_kerberos, protocol=protocol,
return client
def get_impalad_cluster_size(cls):
return len(cls.__get_cluster_host_ports('beeswax'))
def create_client_for_nth_impalad(cls, nth=0):
# TODO Extended it to other protocols
protocol = 'beeswax'
host_port = cls.__get_cluster_host_ports(protocol)[nth]
return ImpalaTestSuite.create_impala_client(host_port, protocol=protocol)
def create_impala_clients(cls):
"""Creates Impala clients for all supported protocols."""
# The default connection (self.client) is Beeswax so that existing tests, which assume
# Beeswax do not need modification (yet).
cls.client = cls.create_impala_client(protocol='beeswax')
cls.hs2_client = None
cls.hs2_client = cls.create_impala_client(protocol='hs2')
except Exception, e:
# HS2 connection can fail for benign reasons, e.g. running with unsupported auth."HS2 connection setup failed, continuing...: {0}".format(e))
cls.hs2_http_client = None
cls.hs2_http_client = cls.create_impala_client(protocol='hs2-http')
except Exception, e:
# HS2 HTTP connection can fail for benign reasons, e.g. running with unsupported
# auth."HS2 HTTP connection setup failed, continuing...: {0}".format(e))
def close_impala_clients(cls):
"""Closes Impala clients created by create_impala_clients()."""
if cls.client:
cls.client = None
if cls.hs2_client:
cls.hs2_client = None
if cls.hs2_http_client:
cls.hs2_http_client = None
def __get_default_host_port(cls, protocol):
if protocol == 'beeswax':
return IMPALAD
elif protocol == 'hs2-http':
assert protocol == 'hs2'
def __get_cluster_host_ports(cls, protocol):
"""Return a list of host/port combinations for all impalads in the cluster."""
if protocol == 'beeswax':
assert protocol in ('hs2', 'hs2-http')
# TODO: support running tests against multiple coordinators for HS2. It should work,
# we just need to update all test runners to pass in all host/port combinations for
# the cluster and then handle it here.
raise NotImplementedError(
"Not yet implemented: only one HS2 host/port can be configured")
def create_impala_service(
cls, host_port=IMPALAD, webserver_interface="", webserver_port=25000):
host, port = host_port.split(':')
if webserver_interface == "":
webserver_interface = host
return ImpaladService(host, beeswax_port=port,
webserver_interface=webserver_interface, webserver_port=webserver_port)
def create_hdfs_client(cls):
if pytest.config.option.namenode_http_address is None:
webhdfs_client = get_webhdfs_client_from_conf(HDFS_CONF)
host, port = pytest.config.option.namenode_http_address.split(":")
webhdfs_client = get_webhdfs_client(host, port)
return DelegatingHdfsClient(webhdfs_client, HadoopFsCommandLineClient())
def all_db_names(cls):
results = cls.client.execute("show databases").data
# Extract first column - database name
return [row.split("\t")[0] for row in results]
def cleanup_db(cls, db_name, sync_ddl=1):
cls.client.execute("use default")
cls.client.set_configuration({'sync_ddl': sync_ddl})
cls.client.execute("drop database if exists `" + db_name + "` cascade")
def __restore_query_options(self, query_options_changed, impalad_client):
Restore the list of modified query options to their default values.
# Populate the default query option if it's empty.
if not self.default_query_options:
query_options = impalad_client.get_default_configuration()
for key, value in query_options.iteritems():
self.default_query_options[key.upper()] = value
# Restore all the changed query options.
for query_option in query_options_changed:
query_option = query_option.upper()
if not query_option in self.default_query_options:
default_val = self.default_query_options[query_option]
query_str = 'SET ' + query_option + '="' + default_val + '"'
except Exception as e:'Unexpected exception when executing ' + query_str + ' : ' + str(e))
def get_impala_partition_info(self, table_name, *include_fields):
Find information about partitions of a table, as returned by a SHOW PARTITION
statement. Return a list that contains one tuple for each partition.
If 'include_fields' is not specified, the tuples will contain all the fields returned
by SHOW PARTITION. Otherwise, return only those fields whose names are listed in
'include_fields'. Field names are compared case-insensitively.
exec_result = self.client.execute('show partitions %s' % table_name)
fieldSchemas = exec_result.schema.fieldSchemas
fields_dict = {}
for idx, fs in enumerate(fieldSchemas):
fields_dict[] = idx
rows = exec_result.get_data().split('\n')
fields_idx = []
for fn in include_fields:
fn = fn.lower()
assert fn in fields_dict, 'Invalid field: %s' % fn
result = []
for row in rows:
fields = row.split('\t')
if not fields_idx:
result_fields = fields
result_fields = []
for i in fields_idx:
return result
def get_debug_page(self, page_url):
"""Returns the content of the debug page 'page_url' as json."""
response = requests.get(page_url)
assert response.status_code ==
return json.loads(response.text)
def get_var_current_val(self, var):
"""Returns the current value of a given Impalad flag variable."""
# Parse the /varz endpoint to get the flag information.
varz = self.get_debug_page(VARZ_URL)
assert 'flags' in varz.keys()
filtered_varz = filter(lambda flag: flag['name'] == var, varz['flags'])
assert len(filtered_varz) == 1
assert 'current' in filtered_varz[0].keys()
return filtered_varz[0]['current'].strip()
def get_metric(self, name):
"""Finds the metric with name 'name' and returns its value as an int."""
def iter_metrics(group):
for m in group['metrics']:
yield m
for c in group['child_groups']:
for m in iter_metrics(c):
yield m
metrics = self.get_debug_page(METRICS_URL)['metric_group']
for m in iter_metrics(metrics):
if m['name'] == name:
return int(m['value'])
assert False, "Could not find metric: %s" % name
def __do_replacements(self, s, use_db=None, extra=None):
globs = globals()
repl = dict(('$' + k, globs[k]) for k in [
'$USER': getuser()})
if use_db:
repl['$DATABASE'] = use_db
elif '$DATABASE' in s:
raise AssertionError("Query contains $DATABASE but no use_db specified")
if extra:
for k, v in extra.iteritems():
if k in repl:
raise RuntimeError("Key {0} is reserved".format(k))
repl[k] = v
for k, v in repl.iteritems():
s = s.replace(k, v)
return s
def __verify_exceptions(self, expected_strs, actual_str, use_db):
Verifies that at least one of the strings in 'expected_str' is either:
* A row_regex: line that matches the actual exception string 'actual_str'
* A substring of the actual exception string 'actual_str'.
actual_str = actual_str.replace('\n', '')
for expected_str in expected_strs:
# In error messages, some paths are always qualified and some are not.
# So, allow both $NAMENODE and $FILESYSTEM_PREFIX to be used in CATCH.
expected_str = self.__do_replacements(expected_str.strip(), use_db=use_db)
# Remove comments
expected_str = re.sub(COMMENT_LINES_REGEX, '', expected_str)
# Strip newlines so we can split error message into multiple lines
expected_str = expected_str.replace('\n', '')
expected_regex = try_compile_regex(expected_str)
if expected_regex:
if expected_regex.match(actual_str): return
# Not a regex - check if expected substring is present in actual.
if expected_str in actual_str: return
assert False, 'Unexpected exception string. Expected: %s\nNot found in actual: %s' % \
(expected_str, actual_str)
def __verify_results_and_errors(self, vector, test_section, result, use_db):
"""Verifies that both results and error sections are as expected. Rewrites both
by replacing $NAMENODE, $DATABASE and $IMPALA_HOME with their actual values, and
optionally rewriting filenames with __HDFS_FILENAME__, to ensure that expected and
actual values are easily compared.
replace_filenames_with_placeholder = True
for section_name in ('RESULTS', 'DBAPI_RESULTS', 'ERRORS'):
if section_name in test_section:
if "$NAMENODE" in test_section[section_name]:
replace_filenames_with_placeholder = False
test_section[section_name] = test_section[section_name] \
.replace('$NAMENODE', NAMENODE) \
.replace('$USER', getuser()) \
if use_db:
test_section[section_name] = test_section[section_name].replace('$DATABASE', use_db)
result_section, type_section = 'RESULTS', 'TYPES'
if vector.get_value('protocol').startswith('hs2'):
# hs2 or hs2-http
if 'DBAPI_RESULTS' in test_section:
assert 'RESULTS' in test_section,\
"Base RESULTS section must always be included alongside DBAPI_RESULTS"
# In some cases Impyla (the HS2 dbapi client) is expected to return different
# results, so use the dbapi-specific section if present.
result_section = 'DBAPI_RESULTS'
if 'HS2_TYPES' in test_section:
assert 'TYPES' in test_section,\
"Base TYPES section must always be included alongside HS2_TYPES"
# In some cases HS2 types are expected differ from Beeswax types (e.g. see
# IMPALA-914), so use the HS2-specific section if present.
type_section = 'HS2_TYPES'
verify_raw_results(test_section, result, vector.get_value('table_format').file_format,
result_section, type_section, pytest.config.option.update_results,
def run_test_case(self, test_file_name, vector, use_db=None, multiple_impalad=False,
encoding=None, test_file_vars=None):
Runs the queries in the specified test based on the vector values
Runs the query using targeting the file format/compression specified in the test
vector and the exec options specified in the test vector. If multiple_impalad=True
a connection to a random impalad will be chosen to execute each test section.
Otherwise, the default impalad client will be used. If 'protocol' (either 'hs2' or
'beeswax') is set in the vector, a client for that protocol is used. Otherwise we
use the default: beeswax.
Additionally, the encoding for all test data can be specified using the 'encoding'
parameter. This is useful when data is ingested in a different encoding (ex.
latin). If not set, the default system encoding will be used.
If a dict 'test_file_vars' is provided, then all keys will be replaced with their
values in queries before they are executed. Callers need to avoid using reserved key
names, see 'reserved_keywords' below.
table_format_info = vector.get_value('table_format')
exec_options = vector.get_value('exec_option')
protocol = vector.get_value('protocol')
target_impalad_clients = list()
if multiple_impalad:
target_impalad_clients =\
[ImpalaTestSuite.create_impala_client(host_port, protocol=protocol)
for host_port in self.__get_cluster_host_ports(protocol)]
if protocol == 'beeswax':
target_impalad_clients = [self.client]
elif protocol == 'hs2-http':
target_impalad_clients = [self.hs2_http_client]
assert protocol == 'hs2'
target_impalad_clients = [self.hs2_client]
# Change the database to reflect the file_format, compression codec etc, or the
# user specified database for all targeted impalad.
for impalad_client in target_impalad_clients:
table_format_info, use_db, pytest.config.option.scale_factor)
def __exec_in_impala(query, user=None):
Helper to execute a query block in Impala, restoring any custom
query options after the completion of the set of queries.
# Support running multiple queries within the same test section, only verifying the
# result of the final query. The main use case is to allow for 'USE database'
# statements before a query executes, but it is not limited to that.
# TODO: consider supporting result verification of all queries in the future
result = None
target_impalad_client = choice(target_impalad_clients)
if user:
# Create a new client so the session will use the new username.
target_impalad_client = self.create_impala_client(protocol=protocol)
query_options_changed = []
for query in query.split(';'):
set_pattern_match = SET_PATTERN.match(query)
if set_pattern_match:
assert set_pattern_match.groups()[0] not in vector.get_value("exec_option"), \
"%s cannot be set in the '.test' file since it is in the test vector. " \
"Consider deepcopy()-ing the vector and removing this option in the " \
"python test." % set_pattern_match.groups()[0]
result = self.__execute_query(target_impalad_client, query, user=user)
if len(query_options_changed) > 0:
self.__restore_query_options(query_options_changed, target_impalad_client)
return result
def __exec_in_hive(query, user=None):
Helper to execute a query block in Hive. No special handling of query
options is done, since we use a separate session for each block.
h = ImpalaTestSuite.create_impala_client(HIVE_HS2_HOST_PORT, protocol='hs2',
result = None
for query in query.split(';'):
result = h.execute(query, user=user)
return result
sections = self.load_query_test_file(self.get_workload(), test_file_name,
# Assumes that it is same across all the coordinators.
lineage_log_dir = self.get_var_current_val('lineage_event_log_dir')
for test_section in sections:
if 'HIVE_MAJOR_VERSION' in test_section:
needed_hive_major_version = int(test_section['HIVE_MAJOR_VERSION'])
assert needed_hive_major_version in [2, 3]
assert HIVE_MAJOR_VERSION in [2, 3]
if needed_hive_major_version != HIVE_MAJOR_VERSION:
if 'SHELL' in test_section:
assert len(test_section) == 1, \
"SHELL test sections can't contain other sections"
cmd = self.__do_replacements(test_section['SHELL'], use_db=use_db,
extra=test_file_vars)"Shell command: " + cmd)
check_call(cmd, shell=True)
if 'QUERY' in test_section:
query_section = test_section['QUERY']
exec_fn = __exec_in_impala
elif 'HIVE_QUERY' in test_section:
query_section = test_section['HIVE_QUERY']
exec_fn = __exec_in_hive
assert 0, ('Error in test file %s. Test cases require a ' +
'-- QUERY or HIVE_QUERY section.\n%s') %\
(test_file_name, pprint.pformat(test_section))
if 'SETUP' in test_section:
self.execute_test_case_setup(test_section['SETUP'], table_format_info)
# TODO: support running query tests against different scale factors
query = QueryTestSectionReader.build_query(
self.__do_replacements(query_section, use_db=use_db, extra=test_file_vars))
if 'QUERY_NAME' in test_section:'Query Name: \n%s\n' % test_section['QUERY_NAME'])
result = None
result = exec_fn(query, user=test_section.get('USER', '').strip() or None)
user = None
if 'USER' in test_section:
user = test_section['USER'].strip()
except Exception as e:
if 'CATCH' in test_section:
self.__verify_exceptions(test_section['CATCH'], str(e), use_db)
if 'CATCH' in test_section and '__NO_ERROR__' not in test_section['CATCH']:
expected_str = self.__do_replacements(" or ".join(test_section['CATCH']).strip(),
assert False, "Expected exception: {0}\n\nwhen running:\n\n{1}".format(
expected_str, query)
assert result is not None
assert result.success, "Query failed: {0}".format(
# Decode the results read back if the data is stored with a specific encoding.
if encoding: = [row.decode(encoding) for row in]
# Replace $NAMENODE in the expected results with the actual namenode URI.
if 'RESULTS' in test_section:
# Combining 'RESULTS' with 'DML_RESULTS" is currently unsupported because
# __verify_results_and_errors calls verify_raw_results which always checks
# ERRORS, TYPES, LABELS, etc. which doesn't make sense if there are two
# different result sets to consider (IMPALA-4471).
assert 'DML_RESULTS' not in test_section
self.__verify_results_and_errors(vector, test_section, result, use_db)
# TODO: Can't validate errors without expected results for now.
assert 'ERRORS' not in test_section,\
"'ERRORS' sections must have accompanying 'RESULTS' sections"
# If --update_results, then replace references to the namenode URI with $NAMENODE.
# TODO(todd) consider running do_replacements in reverse, though that may cause
# some false replacements for things like username.
if pytest.config.option.update_results and 'RESULTS' in test_section:
test_section['RESULTS'] = test_section['RESULTS'] \
.replace(NAMENODE, '$NAMENODE') \
rt_profile_info = None
if 'RUNTIME_PROFILE_%s' % table_format_info.file_format in test_section:
# If this table format has a RUNTIME_PROFILE section specifically for it, evaluate
# that section and ignore any general RUNTIME_PROFILE sections.
rt_profile_info = 'RUNTIME_PROFILE_%s' % table_format_info.file_format
elif 'RUNTIME_PROFILE' in test_section:
rt_profile_info = 'RUNTIME_PROFILE'
if rt_profile_info is not None:
rt_profile = verify_runtime_profile(test_section[rt_profile_info],
if pytest.config.option.update_results:
test_section[rt_profile_info] = "".join(rt_profile)
if 'LINEAGE' in test_section:
# Lineage flusher thread runs every 5s by default and is not configurable. Wait
# for that period. (TODO) Get rid of this for faster test execution.
current_query_lineage = self.get_query_lineage(result.query_id, lineage_log_dir)
assert current_query_lineage is not "",\
"No lineage found for query %s in dir %s" %\
(result.query_id, lineage_log_dir)
if pytest.config.option.update_results:
test_section['LINEAGE'] = json.dumps(current_query_lineage, indent=2,
separators=(',', ': '))
verify_lineage(json.loads(test_section['LINEAGE']), current_query_lineage)
if 'DML_RESULTS' in test_section:
assert 'ERRORS' not in test_section
# The limit is specified to ensure the queries aren't unbounded. We shouldn't have
# test files that are checking the contents of tables larger than that anyways.
dml_results_query = "select * from %s limit 1000" % \
dml_result = exec_fn(dml_results_query)
verify_raw_results(test_section, dml_result,
vector.get_value('table_format').file_format, result_section='DML_RESULTS',
if pytest.config.option.update_results:
output_file = os.path.join(EE_TEST_LOGS_DIR,
test_file_name.replace('/','_') + ".test")
write_test_file(output_file, sections, encoding=encoding)
def get_query_lineage(self, query_id, lineage_dir):
"""Walks through the lineage files in lineage_dir to look for a given query_id.
This is an expensive operation is lineage_dir is large, so use carefully."""
assert lineage_dir and os.path.isdir(lineage_dir),\
"Invalid lineage dir %s" % (lineage_dir)
lineage_files = glob.glob(os.path.join(lineage_dir, 'impala_lineage_log_1.0*'))
assert len(lineage_files) > 0, "Directory %s is empty" % (lineage_dir)
# Sort by mtime. Optimized for most recently written lineages.
lineage_files.sort(key=lambda f: os.path.getmtime(f), reverse=True)
for f in lineage_files:
with open(f) as fd:
# A single file can contain a maxmimum of 5000 entries by default.
lines = fd.readlines()
for line in reversed(lines):
line = line.strip()
if len(line) == 0: continue
lineage = json.loads(line)
assert 'queryId' in lineage.keys()
if lineage['queryId'] == query_id:
return lineage
return ""
def execute_test_case_setup(self, setup_section, table_format):
Executes a test case 'SETUP' section
The test case 'SETUP' section is mainly used for insert tests. These tests need to
have some actions performed before each test case to ensure the target tables are
empty. The current supported setup actions:
RESET <table name> - Drop and recreate the table
DROP PARTITIONS <table name> - Drop all partitions from the table
setup_section = QueryTestSectionReader.build_query(setup_section)
for row in setup_section.split('\n'):
row = row.lstrip()
if row.startswith('RESET'):
db_name, table_name = QueryTestSectionReader.get_table_name_components(\
table_format, row.split('RESET')[1])
self.__reset_table(db_name, table_name)
self.client.execute("invalidate metadata " + db_name + "." + table_name)
elif row.startswith('DROP PARTITIONS'):
db_name, table_name = QueryTestSectionReader.get_table_name_components(\
table_format, row.split('DROP PARTITIONS')[1])
self.__drop_partitions(db_name, table_name)
self.client.execute("invalidate metadata " + db_name + "." + table_name)
assert False, 'Unsupported setup command: %s' % row
def change_database(cls, impala_client, table_format=None,
db_name=None, scale_factor=None):
if db_name == None:
assert table_format != None
db_name = QueryTestSectionReader.get_db_name(table_format,
scale_factor if scale_factor else '')
query = 'use %s' % db_name
# Clear the exec_options before executing a USE statement.
# The USE statement should not fail for negative exec_option tests.
def execute_wrapper(function):
Issues a use database query before executing queries.
Database names are dependent on the input format for table, which the table names
remaining the same. A use database is issued before query execution. As such,
database names need to be build pre execution, this method wraps around the different
execute methods and provides a common interface to issue the proper use command.
def wrapper(*args, **kwargs):
table_format = None
if kwargs.get('table_format'):
table_format = kwargs.get('table_format')
del kwargs['table_format']
if kwargs.get('vector'):
table_format = kwargs.get('vector').get_value('table_format')
del kwargs['vector']
# self is the implicit first argument
if table_format is not None:
args[0].change_database(args[0].client, table_format)
return function(*args, **kwargs)
return wrapper
def execute_query_expect_success(cls, impalad_client, query, query_options=None,
"""Executes a query and asserts if the query fails"""
result = cls.__execute_query(impalad_client, query, query_options, user)
assert result.success
return result
def execute_query_expect_failure(cls, impalad_client, query, query_options=None,
"""Executes a query and asserts if the query succeeds"""
result = None
result = cls.__execute_query(impalad_client, query, query_options, user)
except Exception, e:
return e
assert not result.success, "No failure encountered for query %s" % query
return result
def execute_query_unchecked(self, impalad_client, query, query_options=None, user=None):
return self.__execute_query(impalad_client, query, query_options, user)
def execute_query(self, query, query_options=None):
return self.__execute_query(self.client, query, query_options)
def execute_query_using_client(self, client, query, vector):
self.change_database(client, vector.get_value('table_format'))
query_options = vector.get_value('exec_option')
if query_options is not None: client.set_configuration(query_options)
return client.execute(query)
def execute_query_async_using_client(self, client, query, vector):
self.change_database(client, vector.get_value('table_format'))
query_options = vector.get_value('exec_option')
if query_options is not None: client.set_configuration(query_options)
return client.execute_async(query)
def close_query_using_client(self, client, query):
return client.close_query(query)
def execute_query_async(self, query, query_options=None):
if query_options is not None: self.client.set_configuration(query_options)
return self.client.execute_async(query)
def close_query(self, query):
return self.client.close_query(query)
def execute_scalar(self, query, query_options=None):
result = self.__execute_query(self.client, query, query_options)
assert len( <= 1, 'Multiple values returned from scalar'
return[0] if len( == 1 else None
def exec_and_compare_hive_and_impala_hs2(self, stmt, compare = lambda x, y: x == y):
"""Compare Hive and Impala results when executing the same statment over HS2"""
# execute_using_jdbc expects a Query object. Convert the query string into a Query
# object
query = Query()
query.query_str = stmt
# Run the statement targeting Hive
exec_opts = JdbcQueryExecConfig(impalad=HIVE_HS2_HOST_PORT, transport='SASL')
hive_results = execute_using_jdbc(query, exec_opts).data
# Run the statement targeting Impala
exec_opts = JdbcQueryExecConfig(impalad=IMPALAD_HS2_HOST_PORT, transport='NOSASL')
impala_results = execute_using_jdbc(query, exec_opts).data
# Compare the results
assert (impala_results is not None) and (hive_results is not None)
assert compare(impala_results, hive_results)
def load_query_test_file(self, workload, file_name, valid_section_names=None,
Loads/Reads the specified query test file. Accepts the given section names as valid.
Uses a default list of valid section names if valid_section_names is None.
test_file_path = os.path.join(WORKLOAD_DIR, workload, 'queries', file_name + '.test')
if not os.path.isfile(test_file_path):
assert False, 'Test file not found: %s' % file_name
return parse_query_test_file(test_file_path, valid_section_names, encoding=encoding)
def __drop_partitions(self, db_name, table_name):
"""Drops all partitions in the given table"""
for partition in self.hive_client.get_partition_names(db_name, table_name, -1):
assert self.hive_client.drop_partition_by_name(db_name, table_name, \
partition, True), 'Could not drop partition: %s' % partition
def __execute_query(cls, impalad_client, query, query_options=None, user=None):
"""Executes the given query against the specified Impalad"""
if query_options is not None: impalad_client.set_configuration(query_options)
return impalad_client.execute(query, user=user)
def __reset_table(self, db_name, table_name):
"""Resets a table (drops and recreates the table)"""
table = self.hive_client.get_table(db_name, table_name)
assert table is not None
self.hive_client.drop_table(db_name, table_name, True)
def clone_table(self, src_tbl, dst_tbl, recover_partitions, vector):
src_loc = self._get_table_location(src_tbl, vector)
self.client.execute("create external table {0} like {1} location '{2}'"\
.format(dst_tbl, src_tbl, src_loc))
if recover_partitions:
self.client.execute("alter table {0} recover partitions".format(dst_tbl))
def appx_equals(self, a, b, diff_perc):
"""Returns True if 'a' and 'b' are within 'diff_perc' percent of each other,
False otherwise. 'diff_perc' must be a float in [0,1]."""
if a == b: return True # Avoid division by 0
assert abs(a - b) / float(max(a,b)) <= diff_perc
def _get_table_location(self, table_name, vector):
""" Returns the HDFS location of the table """
result = self.execute_query_using_client(self.client,
"describe formatted %s" % table_name, vector)
for row in
if 'Location:' in row:
return row.split('\t')[1]
# This should never happen.
assert 0, 'Unable to get location for table: ' + table_name
# TODO(todd) make this use Thrift to connect to HS2 instead of shelling
# out to beeline for better performance
def run_stmt_in_hive(self, stmt, username=None):
Run a statement in Hive, returning stdout if successful and throwing
RuntimeError(stderr) if not.
# When HiveServer2 is configured to use "local" mode (i.e., MR jobs are run
# in-process rather than on YARN), Hadoop's LocalDistributedCacheManager has a
# race, wherein it tires to localize jars into
# /tmp/hadoop-$USER/mapred/local/<millis>. Two simultaneous Hive queries
# against HS2 can conflict here. Weirdly LocalJobRunner handles a similar issue
# (with the staging directory) by appending a random number. To overcome this,
# in the case that HS2 is on the local machine (which we conflate with also
# running MR jobs locally), we move the temporary directory into a unique
# directory via configuration. This workaround can be removed when
# is resolved.
# A similar workaround is used in bin/
tmpdir = None
beeline_opts = []
if pytest.config.option.hive_server2.startswith("localhost:"):
tmpdir = tempfile.mkdtemp(prefix="impala-tests-")
beeline_opts += ['--hiveconf', 'mapreduce.cluster.local.dir={0}'.format(tmpdir)]
# Remove HADOOP_CLASSPATH from environment. Beeline doesn't need it,
# and doing so avoids Hadoop 3's classpath de-duplication code from
# placing $HADOOP_CONF_DIR too late in the classpath to get the right
# log4j configuration file picked up. Some log4j configuration files
# in Hadoop's jars send logging to stdout, confusing Impala's test
# framework.
env = os.environ.copy()
env.pop("HADOOP_CLASSPATH", None)
call = subprocess.Popen(
'-u', 'jdbc:hive2://' + pytest.config.option.hive_server2,
'-n', username or getuser(),
'-e', stmt] + beeline_opts,
# Beeline in Hive 2.1 will read from stdin even when "-e"
# is specified; explicitly make sure there's nothing to
# read to avoid hanging, especially when running interactively
# with py.test.
(stdout, stderr) = call.communicate()
if call.returncode != 0:
raise RuntimeError(stderr)
return stdout
# IMPALA-8315: removing the directory may race with Hive's own cleanup and cause
# errors.
if tmpdir is not None: shutil.rmtree(tmpdir, ignore_errors=True)
def hive_partition_names(self, table_name):
"""Find the names of the partitions of a table, as Hive sees them.
The return format is a list of strings. Each string represents a partition
value of a given column in a format like 'column1=7/column2=8'.
return self.run_stmt_in_hive(
'show partitions %s' % table_name).split('\n')[1:-1]
def create_table_info_dimension(cls, exploration_strategy):
# If the user has specified a specific set of table formats to run against, then
# use those. Otherwise, load from the workload test vectors.
if pytest.config.option.table_formats:
table_formats = list()
for tf in pytest.config.option.table_formats.split(','):
dataset = get_dataset_from_workload(cls.get_workload())
table_formats.append(TableFormatInfo.create_from_string(dataset, tf))
tf_dimensions = ImpalaTestDimension('table_format', *table_formats)
tf_dimensions = load_table_info_dimension(cls.get_workload(), exploration_strategy)
# If 'skip_hbase' is specified or the filesystem is isilon, s3 or local, we don't
# need the hbase dimension.
if pytest.config.option.skip_hbase or TARGET_FILESYSTEM.lower() \
in ['s3', 'isilon', 'local', 'abfs', 'adls']:
for tf_dimension in tf_dimensions:
if tf_dimension.value.file_format == "hbase":
return tf_dimensions
def __create_exec_option_dimension(cls):
cluster_sizes = ALL_CLUSTER_SIZES
disable_codegen_options = ALL_DISABLE_CODEGEN_OPTIONS
batch_sizes = ALL_BATCH_SIZES
exec_single_node_option = [0]
if cls.exploration_strategy() == 'core':
disable_codegen_options = [False]
cluster_sizes = ALL_NODES_ONLY
return create_exec_option_dimension(cluster_sizes, disable_codegen_options,
def exploration_strategy(cls):
default_strategy = pytest.config.option.exploration_strategy
if pytest.config.option.workload_exploration_strategy:
workload_strategies = pytest.config.option.workload_exploration_strategy.split(',')
for workload_strategy in workload_strategies:
workload_strategy = workload_strategy.split(':')
if len(workload_strategy) != 2:
raise ValueError, 'Invalid workload:strategy format: %s' % workload_strategy
if cls.get_workload() == workload_strategy[0]:
return workload_strategy[1]
return default_strategy
def wait_for_state(self, handle, expected_state, timeout):
"""Waits for the given 'query_handle' to reach the 'expected_state'. If it does not
reach the given state within 'timeout' seconds, the method throws an AssertionError.
self.wait_for_any_state(handle, [expected_state], timeout)
def wait_for_any_state(self, handle, expected_states, timeout):
"""Waits for the given 'query_handle' to reach one of 'expected_states'. If it does
not reach one of the given states within 'timeout' seconds, the method throws an
AssertionError. Returns the final state.
start_time = time.time()
actual_state = self.client.get_state(handle)
while actual_state not in expected_states and time.time() - start_time < timeout:
actual_state = self.client.get_state(handle)
if actual_state not in expected_states:
raise Timeout("query {0} did not reach one of the expected states {1}, "
"last known state {2}".format(handle.get_handle().id, expected_states,
return actual_state
def wait_for_db_to_appear(self, db_name, timeout_s):
"""Wait until the database with 'db_name' is present in the impalad's local catalog.
Fail after timeout_s if the doesn't appear."""
start_time = time.time()
while time.time() - start_time < timeout_s:
# This will throw an exception if the database is not present.
self.client.execute("describe database `{db_name}`".format(db_name=db_name))
except Exception:
raise Exception("DB {0} didn't show up after {1}s", db_name, timeout_s)
def wait_for_table_to_appear(self, db_name, table_name, timeout_s):
"""Wait until the table with 'table_name' in 'db_name' is present in the
impalad's local catalog. Fail after timeout_s if the doesn't appear."""
start_time = time.time()
while time.time() - start_time < timeout_s:
# This will throw an exception if the table is not present.
self.client.execute("describe `{db_name}`.`{table_name}`".format(
db_name=db_name, table_name=table_name))
except Exception, ex:
print str(ex)
raise Exception("Table {0}.{1} didn't show up after {2}s", db_name, table_name,
def assert_eventually(self, timeout_s, period_s, condition, error_msg=None):
"""Assert that the condition (a function with no parameters) returns True within the
given timeout. The condition is executed every period_s seconds. The check assumes
that once the condition returns True, it continues to return True. Throws a Timeout
if the condition does not return true within timeout_s seconds. 'error_msg' is an
optional function that must return a string. If set, the result of the function will
be included in the Timeout error message."""
count = 0
start_time = time.time()
while not condition() and time.time() - start_time < timeout_s:
count += 1
if not condition():
error_msg_str = " error message: " + error_msg() if error_msg else ""
raise Timeout(
"Check failed to return True after {0} tries and {1} seconds{2}".format(
count, timeout_s, error_msg_str))
def assert_impalad_log_contains(self, level, line_regex, expected_count=1):
Convenience wrapper around assert_log_contains for impalad logs.
self.assert_log_contains("impalad", level, line_regex, expected_count)
def assert_catalogd_log_contains(self, level, line_regex, expected_count=1):
Convenience wrapper around assert_log_contains for catalogd logs.
self.assert_log_contains("catalogd", level, line_regex, expected_count)
def assert_log_contains(self, daemon, level, line_regex, expected_count=1):
Assert that the daemon log with specified level (e.g. ERROR, WARNING, INFO) contains
expected_count lines with a substring matching the regex. When expected_count is -1,
at least one match is expected.
When using this method to check log files of running processes, the caller should
make sure that log buffering has been disabled, for example by adding
'-logbuflevel=-1' to the daemon startup options.
pattern = re.compile(line_regex)
found = 0
if hasattr(self, "impala_log_dir"):
log_dir = self.impala_log_dir
log_dir = EE_TEST_LOGS_DIR
log_file_path = os.path.join(log_dir, daemon + "." + level)
# Resolve symlinks to make finding the file easier.
log_file_path = os.path.realpath(log_file_path)
with open(log_file_path) as log_file:
for line in log_file:
found += 1
if expected_count == -1:
assert found > 0, "Expected at least one line in file %s matching regex '%s'"\
", but found none." % (log_file_path, line_regex)
assert found == expected_count, "Expected %d lines in file %s matching regex '%s'"\
", but found %d lines. Last line was: \n%s" %\
(expected_count, log_file_path, line_regex, found, line)