Run a query and return an execution report. If 'run_set_up' is True, set up sql will be executed before the main query. This should be the case during the binary search phase of the stress test. 'cancel_mech' is optionally a CancelMechanism value that should be used to cancel the query a
(self, query, mem_limit_mb, run_set_up=False,
timeout_secs=maxsize, cancel_mech=None, retain_profile=False)
| 100 | self.impalad_conn = self.impalad.impala.connect(impalad=self.impalad) |
| 101 | |
| 102 | def run_query(self, query, mem_limit_mb, run_set_up=False, |
| 103 | timeout_secs=maxsize, cancel_mech=None, retain_profile=False): |
| 104 | """Run a query and return an execution report. If 'run_set_up' is True, set up sql |
| 105 | will be executed before the main query. This should be the case during the binary |
| 106 | search phase of the stress test. 'cancel_mech' is optionally a CancelMechanism |
| 107 | value that should be used to cancel the query after timeout_secs. |
| 108 | If 'cancel_mech' is provided, don't get the query profile for timed out queries |
| 109 | because the query was purposely cancelled, rather than having some problem that needs |
| 110 | investigation. |
| 111 | """ |
| 112 | assert self.impalad_conn, "connect() must be called before run_query()" |
| 113 | assert cancel_mech is None or cancel_mech in CancelMechanism.ALL_MECHS |
| 114 | |
| 115 | timeout_unix_time = time() + timeout_secs |
| 116 | report = QueryReport(query) |
| 117 | try: |
| 118 | with self.impalad_conn.cursor() as cursor: |
| 119 | start_time = time() |
| 120 | self._set_db_and_options(cursor, query, run_set_up, mem_limit_mb, timeout_secs, |
| 121 | cancel_mech) |
| 122 | error = None |
| 123 | try: |
| 124 | cursor.execute_async( |
| 125 | "/* Mem: %s MB. Coordinator: %s. */\n" |
| 126 | % (mem_limit_mb, self.impalad.host_name) + query.sql) |
| 127 | report.query_id = op_handle_to_query_id(cursor._last_operation.handle if |
| 128 | cursor._last_operation else None) |
| 129 | LOG.debug("Query id is %s", report.query_id) |
| 130 | self._wait_until_fetchable(cursor, report, timeout_unix_time) |
| 131 | if query.query_type == QueryType.SELECT: |
| 132 | report.result_hash = self._fetch_and_hash_result(cursor, timeout_unix_time, |
| 133 | query) |
| 134 | else: |
| 135 | # If query is in error state, this will raise an exception |
| 136 | cursor._wait_to_finish() |
| 137 | if (retain_profile or |
| 138 | query.result_hash and report.result_hash != query.result_hash): |
| 139 | fetch_and_set_profile(cursor, report) |
| 140 | except QueryTimeout: |
| 141 | if not cancel_mech: |
| 142 | fetch_and_set_profile(cursor, report) |
| 143 | # Cancel from this client if a) we hit the timeout unexpectedly or b) we're |
| 144 | # deliberately cancelling the query via the client. |
| 145 | if not cancel_mech or cancel_mech == CancelMechanism.VIA_CLIENT: |
| 146 | self._cancel(cursor, report) |
| 147 | else: |
| 148 | # Wait until the query is cancelled by a different mechanism. |
| 149 | self._wait_until_cancelled(cursor, report.query_id) |
| 150 | report.timed_out = True |
| 151 | return report |
| 152 | except Exception as error: |
| 153 | report.query_id = op_handle_to_query_id(cursor._last_operation.handle if |
| 154 | cursor._last_operation else None) |
| 155 | LOG.debug("Error running query with id %s: %s", report.query_id, error) |
| 156 | if (cancel_mech == CancelMechanism.VIA_OPTION and |
| 157 | is_exec_time_limit_error(error)): |
| 158 | # Timeout via query option counts as a timeout. |
| 159 | report.timed_out = True |
no test coverage detected