Validate that ClientFetchWaitTimer, NumRowsFetched, RowMaterializationRate, and RowMaterializationTimer are set to valid values in the ImpalaServer section of the runtime profile.
(self)
| 32 | """Tests that are independent of whether result spooling is enabled or not.""" |
| 33 | |
| 34 | def test_rows_sent_counters(self): |
| 35 | """Validate that ClientFetchWaitTimer, NumRowsFetched, RowMaterializationRate, |
| 36 | and RowMaterializationTimer are set to valid values in the ImpalaServer section |
| 37 | of the runtime profile.""" |
| 38 | num_rows = 25 |
| 39 | query = "select sleep(100) from functional.alltypes limit {0}".format(num_rows) |
| 40 | client = self.hs2_client |
| 41 | handle = client.execute_async(query) |
| 42 | try: |
| 43 | # Wait until the query is 'FINISHED' and results are available for fetching. |
| 44 | state, num_state_checks = client.wait_for_impala_state(handle, FINISHED, 30) |
| 45 | assert state == FINISHED |
| 46 | # Sleep for 2.5 seconds so that the ClientFetchWaitTimer is >= 1s. |
| 47 | sleep(2.5) |
| 48 | # Fetch the results so that the fetch related counters are updated. |
| 49 | res = client.fetch(query, handle) |
| 50 | assert res.success |
| 51 | |
| 52 | runtime_profile = res.runtime_profile |
| 53 | fetch_timer = re.search("ClientFetchWaitTimer: (.*)", runtime_profile) |
| 54 | assert fetch_timer and len(fetch_timer.groups()) == 1 and \ |
| 55 | parse_duration_string_ms(fetch_timer.group(1)) > 1000 |
| 56 | assert "NumRowsFetched: {0} ({0})".format(num_rows) in runtime_profile |
| 57 | assert re.search("RowMaterializationRate: [1-9]", runtime_profile) |
| 58 | # The query should take at least 1s to materialize all rows since it should sleep |
| 59 | # for at least 1s during materialization. |
| 60 | materialization_timer = re.search("RowMaterializationTimer: (.*)", runtime_profile) |
| 61 | assert materialization_timer and len(materialization_timer.groups()) == 1 and \ |
| 62 | parse_duration_string_ms(materialization_timer.group(1)) > 1000 |
| 63 | rpc_count = int(re.search("RPCCount: ([0-9]+)", runtime_profile).group(1)) |
| 64 | # Apart from the GetOperationStatus RPCs for state checking, there are 5 additional |
| 65 | # RPCs: GetResultSetMetadata, FetchResults * 2, GetLog, GetRuntimeProfile. |
| 66 | assert rpc_count == num_state_checks + 5 |
| 67 | |
| 68 | rpc_read_timer = re.search("RPCReadTimer: (.*)", runtime_profile) |
| 69 | assert rpc_read_timer and len(rpc_read_timer.groups()) == 1 |
| 70 | rpc_read_ns = parse_duration_string_ns(rpc_read_timer.group(1)) |
| 71 | assert 0 < rpc_read_ns and rpc_read_ns < 1000000 |
| 72 | |
| 73 | rpc_write_timer = re.search("RPCWriteTimer: (.*)", runtime_profile) |
| 74 | assert rpc_write_timer and len(rpc_write_timer.groups()) == 1 |
| 75 | rpc_write_ns = parse_duration_string_ns(rpc_write_timer.group(1)) |
| 76 | assert 0 < rpc_write_ns and rpc_write_ns < 10000000 |
| 77 | |
| 78 | create_result_time = re.search("CreateResultSetTime: (.*)", runtime_profile) |
| 79 | assert create_result_time and len(create_result_time.groups()) == 1 |
| 80 | create_result_ms = parse_duration_string_ms(create_result_time.group(1)) |
| 81 | assert 2400 < create_result_ms and create_result_ms < 2600 |
| 82 | |
| 83 | finally: |
| 84 | client.close_query(handle) |
| 85 | |
| 86 | def test_client_fetch_time_stats(self): |
| 87 | num_rows = 27 |
nothing calls this directly
no test coverage detected