diff --git a/src/databricks/sql/client.py b/src/databricks/sql/client.py index d45d51181..a775c4e45 100755 --- a/src/databricks/sql/client.py +++ b/src/databricks/sql/client.py @@ -16,6 +16,7 @@ from databricks.sql import * from databricks.sql.exc import ( OperationalError, + RequestError, SessionAlreadyClosedError, CursorAlreadyClosedError, InterfaceError, @@ -1759,9 +1760,37 @@ def cancel(self) -> None: def close(self) -> None: """Close cursor""" self.open = False - self.active_command_id = None if self.active_result_set: self._close_and_clear_active_result_set() + elif self.active_command_id is not None and self.connection.open: + # Async submission whose result was never fetched (no + # get_execution_result call), so the result-set close path never + # fired. Issue an explicit close_command to free the server-side + # statement handle instead of leaking it until session close. + # Gate on connection.open (mirroring ResultSet.close) so we don't + # attempt a network call on an already-torn-down session. + try: + self.backend.close_command(self.active_command_id) + except RequestError as e: + if len(e.args) > 1 and isinstance(e.args[1], CursorAlreadyClosedError): + # Already-closed handle (e.g. a prior cancel() or concurrent + # session teardown) is an expected, benign case — mirror + # ResultSet.close and log at info, not warning. + logger.info("Operation was canceled by a prior request") + else: + logger.warning("close_command on cursor close failed: %s", e) + except Exception as exc: + # Unlike the RequestError branch above (an expected, often + # transient network failure), reaching here means something we + # did not anticipate — surface the full traceback so it is + # diagnosable rather than indistinguishable from a benign + # network blip. + logger.warning( + "close_command on cursor close failed unexpectedly: %s", + exc, + exc_info=True, + ) + self.active_command_id = None @property def query_id(self) -> Optional[str]: diff --git a/tests/e2e/test_driver.py b/tests/e2e/test_driver.py index 5fe3db037..5515cff7d 100644 --- a/tests/e2e/test_driver.py +++ b/tests/e2e/test_driver.py @@ -342,6 +342,50 @@ def test_execute_async__large_result(self, extra_params): assert len(result) == x_dimension * y_dimension + @pytest.mark.parametrize( + "extra_params", + [ + {}, + { + "use_sea": True, + }, + ], + ) + def test_execute_async__close_without_fetch_frees_handle(self, extra_params): + """Closing a cursor whose async command result was never fetched must free + the server-side statement handle (issue #791). Otherwise the handle leaks + until the session closes.""" + with self.cursor(extra_params) as cursor: + cursor.execute_async("SELECT 1") + + # Capture the server-side command id before we close the cursor. + command_id = cursor.active_command_id + assert command_id is not None + + backend = cursor.backend + + # Sanity: the handle is live and pollable before close. + backend.get_query_state(command_id) + + # User decides not to wait for the result and closes the cursor + # without ever calling get_async_execution_result(). + cursor.close() + + # After close, the server-side handle must have been freed. How a + # re-poll of the saved command id surfaces that is backend-specific: + # - Thrift raises a server error on the closed handle. + # - SEA's get_query_state() does a plain GET and returns the + # status.state, so a freed statement may come back as a terminal + # CLOSED/CANCELLED state instead of raising. + # Accept either signal as proof the handle was freed. Pre-fix (leak), + # the poll instead returns a live/terminal-success state. + try: + state = backend.get_query_state(command_id) + except (RequestError, OperationalError, DatabaseError): + pass + else: + assert state in (CommandState.CLOSED, CommandState.CANCELLED) + # Exclude Retry tests because they require specific setups, and LargeQueries too slow for core # tests diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 66a3722ec..c21ddbc13 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -161,6 +161,93 @@ def test_closing_connection_closes_commands(self, mock_thrift_client_class): # Should NOT have called backend.close_command (already closed) mock_backend.close_command.assert_not_called() + def test_cursor_close_frees_command_when_no_result_set(self): + """Closing a cursor with an unfetched async command frees the server handle. + + When active_result_set is None but active_command_id is set (an async + submission whose result was never fetched), close() must issue an + explicit backend.close_command so the server-side statement handle is + not leaked. This branch is backend-agnostic (Thrift/SEA/kernel). + """ + mock_backend = Mock(spec=ThriftDatabricksClient) + mock_connection = Mock() + cursor = client.Cursor(connection=mock_connection, backend=mock_backend) + + command_id = Mock(spec=CommandId) + cursor.active_command_id = command_id + cursor.active_result_set = None + + cursor.close() + + mock_backend.close_command.assert_called_once_with(command_id) + self.assertIsNone(cursor.active_command_id) + + def test_cursor_close_skips_command_when_connection_closed(self): + """Closing a cursor after its session is gone must not call close_command. + + Mirrors ResultSet.close, which gates backend.close_command on + connection.open, to avoid a network call on a dead session (and a + spurious warning) during shutdown ordering. + """ + mock_backend = Mock(spec=ThriftDatabricksClient) + mock_connection = Mock() + mock_connection.open = False + cursor = client.Cursor(connection=mock_connection, backend=mock_backend) + + cursor.active_command_id = Mock(spec=CommandId) + cursor.active_result_set = None + + cursor.close() + + mock_backend.close_command.assert_not_called() + self.assertIsNone(cursor.active_command_id) + + def test_cursor_close_does_not_double_close_when_result_set_present(self): + """Closing a cursor with an active result set must NOT call close_command. + + The result-set close path is responsible for freeing the handle in that + case, so the elif branch must not fire (no double-close). + """ + mock_backend = Mock(spec=ThriftDatabricksClient) + mock_connection = Mock() + cursor = client.Cursor(connection=mock_connection, backend=mock_backend) + + cursor.active_command_id = Mock(spec=CommandId) + result_set = Mock(spec=ResultSet) + cursor.active_result_set = result_set + + cursor.close() + + result_set.close.assert_called_once() + mock_backend.close_command.assert_not_called() + self.assertIsNone(cursor.active_command_id) + + def test_cursor_close_tolerates_single_arg_request_error(self): + """A RequestError with only one positional arg must not escape close(). + + Some RequestErrors (e.g. from unified_http_client during shutdown) are + built with a single message arg, so e.args[1] would IndexError. close() + must guard the length check and swallow the error like any other + close_command failure. + """ + from databricks.sql.exc import RequestError + + mock_backend = Mock(spec=ThriftDatabricksClient) + mock_backend.close_command.side_effect = RequestError( + "HTTP client is closing or has been closed" + ) + mock_connection = Mock() + cursor = client.Cursor(connection=mock_connection, backend=mock_backend) + + cursor.active_command_id = Mock(spec=CommandId) + cursor.active_result_set = None + + # Should not raise despite the single-arg RequestError. + cursor.close() + + mock_backend.close_command.assert_called_once() + self.assertIsNone(cursor.active_command_id) + @patch("%s.session.ThriftDatabricksClient" % PACKAGE_NAME) def test_cant_open_cursor_on_closed_connection(self, mock_client_class): connection = databricks.sql.connect(**self.DUMMY_CONNECTION_ARGS)