Concurrently execute the query using each cursor and return a list of tuples containing the result information for each cursor. The tuple format is ( , ). If query_timeout_seconds is reached and the connection is killable then the quer
(self, query)
| 349 | cursor.execute(opt) |
| 350 | |
| 351 | def fetch_query_results(self, query): |
| 352 | '''Concurrently execute the query using each cursor and return a list of tuples |
| 353 | containing the result information for each cursor. The tuple format is |
| 354 | (<exception or None>, <data set or None>). |
| 355 | |
| 356 | If query_timeout_seconds is reached and the connection is killable then the |
| 357 | query will be cancelled and the connection reset. Otherwise the query will |
| 358 | continue to run in the background. |
| 359 | |
| 360 | "query" should be an instance of query.Query. |
| 361 | ''' |
| 362 | if query.execution in (StatementExecutionMode.CREATE_TABLE_AS, |
| 363 | StatementExecutionMode.CREATE_VIEW_AS): |
| 364 | self._table_or_view_name = self._create_random_table_name() |
| 365 | elif isinstance(query, (InsertStatement,)): |
| 366 | self._table_or_view_name = query.dml_table.name |
| 367 | |
| 368 | query_threads = list() |
| 369 | for sql_writer, cursor, log_file in zip( |
| 370 | self.sql_writers, self.cursors, self.query_logs |
| 371 | ): |
| 372 | if self.ENABLE_RANDOM_QUERY_OPTIONS and cursor.db_type == IMPALA: |
| 373 | self.set_impala_query_options(cursor) |
| 374 | query_thread = Thread( |
| 375 | target=self._fetch_sql_results, |
| 376 | args=[query, cursor, sql_writer, log_file], |
| 377 | name='{db_type}-exec-{id_}'.format( |
| 378 | db_type=cursor.db_type, id_=id(query))) |
| 379 | query_thread.daemon = True |
| 380 | query_thread.sql = '' |
| 381 | query_thread.data_set = None |
| 382 | query_thread.cursor_description = None |
| 383 | query_thread.exception = None |
| 384 | query_thread.start() |
| 385 | query_threads.append(query_thread) |
| 386 | |
| 387 | end_time = time() + self.query_timeout_seconds |
| 388 | for query_thread, cursor in zip(query_threads, self.cursors): |
| 389 | join_time = end_time - time() |
| 390 | if join_time > 0: |
| 391 | query_thread.join(join_time) |
| 392 | if query_thread.is_alive(): |
| 393 | # Kill connection and reconnect to return cursor to initial state. |
| 394 | if cursor.conn.supports_kill: |
| 395 | LOG.debug('Attempting to kill connection') |
| 396 | cursor.conn.kill() |
| 397 | LOG.debug('Kill connection') |
| 398 | try: |
| 399 | # TODO: Sometimes this takes a very long time causing the program to appear to |
| 400 | # hang. Maybe this should be done in another thread so a timeout can be |
| 401 | # applied? |
| 402 | cursor.close() |
| 403 | except Exception as e: |
| 404 | LOG.info('Error closing cursor: %s', e) |
| 405 | cursor.reconnect() |
| 406 | query_thread.exception = QueryTimeout( |
| 407 | 'Query timed out after %s seconds' % self.query_timeout_seconds) |
| 408 |
no test coverage detected