Class QueryResultDiffSearcher

tests/comparison/discrepancy_searcher.py:613–812

This class uses the query generator (query_generator.py) along with the query profile (query_profile.py) to randomly generate queries then executes the queries on the reference and test databases, then compares the results.

class QueryResultDiffSearcher(object):
  '''This class uses the query generator (query_generator.py) along with the
     query profile (query_profile.py) to randomly generate queries then executes the
     queries on the reference and test databases, then compares the results.
  '''

  # Sometimes things get into a bad state and the same error loops forever
  ABORT_ON_REPEAT_ERROR_COUNT = 2

  COPY_TABLE_SUFFIX = '__qgen_copy'

  def __init__(self, query_profile, ref_conn, test_conn):
    '''query_profile should be an instance of one of the profiles in query_profile.py'''
    self.query_profile = query_profile
    self.ref_conn = ref_conn
    self.test_conn = test_conn
    with ref_conn.cursor() as ref_cursor:
      with test_conn.cursor() as test_cursor:
        self.common_tables = DbCursor.describe_common_tables([ref_cursor, test_cursor])
        if not self.common_tables:
          raise Exception("Unable to find a common set of tables in both databases")

  def _concurrently_copy_table(self, src_table):
    """
    Given a Table object, create another Table with the same schema and return the new
    Table object.  The schema will be created in both the test and reference databases.

    The data is then copied in both the ref and test databases using threads.
    """
    with test_conn.cursor() as test_cursor:
      test_cursor.execute('SHOW CREATE TABLE {0}'.format(src_table.name))
      (create_table_sql,) = test_cursor.fetchall()[0]
      new_table_name = src_table.name + self.COPY_TABLE_SUFFIX
      create_table_sql = create_table_sql.replace(src_table.name, new_table_name, 1)
      test_cursor.drop_table(new_table_name)
      test_cursor.execute(create_table_sql)
      new_table = test_cursor.describe_table(new_table_name)
    with ref_conn.cursor() as ref_cursor:
      ref_cursor.drop_table(new_table_name)
      ref_cursor.create_table(new_table)

    copy_select_query = Query()
    copy_select_query.select_clause = SelectClause(
        [SelectItem(col) for col in src_table.cols])
    copy_select_query.from_clause = FromClause(src_table)

    if new_table.primary_keys:
      conflict_action = InsertClause.CONFLICT_ACTION_IGNORE
    else:
      conflict_action = InsertClause.CONFLICT_ACTION_DEFAULT

    table_copy_statement = InsertStatement(
        insert_clause=InsertClause(new_table, conflict_action=conflict_action),
        select_query=copy_select_query, execution=StatementExecutionMode.DML_SETUP)

    result = self.query_result_comparator.compare_query_results(table_copy_statement)
    if result.error:
      raise Exception('setup SQL to copy table failed: {0}'.format(result.error))