| # 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 |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| from collections import defaultdict |
| from datetime import datetime |
| from tests.beeswax.impala_beeswax import ImpalaBeeswaxException |
| from tests.common.impala_test_suite import ImpalaTestSuite |
| from tests.common.skip import (SkipIfS3, SkipIfABFS, SkipIfADLS, SkipIfIsilon, |
| SkipIfLocal, SkipIfNotHdfsMinicluster) |
| from tests.util.filesystem_utils import IS_EC |
| from time import sleep |
| from RuntimeProfile.ttypes import TRuntimeProfileFormat |
| import pytest |
| import re |
| |
| |
| class TestObservability(ImpalaTestSuite): |
| @classmethod |
| def get_workload(self): |
| return 'functional-query' |
| |
| def test_merge_exchange_num_rows(self): |
| """Regression test for IMPALA-1473 - checks that the exec summary for a merging |
| exchange with a limit reports the number of rows returned as equal to the limit, |
| and that the coordinator fragment portion of the runtime profile reports the number |
| of rows returned correctly.""" |
| query = """select tinyint_col, count(*) from functional.alltypes |
| group by tinyint_col order by tinyint_col limit 5""" |
| result = self.execute_query(query) |
| exchange = result.exec_summary[1] |
| assert exchange['operator'] == '05:MERGING-EXCHANGE' |
| assert exchange['num_rows'] == 5 |
| assert exchange['est_num_rows'] == 5 |
| assert exchange['peak_mem'] > 0 |
| |
| for line in result.runtime_profile.split('\n'): |
| # The first 'RowsProduced' we find is for the coordinator fragment. |
| if 'RowsProduced' in line: |
| assert '(5)' in line |
| break |
| |
| def test_broadcast_num_rows(self): |
| """Regression test for IMPALA-3002 - checks that the num_rows for a broadcast node |
| in the exec summary is correctly set as the max over all instances, not the sum.""" |
| query = """select distinct a.int_col, a.string_col from functional.alltypes a |
| inner join functional.alltypessmall b on (a.id = b.id) |
| where a.year = 2009 and b.month = 2""" |
| result = self.execute_query(query) |
| exchange = result.exec_summary[8] |
| assert exchange['operator'] == '04:EXCHANGE' |
| assert exchange['num_rows'] == 25 |
| assert exchange['est_num_rows'] == 25 |
| assert exchange['peak_mem'] > 0 |
| |
| def test_report_time(self): |
| """ Regression test for IMPALA-6741 - checks that last reporting time exists in |
| profiles of fragment instances.""" |
| query = """select count(distinct a.int_col) from functional.alltypes a |
| inner join functional.alltypessmall b on (a.id = b.id + cast(sleep(15) as INT))""" |
| handle = self.hs2_client.execute_async(query) |
| |
| num_validated = 0 |
| tree = self.hs2_client.get_runtime_profile(handle, TRuntimeProfileFormat.THRIFT) |
| while not self.hs2_client.state_is_finished(handle): |
| assert tree, num_validated |
| for node in tree.nodes: |
| if node.name.startswith('Instance '): |
| info_strings_key = 'Last report received time' |
| assert info_strings_key in node.info_strings |
| report_time_str = node.info_strings[info_strings_key].split(".")[0] |
| # Try converting the string to make sure it's in the expected format |
| assert datetime.strptime(report_time_str, '%Y-%m-%d %H:%M:%S') |
| num_validated += 1 |
| tree = self.hs2_client.get_runtime_profile(handle, TRuntimeProfileFormat.THRIFT) |
| # Let's not hit the backend too hard |
| sleep(0.1) |
| assert num_validated > 0 |
| self.hs2_client.close_query(handle) |
| |
| @SkipIfS3.hbase |
| @SkipIfLocal.hbase |
| @SkipIfIsilon.hbase |
| @SkipIfABFS.hbase |
| @SkipIfADLS.hbase |
| def test_scan_summary(self): |
| """IMPALA-4499: Checks that the exec summary for scans show the table name.""" |
| # HDFS table |
| query = "select count(*) from functional.alltypestiny" |
| result = self.execute_query(query) |
| scan_idx = len(result.exec_summary) - 1 |
| assert result.exec_summary[scan_idx]['operator'] == '00:SCAN HDFS' |
| assert result.exec_summary[scan_idx]['detail'] == 'functional.alltypestiny' |
| |
| # KUDU table |
| query = "select count(*) from functional_kudu.alltypestiny" |
| result = self.execute_query(query) |
| scan_idx = len(result.exec_summary) - 1 |
| assert result.exec_summary[scan_idx]['operator'] == '00:SCAN KUDU' |
| assert result.exec_summary[scan_idx]['detail'] == 'functional_kudu.alltypestiny' |
| |
| # HBASE table |
| query = "select count(*) from functional_hbase.alltypestiny" |
| result = self.execute_query(query) |
| scan_idx = len(result.exec_summary) - 1 |
| assert result.exec_summary[scan_idx]['operator'] == '00:SCAN HBASE' |
| assert result.exec_summary[scan_idx]['detail'] == 'functional_hbase.alltypestiny' |
| |
| def test_sink_summary(self, unique_database): |
| """IMPALA-1048: Checks that the exec summary contains sinks.""" |
| # SELECT query. |
| query = "select count(*) from functional.alltypes" |
| result = self.execute_query(query) |
| # Sanity-check the root sink. |
| root_sink = result.exec_summary[0] |
| assert root_sink['operator'] == 'F01:ROOT' |
| assert root_sink['max_time'] >= 0 |
| assert root_sink['num_rows'] == -1 |
| assert root_sink['est_num_rows'] == -1 |
| assert root_sink['peak_mem'] >= 0 |
| assert root_sink['est_peak_mem'] >= 0 |
| # Sanity-check the exchange sink. |
| found_exchange_sender = False |
| for row in result.exec_summary[1:]: |
| if 'EXCHANGE SENDER' not in row['operator']: |
| continue |
| found_exchange_sender = True |
| assert re.match("F[0-9]+:EXCHANGE SENDER", row['operator']) |
| assert row['max_time'] >= 0 |
| assert row['num_rows'] == -1 |
| assert row['est_num_rows'] == -1 |
| assert row['peak_mem'] >= 0 |
| assert row['est_peak_mem'] >= 0 |
| assert found_exchange_sender, result |
| |
| # INSERT query. |
| query = "create table {0}.tmp as select count(*) from functional.alltypes".format( |
| unique_database) |
| result = self.execute_query(query) |
| # Sanity-check the HDFS writer sink. |
| assert result.exec_summary[0]['operator'] == 'F01:HDFS WRITER' |
| assert result.exec_summary[0]['max_time'] >= 0 |
| assert result.exec_summary[0]['num_rows'] == -1 |
| assert result.exec_summary[0]['est_num_rows'] == -1 |
| assert result.exec_summary[0]['peak_mem'] >= 0 |
| assert result.exec_summary[0]['est_peak_mem'] >= 0 |
| |
| def test_query_options(self): |
| """Test that the query profile shows expected non-default query options, both set |
| explicitly through client and those set by planner""" |
| # Set mem_limit and runtime_filter_wait_time_ms to non-default and default value. |
| query_opts = {'mem_limit': 8589934592, 'runtime_filter_wait_time_ms': 0} |
| profile = self.execute_query("select 1", query_opts).runtime_profile |
| assert "Query Options (set by configuration): MEM_LIMIT=8589934592" in profile,\ |
| profile |
| assert "CLIENT_IDENTIFIER=" + \ |
| "query_test/test_observability.py::TestObservability::()::test_query_options" \ |
| in profile |
| # Get the TIMEZONE value. |
| server_timezone = None |
| for row in self.execute_query("set", query_opts).data: |
| name, val, _ = row.split("\t") |
| if name == "TIMEZONE": |
| server_timezone = val |
| break |
| assert server_timezone is not None |
| |
| # For this query, the planner sets NUM_NODES=1, NUM_SCANNER_THREADS=1, |
| # RUNTIME_FILTER_MODE=0 and MT_DOP=0 |
| expected_str = ("Query Options (set by configuration and planner): " |
| "MEM_LIMIT=8589934592," |
| "NUM_NODES=1,NUM_SCANNER_THREADS=1," |
| "RUNTIME_FILTER_MODE=0,MT_DOP=0,{erasure_coding}TIMEZONE={timezone}," |
| "CLIENT_IDENTIFIER=" |
| "query_test/test_observability.py::TestObservability::()::test_query_options" |
| "\n") |
| expected_str = expected_str.format( |
| erasure_coding="ALLOW_ERASURE_CODED_FILES=1," if IS_EC else "", |
| timezone=server_timezone) |
| assert expected_str in profile, profile |
| |
| def test_exec_summary(self): |
| """Test that the exec summary is populated correctly in every query state""" |
| query = "select count(*) from functional.alltypes" |
| handle = self.execute_query_async(query, |
| {"debug_action": "CRS_BEFORE_ADMISSION:SLEEP@1000"}) |
| # If ExecuteStatement() has completed and the query is paused in the admission control |
| # phase, then the coordinator has not started yet and exec_summary should be empty. |
| exec_summary = self.client.get_exec_summary(handle) |
| assert exec_summary is not None and exec_summary.nodes is None |
| # After completion of the admission control phase, the coordinator would have started |
| # and we should get a populated exec_summary. |
| self.client.wait_for_admission_control(handle) |
| exec_summary = self.client.get_exec_summary(handle) |
| assert exec_summary is not None and exec_summary.nodes is not None |
| |
| self.client.fetch(query, handle) |
| exec_summary = self.client.get_exec_summary(handle) |
| # After fetching the results and reaching finished state, we should still be able to |
| # fetch an exec_summary. |
| assert exec_summary is not None and exec_summary.nodes is not None |
| |
| def test_exec_summary_in_runtime_profile(self): |
| """Test that the exec summary is populated in runtime profile correctly in every |
| query state""" |
| query = "select count(*) from functional.alltypes" |
| handle = self.execute_query_async(query, |
| {"debug_action": "CRS_BEFORE_ADMISSION:SLEEP@1000"}) |
| |
| # If ExecuteStatement() has completed and the query is paused in the admission control |
| # phase, then the coordinator has not started yet and exec_summary should be empty. |
| profile = self.client.get_runtime_profile(handle) |
| assert "ExecSummary:" not in profile, profile |
| # After completion of the admission control phase, the coordinator would have started |
| # and we should get a populated exec_summary. |
| self.client.wait_for_admission_control(handle) |
| profile = self.client.get_runtime_profile(handle) |
| assert "ExecSummary:" in profile, profile |
| |
| self.client.fetch(query, handle) |
| # After fetching the results and reaching finished state, we should still be able to |
| # fetch an exec_summary in profile. |
| profile = self.client.get_runtime_profile(handle) |
| assert "ExecSummary:" in profile, profile |
| |
| @SkipIfLocal.multiple_impalad |
| def test_profile_fragment_instances(self): |
| """IMPALA-6081: Test that the expected number of fragment instances and their exec |
| nodes appear in the runtime profile, even when fragments may be quickly cancelled when |
| all results are already returned.""" |
| results = self.execute_query(""" |
| with l as (select * from tpch.lineitem UNION ALL select * from tpch.lineitem) |
| select STRAIGHT_JOIN count(*) from (select * from tpch.lineitem a LIMIT 1) a |
| join (select * from l LIMIT 2000000) b on a.l_orderkey = -b.l_orderkey;""") |
| # There are 3 scan nodes and each appears in the profile n+1 times (for n fragment |
| # instances + the averaged fragment). n depends on how data is loaded and scheduler's |
| # decision. |
| n = results.runtime_profile.count("HDFS_SCAN_NODE") |
| assert n > 0 and n % 3 == 0 |
| # There are 3 exchange nodes and each appears in the profile 2 times (for 1 fragment |
| # instance + the averaged fragment). |
| assert results.runtime_profile.count("EXCHANGE_NODE") == 6 |
| # The following appear only in the root fragment which has 1 instance. |
| assert results.runtime_profile.count("HASH_JOIN_NODE") == 2 |
| assert results.runtime_profile.count("AGGREGATION_NODE") == 2 |
| assert results.runtime_profile.count("PLAN_ROOT_SINK") == 2 |
| |
| def test_query_profile_contains_query_compilation_static_events(self): |
| """Test that the expected events show up in a query profile. These lines are static |
| and should appear in this exact order.""" |
| event_regexes = [ |
| r'Analysis finished:', |
| r'Authorization finished (.*):', |
| r'Value transfer graph computed:', |
| r'Single node plan created:', |
| r'Runtime filters computed:', |
| r'Distributed plan created:'] |
| query = "select * from functional.alltypes" |
| runtime_profile = self.execute_query(query).runtime_profile |
| self.__verify_profile_event_sequence(event_regexes, runtime_profile) |
| |
| def test_query_profile_contains_query_compilation_metadata_load_events(self, |
| cluster_properties): |
| """Test that the Metadata load started and finished events appear in the query |
| profile when Catalog cache is evicted.""" |
| invalidate_query = "invalidate metadata functional.alltypes" |
| select_query = "select * from functional.alltypes" |
| self.execute_query(invalidate_query).runtime_profile |
| runtime_profile = self.execute_query(select_query).runtime_profile |
| # Depending on whether this is a catalog-v2 cluster or not some of the metadata |
| # loading events are different |
| if not cluster_properties.is_catalog_v2_cluster(): |
| load_event_regexes = [r'Query Compilation:', r'Metadata load started:', |
| r'Metadata load finished. loaded-tables=.*/.* load-requests=.* ' |
| r'catalog-updates=.*:', |
| r'Analysis finished:'] |
| else: |
| load_event_regexes = [ |
| r'Frontend:', |
| r'CatalogFetch.ColumnStats.Hits', |
| r'CatalogFetch.ColumnStats.Misses', |
| r'CatalogFetch.ColumnStats.Requests', |
| r'CatalogFetch.ColumnStats.Time', |
| r'CatalogFetch.Config.Hits', |
| r'CatalogFetch.Config.Misses', |
| r'CatalogFetch.Config.Requests', |
| r'CatalogFetch.Config.Time', |
| r'CatalogFetch.DatabaseList.Hits', |
| r'CatalogFetch.DatabaseList.Misses', |
| r'CatalogFetch.DatabaseList.Requests', |
| r'CatalogFetch.DatabaseList.Time', |
| r'CatalogFetch.PartitionLists.Hits', |
| r'CatalogFetch.PartitionLists.Misses', |
| r'CatalogFetch.PartitionLists.Requests', |
| r'CatalogFetch.PartitionLists.Time', |
| r'CatalogFetch.Partitions.Hits', |
| r'CatalogFetch.Partitions.Misses', |
| r'CatalogFetch.Partitions.Requests', |
| r'CatalogFetch.Partitions.Time', |
| r'CatalogFetch.RPCs.Bytes', |
| r'CatalogFetch.RPCs.Requests', |
| r'CatalogFetch.RPCs.Time', |
| r'CatalogFetch.StorageLoad.Time', |
| r'CatalogFetch.TableNames.Hits', |
| r'CatalogFetch.TableNames.Misses', |
| r'CatalogFetch.TableNames.Requests', |
| r'CatalogFetch.TableNames.Time', |
| r'CatalogFetch.Tables.Hits', |
| r'CatalogFetch.Tables.Misses', |
| r'CatalogFetch.Tables.Requests', |
| r'CatalogFetch.Tables.Time'] |
| self.__verify_profile_event_sequence(load_event_regexes, runtime_profile) |
| |
| def test_query_profile_contains_query_compilation_metadata_cached_event(self): |
| """Test that the Metadata cache available event appears in the query profile when |
| the table is cached.""" |
| refresh_query = "refresh functional.alltypes" |
| select_query = "select * from functional.alltypes" |
| self.execute_query(refresh_query).runtime_profile |
| runtime_profile = self.execute_query(select_query).runtime_profile |
| event_regexes = [r'Query Compilation:', |
| r'Metadata of all .* tables cached:', |
| r'Analysis finished:'] |
| self.__verify_profile_event_sequence(event_regexes, runtime_profile) |
| |
| def test_query_profile_contains_query_compilation_lineage_event(self): |
| """Test that the lineage information appears in the profile in the right place. This |
| event depends on whether the lineage_event_log_dir is configured.""" |
| impalad = self.impalad_test_service |
| lineage_event_log_dir_value = impalad.get_flag_current_value("lineage_event_log_dir") |
| assert lineage_event_log_dir_value is not None |
| if lineage_event_log_dir_value == "": |
| event_regexes = [ |
| r'Distributed plan created:', |
| r'Planning finished:'] |
| else: |
| event_regexes = [ |
| r'Distributed plan created:', |
| r'Lineage info computed:', |
| r'Planning finished:'] |
| query = "select * from functional.alltypes" |
| runtime_profile = self.execute_query(query).runtime_profile |
| self.__verify_profile_event_sequence(event_regexes, runtime_profile) |
| |
| def test_query_profile_contains_query_timeline_events(self): |
| """Test that the expected events show up in a query profile.""" |
| event_regexes = [r'Query Timeline:', |
| r'Query submitted:', |
| r'Planning finished:', |
| r'Submit for admission:', |
| r'Completed admission:', |
| r'Ready to start on .* backends:', |
| r'All .* execution backends \(.* fragment instances\) started:', |
| r'Rows available:', |
| r'First row fetched:', |
| r'Last row fetched:', |
| r'Released admission control resources:'] |
| query = "select * from functional.alltypes" |
| # Use no-limits pool so that it cannot get queued in admission control (which would |
| # add an extra event to the above timeline). |
| query_opts = {'request_pool': 'root.no-limits'} |
| runtime_profile = self.execute_query(query, query_opts).runtime_profile |
| self.__verify_profile_event_sequence(event_regexes, runtime_profile) |
| |
| def test_query_profile_contains_instance_events(self): |
| """Test that /query_profile_encoded contains an event timeline for fragment |
| instances, even when there are errors.""" |
| event_regexes = [r'Fragment Instance Lifecycle Event Timeline', |
| r'Prepare Finished', |
| r'Open Finished', |
| r'First Batch Produced', |
| r'First Batch Sent', |
| r'ExecInternal Finished'] |
| query = "select count(*) from functional.alltypes" |
| runtime_profile = self.execute_query(query).runtime_profile |
| self.__verify_profile_event_sequence(event_regexes, runtime_profile) |
| |
| def test_query_profile_contains_node_events(self): |
| """Test that ExecNode events show up in a profile.""" |
| event_regexes = [r'Node Lifecycle Event Timeline', |
| r'Open Started', |
| r'Open Finished', |
| r'First Batch Requested', |
| r'First Batch Returned', |
| r'Last Batch Returned', |
| r'Closed'] |
| query = "select count(*) from functional.alltypes" |
| runtime_profile = self.execute_query(query).runtime_profile |
| self.__verify_profile_event_sequence(event_regexes, runtime_profile) |
| |
| def __verify_profile_event_sequence(self, event_regexes, runtime_profile): |
| """Check that 'event_regexes' appear in a consecutive series of lines in |
| 'runtime_profile'""" |
| lines = runtime_profile.splitlines() |
| event_regex_index = 0 |
| |
| # Check that the strings appear in the above order with no gaps in the profile. |
| for line in runtime_profile.splitlines(): |
| match = re.search(event_regexes[event_regex_index], line) |
| if match is not None: |
| event_regex_index += 1 |
| if event_regex_index == len(event_regexes): |
| # Found all the lines - we're done. |
| return |
| else: |
| # Haven't found the first regex yet. |
| assert event_regex_index == 0, \ |
| event_regexes[event_regex_index] + " not in " + line + "\n" + runtime_profile |
| assert event_regex_index == len(event_regexes), \ |
| "Didn't find all events in profile: \n" + runtime_profile |
| |
| def test_query_profile_contains_all_events(self, unique_database): |
| """Test that the expected events show up in a query profile for various queries""" |
| # make a data file to load data from |
| path = "test-warehouse/{0}.db/data_file".format(unique_database) |
| self.filesystem_client.create_file(path, "1") |
| use_query = "use {0}".format(unique_database) |
| self.execute_query(use_query) |
| # all the events we will see for every query |
| event_regexes = [ |
| r'Query Compilation:', |
| r'Query Timeline:', |
| r'Planning finished' |
| ] |
| # queries that explore different code paths in Frontend compilation |
| queries = [ |
| 'create table if not exists impala_6568 (i int)', |
| 'select * from impala_6568', |
| 'explain select * from impala_6568', |
| 'describe impala_6568', |
| 'alter table impala_6568 set tblproperties(\'numRows\'=\'10\')', |
| "load data inpath '/{0}' into table impala_6568".format(path) |
| ] |
| # run each query... |
| for query in queries: |
| runtime_profile = self.execute_query(query).runtime_profile |
| # and check that all the expected events appear in the resulting profile |
| self.__verify_profile_contains_every_event(event_regexes, runtime_profile, query) |
| |
| def __verify_profile_contains_every_event(self, event_regexes, runtime_profile, query): |
| """Test that all the expected events show up in a given query profile.""" |
| for regex in event_regexes: |
| assert any(re.search(regex, line) for line in runtime_profile.splitlines()), \ |
| "Didn't find event '" + regex + "' for query '" + query + \ |
| "' in profile: \n" + runtime_profile |
| |
| def test_compute_stats_profile(self, unique_database): |
| """Test that the profile for a 'compute stats' query contains three unique query ids: |
| one for the parent 'compute stats' query and one each for the two child queries.""" |
| table_name = "%s.test_compute_stats_profile" % unique_database |
| self.execute_query( |
| "create table %s as select * from functional.alltypestiny" % table_name) |
| results = self.execute_query("compute stats %s" % table_name) |
| # Search for all query ids (max length 33) in the profile. |
| matches = re.findall("Query \(id=.{,33}\)", results.runtime_profile) |
| query_ids = [] |
| for query_id in matches: |
| if query_id not in query_ids: |
| query_ids.append(query_id) |
| assert len(query_ids) == 3, results.runtime_profile |
| |
| def test_global_resource_counters_in_profile(self): |
| """Test that a set of global resource usage counters show up in the profile.""" |
| query = "select count(*) from functional.alltypes" |
| profile = self.execute_query(query).runtime_profile |
| expected_counters = ["TotalBytesRead", "TotalBytesSent", "TotalScanBytesSent", |
| "TotalInnerBytesSent", "ExchangeScanRatio", |
| "InnerNodeSelectivityRatio"] |
| assert all(counter in profile for counter in expected_counters) |
| |
| @SkipIfNotHdfsMinicluster.tuned_for_minicluster |
| def test_global_exchange_counters(self): |
| """Test that global exchange counters are set correctly.""" |
| query = """select count(*) from tpch_parquet.orders o inner join tpch_parquet.lineitem |
| l on o.o_orderkey = l.l_orderkey group by o.o_clerk limit 10""" |
| profile = self.execute_query(query).runtime_profile |
| |
| # TimeSeriesCounter should be prefixed with a hyphen. |
| assert " MemoryUsage" not in profile |
| assert "- MemoryUsage" in profile |
| |
| assert "ExchangeScanRatio: 3.19" in profile |
| |
| keys = ["TotalBytesSent", "TotalScanBytesSent", "TotalInnerBytesSent"] |
| counters = defaultdict(int) |
| for line in profile.splitlines(): |
| for key in keys: |
| if key in line: |
| # Match byte count within parentheses |
| m = re.search("\(([0-9]+)\)", line) |
| assert m, "Cannot match pattern for key %s in line '%s'" % (key, line) |
| # Only keep first (query-level) counter |
| if counters[key] == 0: |
| counters[key] = int(m.group(1)) |
| |
| # All counters have values |
| assert all(counters[key] > 0 for key in keys) |
| |
| assert counters["TotalBytesSent"] == (counters["TotalScanBytesSent"] + |
| counters["TotalInnerBytesSent"]) |
| |
| def test_query_profile_contains_host_resource_usage(self): |
| """Tests that the profile contains a sub-profile with per node resource usage.""" |
| result = self.execute_query("select count(*), sleep(1000) from functional.alltypes") |
| profile = result.runtime_profile |
| expected_str = "Per Node Profiles:" |
| assert any(expected_str in line for line in profile.splitlines()) |
| |
| def test_query_profile_host_resource_metrics_off(self): |
| """Tests that the query profile does not contain resource usage metrics by default or |
| when disabled explicitly.""" |
| query = "select count(*), sleep(1000) from functional.alltypes" |
| for query_opts in [None, {'resource_trace_ratio': 0.0}]: |
| profile = self.execute_query(query, query_opts).runtime_profile |
| # Assert that no host resource counters exist in the profile |
| for line in profile.splitlines(): |
| assert not re.search("HostCpu.*Percentage", line) |
| assert not re.search("HostNetworkRx", line) |
| assert not re.search("HostDiskReadThroughput", line) |
| |
| def test_query_profile_contains_host_resource_metrics(self): |
| """Tests that the query profile contains various CPU and network metrics.""" |
| query_opts = {'resource_trace_ratio': 1.0} |
| query = "select count(*), sleep(1000) from functional.alltypes" |
| profile = self.execute_query(query, query_opts).runtime_profile |
| # We check for 500ms because a query with 1s duration won't hit the 64 values limit |
| # that would trigger resampling. |
| expected_strs = ["HostCpuIoWaitPercentage (500.000ms):", |
| "HostCpuSysPercentage (500.000ms):", |
| "HostCpuUserPercentage (500.000ms):", |
| "HostNetworkRx (500.000ms):", |
| "HostNetworkTx (500.000ms):", |
| "HostDiskReadThroughput (500.000ms):", |
| "HostDiskWriteThroughput (500.000ms):"] |
| |
| # Assert that all expected counters exist in the profile. |
| for expected_str in expected_strs: |
| assert any(expected_str in line for line in profile.splitlines()), expected_str |
| |
| # Check that there are some values for each counter. |
| for line in profile.splitlines(): |
| if not any(key in line for key in expected_strs): |
| continue |
| values = line.split(':')[1].strip().split(',') |
| assert len(values) > 0 |
| |
| def _find_ts_counters_in_thrift_profile(self, profile, name): |
| """Finds all time series counters in 'profile' with a matching name.""" |
| counters = [] |
| for node in profile.nodes: |
| for counter in node.time_series_counters or []: |
| if counter.name == name: |
| counters.append(counter) |
| return counters |
| |
| @pytest.mark.execute_serially |
| def test_thrift_profile_contains_host_resource_metrics(self): |
| """Tests that the thrift profile contains time series counters for CPU and network |
| resource usage.""" |
| query_opts = {'resource_trace_ratio': 1.0} |
| self.hs2_client.set_configuration(query_opts) |
| result = self.hs2_client.execute("select sleep(2000)", |
| profile_format=TRuntimeProfileFormat.THRIFT) |
| thrift_profile = result.profile |
| |
| expected_keys = ["HostCpuUserPercentage", "HostNetworkRx", "HostDiskReadThroughput"] |
| for key in expected_keys: |
| counters = self._find_ts_counters_in_thrift_profile(thrift_profile, key) |
| # The query will run on a single node, we will only find the counter once. |
| assert len(counters) == 1 |
| counter = counters[0] |
| assert len(counter.values) > 0 |
| |
| @pytest.mark.execute_serially |
| def test_query_profile_thrift_timestamps(self): |
| """Test that the query profile start and end time date-time strings have |
| nanosecond precision. Nanosecond precision is expected by management API clients |
| that consume Impala debug webpages.""" |
| query = "select sleep(5)" |
| result = self.hs2_client.execute(query, profile_format=TRuntimeProfileFormat.THRIFT) |
| tree = result.profile |
| |
| # tree.nodes[1] corresponds to ClientRequestState::summary_profile_ |
| # See be/src/service/client-request-state.[h|cc]. |
| start_time = tree.nodes[1].info_strings["Start Time"] |
| end_time = tree.nodes[1].info_strings["End Time"] |
| # Start and End Times are of the form "2017-12-07 22:26:52.167711000" |
| start_time_sub_sec_str = start_time.split('.')[-1] |
| end_time_sub_sec_str = end_time.split('.')[-1] |
| |
| assert len(end_time_sub_sec_str) == 9, end_time |
| assert len(start_time_sub_sec_str) == 9, start_time |
| |
| @pytest.mark.execute_serially |
| def test_end_time(self): |
| """ Test that verifies that the end time of a query with a coordinator is set once |
| the coordinator releases its admission control resources. This ensures that the |
| duration of the query will be determined by the time taken to do real work rather |
| than the duration for which the query remains open. On the other hand, for queries |
| without coordinators, the End Time is set only when UnregisterQuery() is called.""" |
| # Test the end time of a query with a coordinator. |
| query = "select 1" |
| handle = self.hs2_client.execute_async(query) |
| result = self.hs2_client.fetch(query, handle) |
| # Ensure that the query returns a non-empty result set. |
| assert result is not None |
| # Once the results have been fetched, the query End Time must be set. |
| tree = self.hs2_client.get_runtime_profile(handle, TRuntimeProfileFormat.THRIFT) |
| end_time = tree.nodes[1].info_strings["End Time"] |
| assert end_time |
| self.hs2_client.close_query(handle) |
| |
| # Test the end time of a query without a coordinator. |
| query = "describe functional.alltypes" |
| handle = self.hs2_client.execute_async(query) |
| result = self.hs2_client.fetch(query, handle) |
| # Ensure that the query returns a non-empty result set. |
| assert result |
| # The query End Time must not be set until the query is unregisterted |
| tree = self.hs2_client.get_runtime_profile(handle, TRuntimeProfileFormat.THRIFT) |
| end_time = tree.nodes[1].info_strings["End Time"] |
| assert len(end_time) == 0, end_time |
| # Save the last operation to be able to retrieve the profile after closing the query |
| last_op = handle.get_handle()._last_operation |
| self.hs2_client.close_query(handle) |
| tree = last_op.get_profile(TRuntimeProfileFormat.THRIFT) |
| end_time = tree.nodes[1].info_strings["End Time"] |
| assert end_time is not None |
| |
| def test_query_profile_contains_number_of_fragment_instance(self): |
| """Test that the expected section for number of fragment instance in |
| a query profile.""" |
| event_regexes = [r'Per Host Number of Fragment Instances'] |
| query = "select count (*) from functional.alltypes" |
| runtime_profile = self.execute_query(query).runtime_profile |
| self.__verify_profile_event_sequence(event_regexes, runtime_profile) |
| |
| def test_query_profile_contains_executor_group(self): |
| """Test that the profile contains an info string with the executor group that was |
| picked by admission control.""" |
| query = "select count (*) from functional.alltypes" |
| runtime_profile = self.execute_query(query).runtime_profile |
| assert "Executor Group:" in runtime_profile |
| |
| def test_query_profile_storage_load_time_filesystem(self, unique_database, |
| cluster_properties): |
| """Test that when a query needs load metadata for table(s), the |
| storage load time should be in the profile. Tests file systems.""" |
| table_name = 'ld_prof' |
| self.execute_query( |
| "create table {0}.{1}(col1 int)".format(unique_database, table_name)) |
| self.__check_query_profile_storage_load_time(unique_database, table_name, |
| cluster_properties) |
| |
| @SkipIfS3.hbase |
| @SkipIfLocal.hbase |
| @SkipIfIsilon.hbase |
| @SkipIfABFS.hbase |
| @SkipIfADLS.hbase |
| @pytest.mark.execute_serially |
| def test_query_profile_storage_load_time(self, cluster_properties): |
| """Test that when a query needs load metadata for table(s), the |
| storage load time should be in the profile. Tests kudu and hbase.""" |
| # KUDU table |
| self.__check_query_profile_storage_load_time("functional_kudu", "alltypes", |
| cluster_properties) |
| |
| # HBASE table |
| self.__check_query_profile_storage_load_time("functional_hbase", "alltypes", |
| cluster_properties) |
| |
| def __check_query_profile_storage_load_time(self, db_name, table_name, |
| cluster_properties): |
| """Check query profile for storage load time with a given database.""" |
| self.execute_query("invalidate metadata {0}.{1}".format(db_name, table_name)) |
| query = "select count (*) from {0}.{1}".format(db_name, table_name) |
| runtime_profile = self.execute_query(query).runtime_profile |
| if cluster_properties.is_catalog_v2_cluster(): |
| storageLoadTime = "StorageLoad.Time" |
| else: |
| storageLoadTime = "storage-load-time" |
| assert storageLoadTime in runtime_profile |
| # Call the second time, no metastore loading needed. |
| # Only check this part in Catalog V1 because of V2's random behavior |
| if not cluster_properties.is_catalog_v2_cluster(): |
| runtime_profile = self.execute_query(query).runtime_profile |
| assert storageLoadTime not in runtime_profile |
| |
| def __verify_hashtable_stats_profile(self, runtime_profile): |
| assert "Hash Table" in runtime_profile |
| assert "Probes:" in runtime_profile |
| assert "Travel:" in runtime_profile |
| assert "HashCollisions:" in runtime_profile |
| assert "Resizes:" in runtime_profile |
| nprobes = re.search('Probes:.*\((\d+)\)', runtime_profile) |
| # Probes and travel can be 0. The number can be an integer or float with K. |
| assert nprobes and len(nprobes.groups()) == 1 and nprobes.group(1) >= 0 |
| ntravel = re.search('Travel:.*\((\d+)\)', runtime_profile) |
| assert ntravel and len(ntravel.groups()) == 1 and ntravel.group(1) >= 0 |
| |
| def test_query_profle_hashtable(self): |
| """Test that the profile for join/aggregate contains hash table related |
| information.""" |
| # Join |
| query = """select a.int_col, a.string_col from functional.alltypes a |
| inner join functional.alltypessmall b on a.id = b.id""" |
| result = self.execute_query(query) |
| assert result.success |
| self.__verify_hashtable_stats_profile(result.runtime_profile) |
| # Group by |
| query = """select year, count(*) from |
| functional.alltypesagg where int_col < 7 and year = 2010 group by year""" |
| result = self.execute_query(query) |
| assert result.success |
| self.__verify_hashtable_stats_profile(result.runtime_profile) |
| |
| |
| class TestQueryStates(ImpalaTestSuite): |
| """Test that the 'Query State' and 'Impala Query State' are set correctly in the |
| runtime profile.""" |
| |
| @classmethod |
| def get_workload(self): |
| return 'functional-query' |
| |
| def test_query_states(self): |
| """Tests that the query profile shows expected query states.""" |
| query = "select count(*) from functional.alltypes where bool_col = sleep(10)" |
| handle = self.execute_query_async(query, |
| {"debug_action": "CRS_BEFORE_ADMISSION:SLEEP@1000"}) |
| # If ExecuteStatement() has completed and the query is paused in the admission control |
| # phase, then the query must be in COMPILED state. |
| profile = self.client.get_runtime_profile(handle) |
| assert self.__is_line_in_profile("Query State: COMPILED", profile) |
| assert self.__is_line_in_profile("Impala Query State: PENDING", profile) |
| # After completion of the admission control phase, the query must have at least |
| # reached RUNNING state. |
| self.client.wait_for_admission_control(handle) |
| profile = self.client.get_runtime_profile(handle) |
| assert self.__is_line_in_profile("Query State: RUNNING", profile), profile |
| assert self.__is_line_in_profile("Impala Query State: RUNNING", profile), profile |
| |
| self.client.fetch(query, handle) |
| profile = self.client.get_runtime_profile(handle) |
| # After fetching the results, the query must be in state FINISHED. |
| assert self.__is_line_in_profile("Query State: FINISHED", profile), profile |
| assert self.__is_line_in_profile("Impala Query State: FINISHED", profile), profile |
| |
| def test_error_query_state(self): |
| """Tests that the query profile shows the proper error state.""" |
| query = "select * from functional.alltypes limit 10" |
| handle = self.execute_query_async(query, {"abort_on_error": "1", |
| "debug_action": "0:GETNEXT:FAIL"}) |
| |
| def assert_finished(): |
| profile = self.client.get_runtime_profile(handle) |
| return self.__is_line_in_profile("Query State: FINISHED", profile) and \ |
| self.__is_line_in_profile("Impala Query State: FINISHED", profile) |
| |
| self.assert_eventually(30, 1, assert_finished, |
| lambda: self.client.get_runtime_profile(handle)) |
| |
| try: |
| self.client.fetch(query, handle) |
| assert False |
| except ImpalaBeeswaxException: |
| pass |
| |
| profile = self.client.get_runtime_profile(handle) |
| assert self.__is_line_in_profile("Query State: EXCEPTION", profile), profile |
| assert self.__is_line_in_profile("Impala Query State: ERROR", profile), profile |
| |
| def __is_line_in_profile(self, line, profile): |
| """Returns true if the given 'line' is in the given 'profile'. A single line of the |
| profile must exactly match the given 'line' (excluding whitespaces).""" |
| return re.search("^\s*{0}\s*$".format(line), profile, re.M) |
