Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 30 additions & 1 deletion src/databricks/sql/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
from databricks.sql import *
from databricks.sql.exc import (
OperationalError,
RequestError,
SessionAlreadyClosedError,
CursorAlreadyClosedError,
InterfaceError,
Expand Down Expand Up @@ -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)
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
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:
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
# 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]:
Expand Down
44 changes: 44 additions & 0 deletions tests/e2e/test_driver.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
[
{},
{
"use_sea": True,
},
],
)
def test_execute_async__close_without_fetch_frees_handle(self, extra_params):
Comment thread
peco-review-bot[bot] marked this conversation as resolved.
"""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)

Comment thread
peco-review-bot[bot] marked this conversation as resolved.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔵 Low — The SEA branch of this test may be brittle. After close_command issues a DELETE on the statement, the subsequent get_query_state does a plain GET (_poll_query) and returns whatever status.state the server reports. The test only tolerates a raised error or state in (CommandState.CLOSED, CommandState.CANCELLED). If SEA responds to a GET on a just-deleted statement with any other terminal/transient state (e.g. it still echoes SUCCEEDED briefly, or a state not in that tuple), the else assert fails and the test flakes — even though the handle was correctly freed. Since get_query_state intentionally does not call _check_command_not_in_failed_or_closed_state, there's no guarantee of a CLOSED/CANCELLED signal. Consider confirming empirically what SEA returns post-delete, or broadening the accepted signal (e.g. also accept a 404-derived error) so the test isn't tied to an unverified state assumption.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

NEEDS HUMAN DECISION — the bots can't resolve this thread; a maintainer's input is required.

The comment targets the SEA branch of an E2E test whose correct accepted-state set depends on what a live SEA warehouse actually returns from get_query_state() after close_command issues the statement DELETE. This job has no live-warehouse connection and must not run/add e2e tests, so I cannot verify SEA's real post-delete signal here. The reviewer's two options both hinge on that empirical fact: (a) confirm empirically — impossible in this job; (b) broaden the accepted states — unsafe to do blind, because the pre-fix leak "returns a live/terminal-success state," so broadening (e.g. accepting SUCCEEDED) without knowing SEA's freed-handle behavior would let the test pass in the leak case and destroy the regression it guards. A human needs to run this against a live SEA warehouse to observe the actual post-DELETE state (or 404), then either narrow the assertion to that confirmed signal or, if SEA gives no reliable freed signal, restructure the SEA branch (e.g. skip the re-poll assertion for SEA). Flagging for human judgment rather than guessing at a broadening that could mask the leak.


# Exclude Retry tests because they require specific setups, and LargeQueries too slow for core
# tests
Expand Down
87 changes: 87 additions & 0 deletions tests/unit/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading