Method test_async_load

tests/metadata/test_load.py:162–236

(self, vector, unique_database)

  # query at the backend through beewax or hs2 clients. The objective is to assure
  # the load query completes successfully.
  def test_async_load(self, vector, unique_database):
    enable_async_load_data = vector.get_value('enable_async_load_data_execution')
    protocol = vector.get_value('protocol')
    client = self.create_impala_client(protocol=protocol)

    # Form a fully qualified table name with '-' in protocol 'hs2-http' dropped as
    # '-' is not allowed in Impala table name even delimited with ``.
    qualified_table_name = '{0}.{1}_{2}_{3}'.format(unique_database, TEST_TBL_NOPART,
        protocol if protocol != 'hs2-http' else 'hs2http', enable_async_load_data)

    # Form a staging path that is protocol and enable_async_load_data dependent to
    # allow parallel creating distinct HDFS directories for each test object.
    staging_path = "{0}_{1}_{2}".format(STAGING_PATH, protocol, enable_async_load_data)

    # Put some data into the staging path
    self.filesystem_client.delete_file_dir(staging_path, recursive=True)
    self.filesystem_client.make_dir(staging_path, permission=777)
    self.filesystem_client.copy(ALLTYPES_PATH, "{0}/100101.txt".format(staging_path))

    # Create a table with the staging path
    self.client.execute("create table {0} like functional.alltypesnopart \
        location \'{1}\'".format(qualified_table_name, staging_path))

    try:

      # The load data is going to need the metadata of the table. To avoid flakiness
      # about metadata loading, this selects from the table first to get the metadata
      # loaded.
      self.execute_query_expect_success(client,
          "select count(*) from {0}".format(qualified_table_name))

      # Configure whether to use async LOAD and add an appropriate delay of 3 seconds
      new_vector = deepcopy(vector)
      new_vector.get_value('exec_option')['enable_async_load_data_execution'] = \
           enable_async_load_data
      delay = "CRS_DELAY_BEFORE_LOAD_DATA:SLEEP@3000"
      new_vector.get_value('exec_option')['debug_action'] = "{0}".format(delay)
      load_stmt = "load data inpath \'{1}\' \
          into table {0}".format(qualified_table_name, staging_path)
      exec_start = time.time()
      handle = self.execute_query_async_using_client(client, load_stmt, new_vector)
      exec_end = time.time()
      exec_time = exec_end - exec_start
      exec_end_state = client.get_impala_exec_state(handle)

      # Wait for the statement to finish with a timeout of 20 seconds
      # (30 seconds without shortcircuit reads)
      wait_time = 20 if IS_HDFS else 30
      wait_start = time.time()
      client.wait_for_impala_state(handle, FINISHED, wait_time)
      wait_end = time.time()
      wait_time = wait_end - wait_start
      self.close_query_using_client(client, handle)
      if enable_async_load_data:
        # In async mode:
        #  The compilation of LOAD is processed in the exec step without delay. And the
        #  processing of the LOAD plan is in wait step with delay. Relax the wait time
        #  to at least 2 seconds because wait_start not strictly starts at the point