Merge branch 'main' into sea-migration

Author: varun-edachali-dbx

CHANGELOG.md

@@ -1,5 +1,8 @@
 # Release History
 
+# 4.0.5 (2025-06-24)
+- Fix: Reverted change in cursor close handling which led to errors impacting users (databricks/databricks-sql-python#613 by @madhav-db)
+
 # 4.0.4 (2025-06-16)
 
 - Update thrift client library after cleaning up unused fields and structs (databricks/databricks-sql-python#553 by @vikrantpuppala)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "databricks-sql-connector"
-version = "4.0.4"
+version = "4.0.5"
 description = "Databricks SQL Connector for Python"
 authors = ["Databricks <[email protected]>"]
 license = "Apache-2.0"

src/databricks/sql/__init__.py

@@ -68,7 +68,7 @@ def __repr__(self):
 DATE = DBAPITypeObject("date")
 ROWID = DBAPITypeObject()
 
-__version__ = "4.0.4"
+__version__ = "4.0.5"
 USER_AGENT_NAME = "PyDatabricksSqlConnector"
 
 # These two functions are pyhive legacy

src/databricks/sql/client.py

@@ -281,13 +281,7 @@ def __enter__(self) -> "Connection":
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
-        try:
-            self.close()
-        except BaseException as e:
-            logger.warning(f"Exception during connection close in __exit__: {e}")
-            if exc_type is None:
-                raise
-        return False
+        self.close()
 
     def __del__(self):
         if self.open:
@@ -417,14 +411,7 @@ def __enter__(self) -> "Cursor":
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
-        try:
-            logger.debug("Cursor context manager exiting, calling close()")
-            self.close()
-        except BaseException as e:
-            logger.warning(f"Exception during cursor close in __exit__: {e}")
-            if exc_type is None:
-                raise
-        return False
+        self.close()
 
     def __iter__(self):
         if self.active_result_set:
@@ -1091,21 +1078,7 @@ def cancel(self) -> None:
     def close(self) -> None:
         """Close cursor"""
         self.open = False
-
-        # Close active operation handle if it exists
-        if self.active_command_id:
-            try:
-                self.backend.close_command(self.active_command_id)
-            except RequestError as e:
-                if isinstance(e.args[1], CursorAlreadyClosedError):
-                    logger.info("Operation was canceled by a prior request")
-                else:
-                    logging.warning(f"Error closing operation handle: {e}")
-            except Exception as e:
-                logging.warning(f"Error closing operation handle: {e}")
-            finally:
-                self.active_command_id = None
-
+        self.active_op_handle = None
         if self.active_result_set:
             self._close_and_clear_active_result_set()
 

tests/e2e/common/large_queries_mixin.py

@@ -83,11 +83,11 @@ def test_query_with_large_narrow_result_set(self):
                 assert row[0] == row_id
 
     def test_long_running_query(self):
-        """Incrementally increase query size until it takes at least 5 minutes,
+        """Incrementally increase query size until it takes at least 3 minutes,
         and asserts that the query completes successfully.
         """
         minutes = 60
-        min_duration = 5 * minutes
+        min_duration = 3 * minutes
 
         duration = -1
         scale0 = 10000
@@ -113,5 +113,5 @@ def test_long_running_query(self):
                 duration = time.time() - start
                 current_fraction = duration / min_duration
                 print("Took {} s with scale factor={}".format(duration, scale_factor))
-                # Extrapolate linearly to reach 5 min and add 50% padding to push over the limit
+                # Extrapolate linearly to reach 3 min and add 50% padding to push over the limit
                 scale_factor = math.ceil(1.5 * scale_factor / current_fraction)

tests/e2e/common/staging_ingestion_tests.py

@@ -46,7 +46,7 @@ def test_staging_ingestion_life_cycle(self, ingestion_user):
         ) as conn:
 
             cursor = conn.cursor()
-            query = f"PUT '{temp_path}' INTO 'stage://tmp/{ingestion_user}/tmp/11/15/file1.csv' OVERWRITE"
+            query = f"PUT '{temp_path}' INTO 'stage://tmp/{ingestion_user}/tmp/11/16/file1.csv' OVERWRITE"
             cursor.execute(query)
 
         # GET should succeed
@@ -57,7 +57,7 @@ def test_staging_ingestion_life_cycle(self, ingestion_user):
             extra_params={"staging_allowed_local_path": new_temp_path}
         ) as conn:
             cursor = conn.cursor()
-            query = f"GET 'stage://tmp/{ingestion_user}/tmp/11/15/file1.csv' TO '{new_temp_path}'"
+            query = f"GET 'stage://tmp/{ingestion_user}/tmp/11/16/file1.csv' TO '{new_temp_path}'"
             cursor.execute(query)
 
         with open(new_fh, "rb") as fp:
@@ -67,7 +67,7 @@ def test_staging_ingestion_life_cycle(self, ingestion_user):
 
         # REMOVE should succeed
 
-        remove_query = f"REMOVE 'stage://tmp/{ingestion_user}/tmp/11/15/file1.csv'"
+        remove_query = f"REMOVE 'stage://tmp/{ingestion_user}/tmp/11/16/file1.csv'"
 
         with self.connection(extra_params={"staging_allowed_local_path": "/"}) as conn:
             cursor = conn.cursor()
@@ -79,7 +79,7 @@ def test_staging_ingestion_life_cycle(self, ingestion_user):
                 Error, match="Staging operation over HTTP was unsuccessful: 404"
             ):
                 cursor = conn.cursor()
-                query = f"GET 'stage://tmp/{ingestion_user}/tmp/11/15/file1.csv' TO '{new_temp_path}'"
+                query = f"GET 'stage://tmp/{ingestion_user}/tmp/11/16/file1.csv' TO '{new_temp_path}'"
                 cursor.execute(query)
 
         os.remove(temp_path)

tests/e2e/test_driver.py

@@ -51,7 +51,7 @@
 
 from tests.e2e.common.uc_volume_tests import PySQLUCVolumeTestSuiteMixin
 
-from databricks.sql.exc import SessionAlreadyClosedError, CursorAlreadyClosedError
+from databricks.sql.exc import SessionAlreadyClosedError
 
 log = logging.getLogger(__name__)
 
@@ -809,142 +809,6 @@ def test_catalogs_returns_arrow_table(self):
             results = cursor.fetchall_arrow()
             assert isinstance(results, pyarrow.Table)
 
-    def test_close_connection_closes_cursors(self):
-
-        from databricks.sql.thrift_api.TCLIService import ttypes
-
-        with self.connection() as conn:
-            cursor = conn.cursor()
-            cursor.execute(
-                "SELECT id, id `id2`, id `id3` FROM RANGE(1000000) order by RANDOM()"
-            )
-            ars = cursor.active_result_set
-
-            # We must manually run this check because thrift_backend always forces `has_been_closed_server_side` to True
-            # Cursor op state should be open before connection is closed
-            status_request = ttypes.TGetOperationStatusReq(
-                operationHandle=ars.command_id.to_thrift_handle(),
-                getProgressUpdate=False,
-            )
-            op_status_at_server = ars.backend._client.GetOperationStatus(status_request)
-            assert op_status_at_server.operationState != CommandState.CLOSED
-
-            conn.close()
-
-            # When connection closes, any cursor operations should no longer exist at the server
-            with pytest.raises(SessionAlreadyClosedError) as cm:
-                op_status_at_server = ars.backend._client.GetOperationStatus(
-                    status_request
-                )
-
-    def test_closing_a_closed_connection_doesnt_fail(self, caplog):
-        caplog.set_level(logging.DEBUG)
-        # Second .close() call is when this context manager exits
-        with self.connection() as conn:
-            # First .close() call is explicit here
-            conn.close()
-        assert "Session appears to have been closed already" in caplog.text
-
-        conn = None
-        try:
-            with pytest.raises(KeyboardInterrupt):
-                with self.connection() as c:
-                    conn = c
-                    raise KeyboardInterrupt("Simulated interrupt")
-        finally:
-            if conn is not None:
-                assert (
-                    not conn.open
-                ), "Connection should be closed after KeyboardInterrupt"
-
-    def test_cursor_close_properly_closes_operation(self):
-        """Test that Cursor.close() properly closes the active operation handle on the server."""
-        with self.connection() as conn:
-            cursor = conn.cursor()
-            try:
-                cursor.execute("SELECT 1 AS test")
-                assert cursor.active_command_id is not None
-                cursor.close()
-                assert cursor.active_command_id is None
-                assert not cursor.open
-            finally:
-                if cursor.open:
-                    cursor.close()
-
-        conn = None
-        cursor = None
-        try:
-            with self.connection() as c:
-                conn = c
-                with pytest.raises(KeyboardInterrupt):
-                    with conn.cursor() as cur:
-                        cursor = cur
-                        raise KeyboardInterrupt("Simulated interrupt")
-        finally:
-            if cursor is not None:
-                assert (
-                    not cursor.open
-                ), "Cursor should be closed after KeyboardInterrupt"
-
-    def test_nested_cursor_context_managers(self):
-        """Test that nested cursor context managers properly close operations on the server."""
-        with self.connection() as conn:
-            with conn.cursor() as cursor1:
-                cursor1.execute("SELECT 1 AS test1")
-                assert cursor1.active_command_id is not None
-
-                with conn.cursor() as cursor2:
-                    cursor2.execute("SELECT 2 AS test2")
-                    assert cursor2.active_command_id is not None
-
-                # After inner context manager exit, cursor2 should be not open
-                assert not cursor2.open
-                assert cursor2.active_command_id is None
-
-            # After outer context manager exit, cursor1 should be not open
-            assert not cursor1.open
-            assert cursor1.active_command_id is None
-
-    def test_cursor_error_handling(self):
-        """Test that cursor close handles errors properly to prevent orphaned operations."""
-        with self.connection() as conn:
-            cursor = conn.cursor()
-
-            cursor.execute("SELECT 1 AS test")
-
-            op_handle = cursor.active_command_id
-
-            assert op_handle is not None
-
-            # Manually close the operation to simulate server-side closure
-            conn.session.backend.close_command(op_handle)
-
-            cursor.close()
-
-            assert not cursor.open
-
-    def test_result_set_close(self):
-        """Test that ResultSet.close() properly closes operations on the server and handles state correctly."""
-        with self.connection() as conn:
-            cursor = conn.cursor()
-            try:
-                cursor.execute("SELECT * FROM RANGE(10)")
-
-                result_set = cursor.active_result_set
-                assert result_set is not None
-
-                initial_op_state = result_set.status
-
-                result_set.close()
-
-                assert result_set.status == CommandState.CLOSED
-                assert result_set.status != initial_op_state
-
-                # Closing the result set again should be a no-op and not raise exceptions
-                result_set.close()
-            finally:
-                cursor.close()
-
 
 # use a RetrySuite to encapsulate these tests which we'll typically want to run together; however keep
 # the 429/503 subsuites separate since they execute under different circumstances.