Update test logging to print executable SQL statements and log all actions executed

This is the first step in cleaning up the test logging. It provides a common connection
interface that provides tracing around all operations. When a test fails the output will
be executable SQL. It also logs actions such as when a connection is opened, close, or
when an operation is cancelled. Currently only beeswax connections are supported, but
I have a seperate patch that adds support for executing using HS2 as well as Beeswax.

Example of new logging:
-- connecting to: localhost:21000
-- executing against localhost:21000
use functional;

SET disable_codegen=False;
SET abort_on_error=1;
SET batch_size=0;
SET num_nodes=0;

-- executing against localhost:21000
select a.timestamp_col from alltypessmall a inner join alltypessmall b on
(a.timestamp_col = b.timestamp_col)
where a.year=2009 and a.month=1 and b.year=2009 and b.month=1;
-- closing connection to: localhost:21000

Change-Id: Iedc7d4d3a84bfeff6cc1daae6ed1ca97613d7700
Reviewed-on: http://gerrit.ent.cloudera.com:8080/1133
Tested-by: jenkins
Reviewed-by: Lenni Kuff <lskuff@cloudera.com>

Author: lskuff

testdata/workloads/functional-query/queries/QueryTest/hdfs-scan-node.test

@@ -701,8 +701,8 @@ bigint
 ---- RESULTS
 10000
 ====
-# Select from old (pre-hive 9) rc file table
 ---- QUERY
+# Select from old (pre-hive 9) rc file table
 SELECT * FROM functional_rc.old_rcfile_table
 ---- TYPES
 int, string

tests/beeswax/impala_beeswax.py

@@ -95,7 +95,7 @@ def __options_to_string_list(self):
     return ["%s=%s" % (k,v) for (k,v) in self.__query_options.iteritems()]
 
   def get_query_options(self):
-    return '\n'.join(["\t%s: %s" % (k,v) for (k,v) in self.__query_options.iteritems()])
+    return self.__query_options
 
   def set_query_option(self, name, value):
     self.__query_options[name.upper()] = value
@@ -192,9 +192,15 @@ def wait_for_completion(self, query_handle):
         raise ImpalaBeeswaxException("Query aborted:" + error_log, None)
       time.sleep(0.05)
 
+  def get_default_configuration(self):
+    return self.__do_rpc(lambda: self.imp_service.get_default_configuration(False))
+
   def get_state(self, query_handle):
     return self.__do_rpc(lambda: self.imp_service.get_state(query_handle))
 
+  def get_log(self, query_handle):
+    return self.__do_rpc(lambda: self.imp_service.get_log(query_handle))
+
   def refresh(self):
     """Invalidate the Impalad catalog"""
     return self.execute("invalidate metadata")

