Method test_ctas

tests/metadata/test_ddl.py:1193–1248

(self, vector, unique_database)

      client.close()

  def test_ctas(self, vector, unique_database):
    enable_async_ddl = vector.get_value('enable_async_ddl_execution')
    client = self.create_impala_client(protocol=vector.get_value('protocol'))

    try:
      # The CTAS is going to need the metadata of the source table in the
      # select. To avoid flakiness about metadata loading, this selects from
      # that source table first to get the metadata loaded.
      self.execute_query_expect_success(client,
          "select count(*) from functional_parquet.alltypes")

      # Configure whether to use async DDL and add appropriate delays
      new_vector = deepcopy(vector)
      new_vector.get_value('exec_option')['enable_async_ddl_execution'] = enable_async_ddl
      create_delay = "CRS_DELAY_BEFORE_CATALOG_OP_EXEC:SLEEP@10000"
      insert_delay = "CRS_BEFORE_COORD_STARTS:SLEEP@2000"
      new_vector.get_value('exec_option')['debug_action'] = \
          "{0}|{1}".format(create_delay, insert_delay)
      dest_tbl = "{0}.ctas_test".format(unique_database)
      source_tbl = "functional_parquet.alltypes"
      ctas_stmt = 'create external table {0} as select * from {1}'.format(
          dest_tbl, source_tbl)
      exec_start = time.time()
      handle = self.execute_query_async_using_client(client, ctas_stmt, new_vector)
      exec_end = time.time()
      exec_time = exec_end - exec_start
      # The CRS_BEFORE_COORD_STARTS delay postpones the transition from PENDING
      # to RUNNING, so the sync case should be in PENDING state at the end of
      # the execute call. This means that the sync and async cases are the same.
      assert client.is_pending(handle)

      # Wait for the statement to finish with a timeout of 40 seconds
      # (60 seconds without shortcircuit reads). There are other tests running
      # in parallel and ASAN can be slow, so this timeout has been bumped
      # substantially to avoid flakiness. The actual test case does not depend
      # on the statement finishing in a particular amount of time.
      wait_time = 40 if IS_HDFS else 60
      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)
      # In sync mode:
      #  The entire CTAS is processed in the exec step with delay. exec_time should be
      #  more than 10 seconds.
      #
      # In async mode:
      #  The compilation of CTAS is processed in the exec step without delay. And the
      #  processing of the CTAS plan is in wait step with delay. The wait time should
      #  definitely take more time than 10 seconds.
      if enable_async_ddl:
        assert(wait_time >= 10)
      else:
        assert(exec_time >= 10)
    finally:
      client.close()