Runs the given query by itself repeatedly until the minimum memory is determined with and without spilling. Potentially all fields in the Query class (except 'sql') will be populated by this method. 'required_mem_mb_without_spilling' and the corresponding runtime field may still be None if the
(query, impala, converted_args, timeout_secs=maxsize)
| 854 | |
| 855 | |
| 856 | def populate_runtime_info(query, impala, converted_args, timeout_secs=maxsize): |
| 857 | """Runs the given query by itself repeatedly until the minimum memory is determined |
| 858 | with and without spilling. Potentially all fields in the Query class (except |
| 859 | 'sql') will be populated by this method. 'required_mem_mb_without_spilling' and |
| 860 | the corresponding runtime field may still be None if the query could not be run |
| 861 | without spilling. |
| 862 | |
| 863 | converted_args.samples and converted_args.max_conflicting_samples control the |
| 864 | reliability of the collected information. The problem is that memory spilling or usage |
| 865 | may differ (by a large amount) from run to run due to races during execution. The |
| 866 | parameters provide a way to express "X out of Y runs must have resulted in the same |
| 867 | outcome". Increasing the number of samples and decreasing the tolerance (max conflicts) |
| 868 | increases confidence but also increases the time to collect the data. |
| 869 | """ |
| 870 | LOG.info("Collecting runtime info for query %s: \n%s", query.name, query.sql) |
| 871 | samples = converted_args.samples |
| 872 | max_conflicting_samples = converted_args.max_conflicting_samples |
| 873 | results_dir = converted_args.results_dir |
| 874 | mem_limit_eq_threshold_mb = converted_args.mem_limit_eq_threshold_mb |
| 875 | mem_limit_eq_threshold_percent = converted_args.mem_limit_eq_threshold_percent |
| 876 | runner = QueryRunner(impalad=impala.impalads[0], results_dir=results_dir, |
| 877 | common_query_options=converted_args.common_query_options, |
| 878 | test_admission_control=converted_args.test_admission_control, |
| 879 | use_kerberos=converted_args.use_kerberos, check_if_mem_was_spilled=True) |
| 880 | runner.connect() |
| 881 | limit_exceeded_mem = 0 |
| 882 | non_spill_mem = None |
| 883 | spill_mem = None |
| 884 | |
| 885 | report = None |
| 886 | mem_limit = None |
| 887 | |
| 888 | old_required_mem_mb_without_spilling = query.required_mem_mb_without_spilling |
| 889 | old_required_mem_mb_with_spilling = query.required_mem_mb_with_spilling |
| 890 | |
| 891 | profile_error_prefix = query.logical_query_id + "_binsearch_error" |
| 892 | |
| 893 | # TODO: This method is complicated enough now that breaking it out into a class may be |
| 894 | # helpful to understand the structure. |
| 895 | |
| 896 | def update_runtime_info(): |
| 897 | required_mem = min(mem_limit, impala.min_impalad_mem_mb) |
| 898 | if report.mem_was_spilled: |
| 899 | if ( |
| 900 | query.required_mem_mb_with_spilling is None or |
| 901 | required_mem < query.required_mem_mb_with_spilling |
| 902 | ): |
| 903 | query.required_mem_mb_with_spilling = required_mem |
| 904 | query.solo_runtime_secs_with_spilling = report.runtime_secs |
| 905 | query.solo_runtime_profile_with_spilling = report.profile |
| 906 | elif ( |
| 907 | query.required_mem_mb_without_spilling is None or |
| 908 | required_mem < query.required_mem_mb_without_spilling |
| 909 | ): |
| 910 | query.required_mem_mb_without_spilling = required_mem |
| 911 | query.solo_runtime_secs_without_spilling = report.runtime_secs |
| 912 | assert report.runtime_secs is not None, report |
| 913 | query.solo_runtime_profile_without_spilling = report.profile |
no test coverage detected