blob: 0bd65df8b832ccf78737d3fa0d911d4e37557da7 [file] [log] [blame]
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you 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
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import pytest
import re
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.test_vector import ImpalaTestDimension
from tests.util.parse_util import parse_duration_string_ms
class TestRowsAvailability(ImpalaTestSuite):
"""Tests that the 'Rows available' timeline event is marked only after rows are
truly available. We mark the 'Rows available' event once we advance the query
status to a 'ready' state; this signals to the client that rows can be fetched.
Long fetch times can trigger client timeouts at various levels (socket, app, etc.).
This is a regression test against IMPALA-924."""
# These queries are chosen to have different plan roots in the coordinator's fragment.
# The WHERE clause is carefully crafted to control when result rows become available at
# the coordinator. The selected partition of 'functional.alltypestiny' has exactly
# two rows stored in a single file. In the scan node we sleep one second for each
# result row. Therefore, result rows can become available no earlier that after 2s.
TABLE = 'functional.alltypestiny'
WHERE_CLAUSE = 'where month = 1 and bool_col = sleep(1000)'
QUERIES = ['select * from %s %s' % (TABLE, WHERE_CLAUSE),
'select * from %s %s order by id limit 1' % (TABLE, WHERE_CLAUSE),
'select * from %s %s order by id' % (TABLE, WHERE_CLAUSE),
'select count(*) from %s %s' % (TABLE, WHERE_CLAUSE),
'select 1 union all select count(*) from %s %s' % (TABLE, WHERE_CLAUSE),
'select count(*) over () from %s %s' % (TABLE, WHERE_CLAUSE)]
def get_workload(self):
return 'functional-query'
def add_test_dimensions(cls):
super(TestRowsAvailability, cls).add_test_dimensions()
cls.ImpalaTestMatrix.add_dimension(ImpalaTestDimension('query', *cls.QUERIES))
cls.ImpalaTestMatrix.add_constraint(lambda v: cls.__is_valid_test_vector(v))
def __is_valid_test_vector(cls, vector):
return vector.get_value('table_format').file_format == 'text' and\
vector.get_value('table_format').compression_codec == 'none' and\
vector.get_value('exec_option')['batch_size'] == 0 and\
vector.get_value('exec_option')['disable_codegen'] == False and\
vector.get_value('exec_option')['num_nodes'] == 0
def test_rows_availability(self, vector):
# This test is run serially because it requires the query to come back within
# some amount of time. Running this with other tests makes it hard to bound
# that time.
query = vector.get_value('query')
# Execute async to get a handle. Wait until the query has completed.
handle = self.execute_query_async(query, vector.get_value('exec_option'))
self.impalad_test_service.wait_for_query_state(self.client, handle,
self.client.QUERY_STATES['FINISHED'], timeout=20)
profile = self.client.get_runtime_profile(handle)
start_time_ms = None
rows_avail_time_ms = None
for line in profile.split("\n"):
if "Ready to start on" in line:
start_time_ms = parse_duration_string_ms(self.__find_time(line))
elif "Rows available:" in line:
rows_avail_time_ms = parse_duration_string_ms(self.__find_time(line))
if start_time_ms is None:
assert False, "Failed to find the 'Ready to start' timeline event in the " \
"query profile:\n%s" % profile
if rows_avail_time_ms is None:
assert False, "Failed to find the 'Rows available' timeline event in the " \
"query profile:\n%s" % profile
time_diff = rows_avail_time_ms - start_time_ms
assert time_diff >= self.ROWS_AVAIL_LOWER_BOUND_MS,\
"The 'Rows available' timeline event was marked prematurely %sms after the "\
"'Ready to start' event.\nExpected the event to be marked no earlier than "\
"%sms after the 'Ready to start' event.\nQuery: %s"\
% (time_diff, self.ROWS_AVAIL_LOWER_BOUND_MS, query)
def __find_time(line):
"""Find event time point in a line from the runtime profile timeline."""
# Given line "- Rows available: 3s311ms (2s300ms)", this function returns "3s311ms"
match =': (.*) \(', line)
if match is None:
assert False, "Failed to find time in runtime profile"