tests/common/impala_connection.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python
+# Copyright (c) 2012 Cloudera, Inc. All rights reserved.
+#
+# Licensed 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.
+#
+# Common for connections to Impala. Currently supports Beeswax connections and
+# in the future will support HS2 connections. Provides tracing around all
+# operations.
+
+from tests.beeswax.impala_beeswax import ImpalaBeeswaxClient, QueryResult
+from thrift.transport.TSocket import TSocket
+from thrift.protocol import TBinaryProtocol
+from thrift.transport.TTransport import TBufferedTransport, TTransportException
+from getpass import getuser
+
+import abc
+import logging
+import os
+
+LOG = logging.getLogger('impala_connection')
+console_handler = logging.StreamHandler()
+console_handler.setLevel(logging.INFO)
+# All logging needs to be either executable SQL or a SQL comment (prefix with --).
+console_handler.setFormatter(logging.Formatter('%(message)s'))
+LOG.addHandler(console_handler)
+LOG.propagate = False
+
+# Common wrapper around the internal types of HS2/Beeswax operation/query handles.
+class OperationHandle(object):
+  def __init__(self, handle):
+    self.__handle = handle
+
+  def get_handle(self): return self.__handle
+
+
+# Represents an Impala connection.
+class ImpalaConnection(object):
+  __metaclass__ = abc.ABCMeta
+
+  @abc.abstractmethod
+  def set_configuration_option(self, name, value):
+    """Sets a configuraiton option name to the given value"""
+    pass
+
+  @abc.abstractmethod
+  def get_configuration(self):
+    """Returns the configuration (a dictionary of key-value pairs) for this connection"""
+    pass
+
+  @abc.abstractmethod
+  def set_configuration(self, configuration_option_dict):
+    """Replaces existing configuration with the given dictionary"""
+    pass
+
+  @abc.abstractmethod
+  def clear_configuration(self):
+    """Clears all existing configuration."""
+    pass
+
+  @abc.abstractmethod
+  def connect(self):
+    """Opens the connection"""
+    pass
+
+  @abc.abstractmethod
+  def close(self):
+    """Closes the connection. Can be called multiple times"""
+    pass
+
+  @abc.abstractmethod
+  def get_state(self, operation_handle):
+    """Returns the state of a query"""
+    pass
+
+  @abc.abstractmethod
+  def get_log(self, operation_handle):
+    """Returns the log of an operation"""
+    pass
+
+  @abc.abstractmethod
+  def cancel(self, operation_handle):
+    """Cancels an in-flight operation"""
+    pass
+
+  def execute(self, sql_stmt):
+    """Executes a query and fetches the results"""
+    pass
+
+  @abc.abstractmethod
+  def execute_async(self, sql_stmt):
+    """Issues a query and returns the handle to the caller for processing"""
+    pass
+
+  @abc.abstractmethod
+  def fetch(self, sql_stmt, operation_handle, batch_size=1024):
+    """Fetches all query results given a handle and sql statement.
+    TODO: Support fetching single batch"""
+    pass
+
+
+# Represents a connection to Impala using the Beeswax API.
+class BeeswaxConnection(ImpalaConnection):
+  def __init__(self, host_port, use_kerberos=False):
+    self.__beeswax_client = ImpalaBeeswaxClient(host_port, use_kerberos)
+    self.__host_port = host_port
+    self.QUERY_STATES = self.__beeswax_client.query_states
+
+  def set_configuration_option(self, name, value):
+    # Only set the option if it's not already set to the same value.
+    if self.__beeswax_client.get_query_option(name) != value:
+      LOG.info('SET %s=%s;' % (name, value))
+      self.__beeswax_client.set_query_option(name, value)
+
+  def get_configuration(self):
+    return self.__beeswax_client.get_query_options
+
+  def set_configuration(self, config_option_dict):
+    assert config_option_dict is not None, "config_option_dict cannot be None"
+    self.clear_configuration()
+    for name, value in config_option_dict.iteritems():
+      self.set_configuration_option(name, value)
+
+  def clear_configuration(self):
+    self.__beeswax_client.clear_query_options()
+
+  def connect(self):
+    LOG.info("-- connecting to: %s" % self.__host_port)
+    self.__beeswax_client.connect()
+
+  def close(self):
+    LOG.info("-- closing connection to: %s" % self.__host_port)
+    self.__beeswax_client.close_connection()
+
+  def execute(self, sql_stmt):
+    LOG.info("-- executing against %s\n%s;\n" % (self.__host_port, sql_stmt))
+    return self.__beeswax_client.execute(sql_stmt)
+
+  def execute_async(self, sql_stmt):
+    LOG.info("-- executing async: %s\n%s;\n" % (self.__host_port, sql_stmt))
+    return OperationHandle(self.__beeswax_client.execute_query_async(sql_stmt))
+
+  def cancel(self, operation_handle):
+    LOG.info("-- canceling operation: %s" % operation_handle)
+    return self.__beeswax_client.cancel_query(operation_handle.get_handle())
+
+  def get_state(self, operation_handle):
+    LOG.info("-- getting state for operation: %s" % operation_handle)
+    return self.__beeswax_client.get_state(operation_handle.get_handle())
+
+  def get_log(self, operation_handle):
+    LOG.info("-- getting log for operation: %s" % operation_handle)
+    return self.__beeswax_client.get_log(operation_handle.get_handle())
+
+  def refresh(self):
+    """Invalidate the Impalad catalog"""
+    return self.execute("invalidate metadata")
+
+  def refresh_table(self, db_name, table_name):
+    """Refresh a specific table from the catalog"""
+    return self.execute("refresh %s.%s" % (db_name, table_name))
+
+  def fetch(self, sql_stmt, operation_handle):
+    LOG.info("-- fetching results from: %s" % operation_handle)
+    return self.__beeswax_client.fetch_results(sql_stmt, operation_handle.get_handle())
+
+def create_connection(host_port, use_kerberos=False):
+  # TODO: Support HS2 connections.
+  return BeeswaxConnection(host_port=host_port, use_kerberos=use_kerberos)

