Tests FETCH_ROWS_TIMEOUT_MS by running a query that produces RowBatches with a large delay. The test waits for the query to 'finish' and then fetches the first RowBatch, which should always be available since a query is only considered 'finished' if rows are available. Subsequent fetches
(self)
| 43 | self.__test_fetch_timeout() |
| 44 | |
| 45 | def __test_fetch_timeout(self): |
| 46 | """Tests FETCH_ROWS_TIMEOUT_MS by running a query that produces RowBatches with a |
| 47 | large delay. The test waits for the query to 'finish' and then fetches the first |
| 48 | RowBatch, which should always be available since a query is only considered |
| 49 | 'finished' if rows are available. Subsequent fetches should time out because |
| 50 | RowBatch production has been delayed.""" |
| 51 | # Construct a query where there is a large delay between RowBatch production. |
| 52 | num_rows = 2 |
| 53 | statement = "select bool_col, avg(id) from functional.alltypes group by bool_col " \ |
| 54 | "having avg(id) != sleep(5000)" |
| 55 | execute_statement_resp = self.execute_statement(statement, |
| 56 | conf_overlay={'fetch_rows_timeout_ms': '1', 'batch_size': '1', 'num_nodes': '1'}) |
| 57 | HS2TestSuite.check_response(execute_statement_resp) |
| 58 | |
| 59 | # Wait for rows to be available for fetch. |
| 60 | get_operation_status_resp = self.wait_for_operation_state( |
| 61 | execute_statement_resp.operationHandle, |
| 62 | TCLIService.TOperationState.FINISHED_STATE, timeout=30) |
| 63 | HS2TestSuite.check_response(get_operation_status_resp) |
| 64 | |
| 65 | # Assert that exactly 1 row can be fetched. |
| 66 | FetchTimeoutUtils.fetch_num_rows(self.hs2_client, |
| 67 | execute_statement_resp.operationHandle, 1, statement) |
| 68 | |
| 69 | # Assert that the next fetch request times out while waiting for a RowBatch to be |
| 70 | # produced. |
| 71 | fetch_results_resp = self.hs2_client.FetchResults( |
| 72 | TCLIService.TFetchResultsReq( |
| 73 | operationHandle=execute_statement_resp.operationHandle, maxRows=num_rows)) |
| 74 | HS2TestSuite.check_response(fetch_results_resp) |
| 75 | num_rows_fetched = HS2TestSuite.get_num_rows(fetch_results_resp.results) |
| 76 | assert num_rows_fetched == 0 |
| 77 | assert fetch_results_resp.hasMoreRows |
| 78 | FetchTimeoutUtils.fetch_num_rows(self.hs2_client, |
| 79 | execute_statement_resp.operationHandle, 1, statement) |
| 80 | |
| 81 | @needs_session() |
| 82 | def test_fetch_materialization_timeout(self): |
no test coverage detected