Execute the query using the cursor and set the result or exception on the local thread.
(self, query, cursor, sql_writer, log_file)
| 416 | query_thread.cursor_description) for query_thread in query_threads] |
| 417 | |
| 418 | def _fetch_sql_results(self, query, cursor, sql_writer, log_file): |
| 419 | '''Execute the query using the cursor and set the result or exception on the local |
| 420 | thread. |
| 421 | ''' |
| 422 | try: |
| 423 | log_file.write('/***** Start Query *****/\n') |
| 424 | if sql_writer.DIALECT == self.flatten_dialect: |
| 425 | # Converts the query model for the flattened version of the data. This is for |
| 426 | # testing of Impala nested types support. |
| 427 | query = deepcopy(query) |
| 428 | QueryFlattener().flatten(query) |
| 429 | if query.execution == StatementExecutionMode.CREATE_TABLE_AS: |
| 430 | setup_sql = sql_writer.write_create_table_as(query, self._table_or_view_name) |
| 431 | query_sql = 'SELECT * FROM ' + self._table_or_view_name |
| 432 | elif query.execution == StatementExecutionMode.CREATE_VIEW_AS: |
| 433 | setup_sql = sql_writer.write_create_view(query, self._table_or_view_name) |
| 434 | query_sql = 'SELECT * FROM ' + self._table_or_view_name |
| 435 | elif isinstance(query, (InsertStatement,)): |
| 436 | setup_sql = sql_writer.write_query(query) |
| 437 | # TODO: improve validation (IMPALA-4599). This is good enough for looking for |
| 438 | # crashes on DML statements |
| 439 | query_sql = 'SELECT COUNT(*) FROM ' + self._table_or_view_name |
| 440 | else: |
| 441 | setup_sql = None |
| 442 | query_sql = sql_writer.write_query(query) |
| 443 | if setup_sql: |
| 444 | LOG.debug("Executing on %s:\n%s", cursor.db_type, setup_sql) |
| 445 | current_thread().sql = setup_sql + ';\n' |
| 446 | log_file.write(setup_sql + ';\n') |
| 447 | log_file.flush() |
| 448 | cursor.execute(setup_sql) |
| 449 | LOG.debug("Executing on %s:\n%s", cursor.db_type, query_sql) |
| 450 | current_thread().sql += query_sql |
| 451 | log_file.write(query_sql + ';\n') |
| 452 | log_file.write('/***** End Query *****/\n') |
| 453 | log_file.flush() |
| 454 | cursor.execute(query_sql) |
| 455 | col_count = len(cursor.description) |
| 456 | batch_size = max(10000 // col_count, 1) |
| 457 | row_limit = self.TOO_MUCH_DATA // col_count |
| 458 | data_set = list() |
| 459 | current_thread().data_set = data_set |
| 460 | current_thread().cursor_description = cursor.description |
| 461 | LOG.debug("Fetching results from %s", cursor.db_type) |
| 462 | while True: |
| 463 | batch = cursor.fetchmany(batch_size) |
| 464 | data_set.extend(batch) |
| 465 | if len(batch) < batch_size: |
| 466 | if cursor.db_type == IMPALA: |
| 467 | impala_log = cursor.get_log() |
| 468 | if 'Expression overflowed, returning NULL' in impala_log: |
| 469 | raise TypeOverflow('Numeric overflow; data may not match') |
| 470 | break |
| 471 | if len(data_set) > row_limit: |
| 472 | raise DataLimitExceeded('Too much data') |
| 473 | if isinstance(query, (InsertStatement,)): |
| 474 | LOG.debug('Total row count for {0}: {1}'.format( |
| 475 | cursor.db_type, str(data_set))) |
nothing calls this directly
no test coverage detected