tests/common/impala_service.py

@@ -25,7 +25,7 @@
 
 from collections import defaultdict
 from HTMLParser import HTMLParser
-from tests.beeswax.impala_beeswax import ImpalaBeeswaxClient
+from tests.common.impala_connection import ImpalaConnection, create_connection
 from time import sleep, time
 
 logging.basicConfig(level=logging.ERROR, format='%(threadName)s: %(message)s')
@@ -159,8 +159,7 @@ def wait_for_query_status(self, client, query_id, expected_content,
 
   def create_beeswax_client(self, use_kerberos=False):
     """Creates a new beeswax client connection to the impalad"""
-    client =\
-        ImpalaBeeswaxClient('%s:%d' % (self.hostname, self.beeswax_port), use_kerberos)
+    client = create_connection('%s:%d' % (self.hostname, self.beeswax_port), use_kerberos)
     client.connect()
     return client
 

tests/common/impala_test_suite.py

@@ -21,8 +21,8 @@
 import pytest
 from functools import wraps
 from random import choice
-from tests.beeswax.impala_beeswax import ImpalaBeeswaxClient
 from tests.common.impala_service import ImpaladService
+from tests.common.impala_connection import ImpalaConnection, create_connection
 from tests.common.test_dimensions import *
 from tests.common.test_result_verifier import *
 from tests.common.test_vector import *
@@ -38,7 +38,7 @@
 from thrift.transport import TTransport, TSocket
 from thrift.protocol import TBinaryProtocol
 
-logging.basicConfig(level=logging.INFO, format='%(threadName)s: %(message)s')
+logging.basicConfig(level=logging.INFO, format='-- %(message)s')
 LOG = logging.getLogger('impala_test_suite')
 
 IMPALAD_HOST_PORT_LIST = pytest.config.option.impalad.split(',')
@@ -80,7 +80,7 @@ def setup_class(cls):
     cls.hive_client = ThriftHiveMetastore.Client(protocol)
     cls.hive_transport.open()
 
-    # The ImpalaBeeswaxClient is used to execute queries in the test suite
+    # Create a connection to Impala.
     cls.client = cls.create_impala_client(IMPALAD)
 
     cls.impalad_test_service = ImpaladService(IMPALAD.split(':')[0])
@@ -98,19 +98,19 @@ def teardown_class(cls):
       cls.hive_transport.close()
 
     if cls.client:
-      cls.client.close_connection()
+      cls.client.close()
 
   @classmethod
   def create_impala_client(cls, host_port=IMPALAD):
-    client = ImpalaBeeswaxClient(host_port,
+    client = create_connection(host_port=host_port,
         use_kerberos=pytest.config.option.use_kerberos)
     client.connect()
     return client
 
   def cleanup_db(self, db_name):
     # To drop a db, we need to first drop all the tables in that db
     self.client.execute("use default")
-    self.client.set_query_options({'sync_ddl': 1})
+    self.client.set_configuration({'sync_ddl': 1})
     if db_name in self.client.execute("show databases", ).data:
       for tbl_name in self.client.execute("show tables in " + db_name).data:
         full_tbl_name = '%s.%s' % (db_name, tbl_name)
@@ -148,6 +148,7 @@ def run_test_case(self, test_file_name, vector, use_db=None, multiple_impalad=Fa
     # user specified database for all targeted impalad.
     for impalad_client in target_impalad_clients:
       ImpalaTestSuite.change_database(impalad_client, table_format_info, use_db)
+      impalad_client.set_configuration(exec_options)
 
     sections = self.load_query_test_file(self.get_workload(), test_file_name)
     for test_section in sections:
@@ -172,7 +173,7 @@ def run_test_case(self, test_file_name, vector, use_db=None, multiple_impalad=Fa
       target_impalad_client = choice(target_impalad_clients)
       for query in query.split(';'):
         result =\
-            self.execute_query_expect_success(target_impalad_client, query, exec_options)
+            self.execute_query_expect_success(target_impalad_client, query)
       assert result is not None
 
       verify_raw_results(test_section, result,
@@ -219,7 +220,7 @@ def change_database(cls, impala_client, table_format=None, db_name=None):
     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.
-    impala_client.clear_query_options()
+    impala_client.clear_configuration()
     impala_client.execute(query)
 
   def execute_wrapper(function):
@@ -247,28 +248,28 @@ def wrapper(*args, **kwargs):
     return wrapper
 
   @execute_wrapper
-  def execute_query_expect_success(self, impalad_client, query, query_exec_options=None):
+  def execute_query_expect_success(self, impalad_client, query, query_options=None):
     """Executes a query and asserts if the query fails"""
-    result = self.__execute_query(impalad_client, query, query_exec_options)
+    result = self.__execute_query(impalad_client, query, query_options)
     assert result.success
     return result
 
   @execute_wrapper
-  def execute_query(self, query, query_exec_options=None):
-    return self.__execute_query(self.client, query, query_exec_options)
+  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'))
     return client.execute(query)
 
   @execute_wrapper
-  def execute_query_async(self, query, query_exec_options=None):
-    self.__set_exec_options(self.client, query_exec_options)
-    return self.client.execute_query_async(query)
+  def execute_query_async(self, query, query_options=None):
+    self.client.set_configuration(query_options)
+    return self.client.execute_async(query)
 
   @execute_wrapper
-  def execute_scalar(self, query, query_exec_options=None):
-    result = self.__execute_query(self.client, query, query_exec_options)
+  def execute_scalar(self, query, query_options=None):
+    result = self.__execute_query(self.client, query, query_options)
     assert len(result.data) <= 1, 'Multiple values returned from scalar'
     return result.data[0] if len(result.data) == 1 else None
 
@@ -306,25 +307,18 @@ def __drop_partitions(self, db_name, table_name):
     for partition in self.hive_client.get_partition_names(db_name, table_name, 0):
       self.hive_client.drop_partition_by_name(db_name, table_name, partition, True)
 
-  def __execute_query(self, impalad_client, query, query_exec_options=None):
+  def __execute_query(self, impalad_client, query, query_options=None):
     """Executes the given query against the specified Impalad"""
-    LOG.info('Executing Query(%s): \n%s\n' % (impalad_client.impalad, query))
-    self.__set_exec_options(impalad_client, query_exec_options)
+    if query_options is not None: impalad_client.set_configuration(query_options)
     return impalad_client.execute(query)
 
-  def __execute_query_new_client(self, query, query_exec_options=None,
+  def __execute_query_new_client(self, query, query_options=None,
       use_kerberos=False):
     """Executes the given query against the specified Impalad"""
     new_client = self.create_impala_client()
-    self.__set_exec_options(new_client, query_exec_options)
+    new_client.set_configuration(query_options)
     return new_client.execute(query)
 
-  def __set_exec_options(self, impalad_client, query_exec_options):
-    # Set the specified query exec options, if specified
-    impalad_client.clear_query_options()
-    if query_exec_options is not None:
-      impalad_client.set_query_options(query_exec_options)
-
   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)

tests/common/test_result_verifier.py

@@ -153,6 +153,8 @@ def __eq__(self, other):
       return compare_float(float(self.value), float(other.value), 10e-5)
     elif self.column_type == 'double':
       return compare_float(float(self.value), float(other.value), 10e-10)
+    elif self.column_type == 'boolean':
+      return str(self.value).lower() == str(other.value).lower()
     else:
       return self.value == other.value
 

tests/conftest.py

@@ -53,7 +53,7 @@ def pytest_addoption(parser):
                    "for the hive metastore client when using kerberos.")
 
   parser.addoption("--use_kerberos", action="store_true", default=False,
-                   help="Use kerberos transport for running tests")
+                   help="use kerberos transport for running tests")
 
   parser.addoption("--sanity", action="store_true", default=False,
                    help="Runs a single test vector from each test to provide a quick "\