Method test_catalog_operation_fields

tests/webserver/test_web_pages.py:1148–1204

Verify the CREATE_DATABASE operation is consistent with the statement shown in the /queries page of impalad.

(self, unique_database)

    assert page.status_code == requests.codes.ok

  def test_catalog_operation_fields(self, unique_database):
    """Verify the CREATE_DATABASE operation is consistent with the statement shown in the
       /queries page of impalad."""
    catalog_operations_page = requests.get("http://localhost:25020/operations?json").text
    catalog_operations = json.loads(catalog_operations_page)
    assert "finished_catalog_operations" in catalog_operations

    queries_page = requests.get("http://localhost:25000/queries?json").text
    queries = json.loads(queries_page)
    assert "completed_queries" in queries

    # Find the CREATE_DATABASE operation in catalogd
    ts_format = "%Y-%m-%d %H:%M:%S.%f"
    found = False
    for op in catalog_operations["finished_catalog_operations"]:
      if op["target_name"] == unique_database \
          and op["catalog_op_name"] == "CREATE_DATABASE":
        catalog_op_query_id = op["query_id"]
        catalog_op_user = op["user"]
        catalog_op_start_time = datetime.strptime(op["start_time"], ts_format)
        catalog_op_end_time = datetime.strptime(op["finish_time"], ts_format)
        catalog_op_duration = parse_duration_string_ms(op["duration"])
        found = True
        LOG.info("Found query id in catalog operations: " + catalog_op_query_id)
        break
    assert found

    def verify_query_record(query):
      assert "CREATE DATABASE" in query["stmt"]
      assert unique_database in query["stmt"]
      assert catalog_op_user == query["effective_user"]
      assert datetime.strptime(query["start_time"], ts_format) <= catalog_op_start_time
      assert datetime.strptime(query["end_time"], ts_format) >= catalog_op_end_time
      assert parse_duration_string_ms(query["duration"]) >= catalog_op_duration

    # Find the query in impalad
    matched = False
    for query in queries["completed_queries"]:
      if query["query_id"] == catalog_op_query_id:
        verify_query_record(query)
        matched = True
    if not matched:
      LOG.info("Query id {0} not found in the completed queries list".format(
          catalog_op_query_id))
      # Try to find the query in the in-flight queries list. It could be waiting to
      # be closed.
      for query in queries["in_flight_queries"]:
        if query["query_id"] == catalog_op_query_id:
          verify_query_record(query)
          matched = True

    # Dump web pages for debug
    if not matched:
      LOG.info("Query id {0} not found in queries page!".format(catalog_op_query_id))
      LOG.info("Catalog operations: " + catalog_operations_page)
      LOG.info("Queries: " + queries_page)
    assert matched