diff --git a/src/databricks/sql/backend/databricks_client.py b/src/databricks/sql/backend/databricks_client.py index 20b059fa7..8fda71e1e 100644 --- a/src/databricks/sql/backend/databricks_client.py +++ b/src/databricks/sql/backend/databricks_client.py @@ -16,8 +16,6 @@ from databricks.sql.thrift_api.TCLIService import ttypes from databricks.sql.backend.types import SessionId, CommandId, CommandState -from databricks.sql.utils import ExecuteResponse -from databricks.sql.types import SSLOptions # Forward reference for type hints from typing import TYPE_CHECKING diff --git a/src/databricks/sql/backend/sea/models/requests.py b/src/databricks/sql/backend/sea/models/requests.py index 7966cb502..3175132bd 100644 --- a/src/databricks/sql/backend/sea/models/requests.py +++ b/src/databricks/sql/backend/sea/models/requests.py @@ -4,7 +4,7 @@ @dataclass class CreateSessionRequest: - """Request to create a new session.""" + """Representation of a request to create a new session.""" warehouse_id: str session_confs: Optional[Dict[str, str]] = None @@ -29,7 +29,7 @@ def to_dict(self) -> Dict[str, Any]: @dataclass class DeleteSessionRequest: - """Request to delete a session.""" + """Representation of a request to delete a session.""" warehouse_id: str session_id: str diff --git a/src/databricks/sql/backend/sea/models/responses.py b/src/databricks/sql/backend/sea/models/responses.py index 1bb54590f..4eeb9eef7 100644 --- a/src/databricks/sql/backend/sea/models/responses.py +++ b/src/databricks/sql/backend/sea/models/responses.py @@ -4,7 +4,7 @@ @dataclass class CreateSessionResponse: - """Response from creating a new session.""" + """Representation of the response from creating a new session.""" session_id: str diff --git a/src/databricks/sql/backend/sea/utils/http_client.py b/src/databricks/sql/backend/sea/utils/http_client.py index f0b931ee4..fe292919c 100644 --- a/src/databricks/sql/backend/sea/utils/http_client.py +++ b/src/databricks/sql/backend/sea/utils/http_client.py @@ -1,7 +1,7 @@ import json import logging import requests -from typing import Callable, Dict, Any, Optional, Union, List, Tuple +from typing import Callable, Dict, Any, Optional, List, Tuple from urllib.parse import urljoin from databricks.sql.auth.authenticators import AuthProvider diff --git a/src/databricks/sql/backend/thrift_backend.py b/src/databricks/sql/backend/thrift_backend.py index de388f1d4..e824de1c2 100644 --- a/src/databricks/sql/backend/thrift_backend.py +++ b/src/databricks/sql/backend/thrift_backend.py @@ -3,23 +3,21 @@ import logging import math import time -import uuid import threading -from typing import List, Optional, Union, Any, TYPE_CHECKING +from typing import List, Union, Any, TYPE_CHECKING if TYPE_CHECKING: from databricks.sql.client import Cursor - from databricks.sql.result_set import ResultSet, ThriftResultSet -from databricks.sql.thrift_api.TCLIService.ttypes import TOperationState from databricks.sql.backend.types import ( CommandState, SessionId, CommandId, - BackendType, + ExecuteResponse, ) from databricks.sql.backend.utils import guid_to_hex_id + try: import pyarrow except ImportError: @@ -42,7 +40,7 @@ ) from databricks.sql.utils import ( - ExecuteResponse, + ResultSetQueueFactory, _bound, RequestErrorInfo, NoRetryReason, @@ -53,6 +51,7 @@ ) from databricks.sql.types import SSLOptions from databricks.sql.backend.databricks_client import DatabricksClient +from databricks.sql.result_set import ResultSet, ThriftResultSet logger = logging.getLogger(__name__) @@ -758,11 +757,13 @@ def _results_message_to_execute_response(self, resp, operation_state): ) direct_results = resp.directResults has_been_closed_server_side = direct_results and direct_results.closeOperation - has_more_rows = ( + + is_direct_results = ( (not direct_results) or (not direct_results.resultSet) or direct_results.resultSet.hasMoreRows ) + description = self._hive_schema_to_description( t_result_set_metadata_resp.schema ) @@ -778,42 +779,28 @@ def _results_message_to_execute_response(self, resp, operation_state): schema_bytes = None lz4_compressed = t_result_set_metadata_resp.lz4Compressed - is_staging_operation = t_result_set_metadata_resp.isStagingOperation - if direct_results and direct_results.resultSet: - assert direct_results.resultSet.results.startRowOffset == 0 - assert direct_results.resultSetMetadata - - arrow_queue_opt = ResultSetQueueFactory.build_queue( - row_set_type=t_result_set_metadata_resp.resultFormat, - t_row_set=direct_results.resultSet.results, - arrow_schema_bytes=schema_bytes, - max_download_threads=self.max_download_threads, - lz4_compressed=lz4_compressed, - description=description, - ssl_options=self._ssl_options, - ) - else: - arrow_queue_opt = None - command_id = CommandId.from_thrift_handle(resp.operationHandle) - return ExecuteResponse( - arrow_queue=arrow_queue_opt, - status=CommandState.from_thrift_state(operation_state), - has_been_closed_server_side=has_been_closed_server_side, - has_more_rows=has_more_rows, - lz4_compressed=lz4_compressed, - is_staging_operation=is_staging_operation, + status = CommandState.from_thrift_state(operation_state) + if status is None: + raise ValueError(f"Unknown command state: {operation_state}") + + execute_response = ExecuteResponse( command_id=command_id, + status=status, description=description, + has_been_closed_server_side=has_been_closed_server_side, + lz4_compressed=lz4_compressed, + is_staging_operation=t_result_set_metadata_resp.isStagingOperation, arrow_schema_bytes=schema_bytes, + result_format=t_result_set_metadata_resp.resultFormat, ) + return execute_response, is_direct_results + def get_execution_result( self, command_id: CommandId, cursor: "Cursor" ) -> "ResultSet": - from databricks.sql.result_set import ThriftResultSet - thrift_handle = command_id.to_thrift_handle() if not thrift_handle: raise ValueError("Not a valid Thrift command ID") @@ -835,9 +822,6 @@ def get_execution_result( t_result_set_metadata_resp = resp.resultSetMetadata - lz4_compressed = t_result_set_metadata_resp.lz4Compressed - is_staging_operation = t_result_set_metadata_resp.isStagingOperation - has_more_rows = resp.hasMoreRows description = self._hive_schema_to_description( t_result_set_metadata_resp.schema ) @@ -852,26 +836,21 @@ def get_execution_result( else: schema_bytes = None - queue = ResultSetQueueFactory.build_queue( - row_set_type=resp.resultSetMetadata.resultFormat, - t_row_set=resp.results, - arrow_schema_bytes=schema_bytes, - max_download_threads=self.max_download_threads, - lz4_compressed=lz4_compressed, - description=description, - ssl_options=self._ssl_options, - ) + lz4_compressed = t_result_set_metadata_resp.lz4Compressed + is_staging_operation = t_result_set_metadata_resp.isStagingOperation + is_direct_results = resp.hasMoreRows + + status = self.get_query_state(command_id) execute_response = ExecuteResponse( - arrow_queue=queue, - status=CommandState.from_thrift_state(resp.status), + command_id=command_id, + status=status, + description=description, has_been_closed_server_side=False, - has_more_rows=has_more_rows, lz4_compressed=lz4_compressed, is_staging_operation=is_staging_operation, - command_id=command_id, - description=description, arrow_schema_bytes=schema_bytes, + result_format=t_result_set_metadata_resp.resultFormat, ) return ThriftResultSet( @@ -881,6 +860,10 @@ def get_execution_result( buffer_size_bytes=cursor.buffer_size_bytes, arraysize=cursor.arraysize, use_cloud_fetch=cursor.connection.use_cloud_fetch, + t_row_set=resp.results, + max_download_threads=self.max_download_threads, + ssl_options=self._ssl_options, + is_direct_results=is_direct_results, ) def _wait_until_command_done(self, op_handle, initial_operation_status_resp): @@ -947,8 +930,6 @@ def execute_command( async_op=False, enforce_embedded_schema_correctness=False, ) -> Union["ResultSet", None]: - from databricks.sql.result_set import ThriftResultSet - thrift_handle = session_id.to_thrift_handle() if not thrift_handle: raise ValueError("Not a valid Thrift session ID") @@ -995,7 +976,13 @@ def execute_command( self._handle_execute_response_async(resp, cursor) return None else: - execute_response = self._handle_execute_response(resp, cursor) + execute_response, is_direct_results = self._handle_execute_response( + resp, cursor + ) + + t_row_set = None + if resp.directResults and resp.directResults.resultSet: + t_row_set = resp.directResults.resultSet.results return ThriftResultSet( connection=cursor.connection, @@ -1004,6 +991,10 @@ def execute_command( buffer_size_bytes=max_bytes, arraysize=max_rows, use_cloud_fetch=use_cloud_fetch, + t_row_set=t_row_set, + max_download_threads=self.max_download_threads, + ssl_options=self._ssl_options, + is_direct_results=is_direct_results, ) def get_catalogs( @@ -1013,8 +1004,6 @@ def get_catalogs( max_bytes: int, cursor: "Cursor", ) -> "ResultSet": - from databricks.sql.result_set import ThriftResultSet - thrift_handle = session_id.to_thrift_handle() if not thrift_handle: raise ValueError("Not a valid Thrift session ID") @@ -1027,7 +1016,13 @@ def get_catalogs( ) resp = self.make_request(self._client.GetCatalogs, req) - execute_response = self._handle_execute_response(resp, cursor) + execute_response, is_direct_results = self._handle_execute_response( + resp, cursor + ) + + t_row_set = None + if resp.directResults and resp.directResults.resultSet: + t_row_set = resp.directResults.resultSet.results return ThriftResultSet( connection=cursor.connection, @@ -1036,6 +1031,10 @@ def get_catalogs( buffer_size_bytes=max_bytes, arraysize=max_rows, use_cloud_fetch=cursor.connection.use_cloud_fetch, + t_row_set=t_row_set, + max_download_threads=self.max_download_threads, + ssl_options=self._ssl_options, + is_direct_results=is_direct_results, ) def get_schemas( @@ -1047,8 +1046,6 @@ def get_schemas( catalog_name=None, schema_name=None, ) -> "ResultSet": - from databricks.sql.result_set import ThriftResultSet - thrift_handle = session_id.to_thrift_handle() if not thrift_handle: raise ValueError("Not a valid Thrift session ID") @@ -1063,7 +1060,13 @@ def get_schemas( ) resp = self.make_request(self._client.GetSchemas, req) - execute_response = self._handle_execute_response(resp, cursor) + execute_response, is_direct_results = self._handle_execute_response( + resp, cursor + ) + + t_row_set = None + if resp.directResults and resp.directResults.resultSet: + t_row_set = resp.directResults.resultSet.results return ThriftResultSet( connection=cursor.connection, @@ -1072,6 +1075,10 @@ def get_schemas( buffer_size_bytes=max_bytes, arraysize=max_rows, use_cloud_fetch=cursor.connection.use_cloud_fetch, + t_row_set=t_row_set, + max_download_threads=self.max_download_threads, + ssl_options=self._ssl_options, + is_direct_results=is_direct_results, ) def get_tables( @@ -1085,8 +1092,6 @@ def get_tables( table_name=None, table_types=None, ) -> "ResultSet": - from databricks.sql.result_set import ThriftResultSet - thrift_handle = session_id.to_thrift_handle() if not thrift_handle: raise ValueError("Not a valid Thrift session ID") @@ -1103,7 +1108,13 @@ def get_tables( ) resp = self.make_request(self._client.GetTables, req) - execute_response = self._handle_execute_response(resp, cursor) + execute_response, is_direct_results = self._handle_execute_response( + resp, cursor + ) + + t_row_set = None + if resp.directResults and resp.directResults.resultSet: + t_row_set = resp.directResults.resultSet.results return ThriftResultSet( connection=cursor.connection, @@ -1112,6 +1123,10 @@ def get_tables( buffer_size_bytes=max_bytes, arraysize=max_rows, use_cloud_fetch=cursor.connection.use_cloud_fetch, + t_row_set=t_row_set, + max_download_threads=self.max_download_threads, + ssl_options=self._ssl_options, + is_direct_results=is_direct_results, ) def get_columns( @@ -1125,8 +1140,6 @@ def get_columns( table_name=None, column_name=None, ) -> "ResultSet": - from databricks.sql.result_set import ThriftResultSet - thrift_handle = session_id.to_thrift_handle() if not thrift_handle: raise ValueError("Not a valid Thrift session ID") @@ -1143,7 +1156,13 @@ def get_columns( ) resp = self.make_request(self._client.GetColumns, req) - execute_response = self._handle_execute_response(resp, cursor) + execute_response, is_direct_results = self._handle_execute_response( + resp, cursor + ) + + t_row_set = None + if resp.directResults and resp.directResults.resultSet: + t_row_set = resp.directResults.resultSet.results return ThriftResultSet( connection=cursor.connection, @@ -1152,6 +1171,10 @@ def get_columns( buffer_size_bytes=max_bytes, arraysize=max_rows, use_cloud_fetch=cursor.connection.use_cloud_fetch, + t_row_set=t_row_set, + max_download_threads=self.max_download_threads, + ssl_options=self._ssl_options, + is_direct_results=is_direct_results, ) def _handle_execute_response(self, resp, cursor): diff --git a/src/databricks/sql/backend/types.py b/src/databricks/sql/backend/types.py index 9cd21b5e6..93bd7d525 100644 --- a/src/databricks/sql/backend/types.py +++ b/src/databricks/sql/backend/types.py @@ -1,5 +1,6 @@ +from dataclasses import dataclass from enum import Enum -from typing import Dict, Optional, Any +from typing import Dict, List, Optional, Any, Tuple import logging from databricks.sql.backend.utils import guid_to_hex_id @@ -80,6 +81,26 @@ def from_thrift_state( else: return None + @classmethod + def from_sea_state(cls, state: str) -> Optional["CommandState"]: + """ + Map SEA state string to CommandState enum. + Args: + state: SEA state string + Returns: + CommandState: The corresponding CommandState enum value + """ + state_mapping = { + "PENDING": cls.PENDING, + "RUNNING": cls.RUNNING, + "SUCCEEDED": cls.SUCCEEDED, + "FAILED": cls.FAILED, + "CLOSED": cls.CLOSED, + "CANCELED": cls.CANCELLED, + } + + return state_mapping.get(state, None) + class BackendType(Enum): """ @@ -394,3 +415,17 @@ def to_hex_guid(self) -> str: return guid_to_hex_id(self.guid) else: return str(self.guid) + + +@dataclass +class ExecuteResponse: + """Response from executing a SQL command.""" + + command_id: CommandId + status: CommandState + description: Optional[List[Tuple]] = None + has_been_closed_server_side: bool = False + lz4_compressed: bool = True + is_staging_operation: bool = False + arrow_schema_bytes: Optional[bytes] = None + result_format: Optional[Any] = None diff --git a/src/databricks/sql/client.py b/src/databricks/sql/client.py index 9f7c060a7..e145e4e58 100755 --- a/src/databricks/sql/client.py +++ b/src/databricks/sql/client.py @@ -24,7 +24,6 @@ from databricks.sql.backend.thrift_backend import ThriftDatabricksClient from databricks.sql.backend.databricks_client import DatabricksClient from databricks.sql.utils import ( - ExecuteResponse, ParamEscaper, inject_parameters, transform_paramstyle, diff --git a/src/databricks/sql/result_set.py b/src/databricks/sql/result_set.py index a0d8d3579..cf6940bb2 100644 --- a/src/databricks/sql/result_set.py +++ b/src/databricks/sql/result_set.py @@ -1,11 +1,11 @@ from abc import ABC, abstractmethod -from typing import List, Optional, Any, Union, TYPE_CHECKING +from typing import List, Optional, Any, Union, Tuple, TYPE_CHECKING import logging import time import pandas -from databricks.sql.backend.types import CommandId, CommandState +from databricks.sql.backend.sea.backend import SeaDatabricksClient try: import pyarrow @@ -13,14 +13,14 @@ pyarrow = None if TYPE_CHECKING: - from databricks.sql.backend.databricks_client import DatabricksClient from databricks.sql.backend.thrift_backend import ThriftDatabricksClient from databricks.sql.client import Connection - +from databricks.sql.backend.databricks_client import DatabricksClient from databricks.sql.thrift_api.TCLIService import ttypes from databricks.sql.types import Row from databricks.sql.exc import Error, RequestError, CursorAlreadyClosedError -from databricks.sql.utils import ExecuteResponse, ColumnTable, ColumnQueue +from databricks.sql.utils import ColumnTable, ColumnQueue +from databricks.sql.backend.types import CommandId, CommandState, ExecuteResponse logger = logging.getLogger(__name__) @@ -36,30 +36,45 @@ def __init__( self, connection: "Connection", backend: "DatabricksClient", - command_id: CommandId, - op_state: Optional[CommandState], - has_been_closed_server_side: bool, arraysize: int, buffer_size_bytes: int, + command_id: CommandId, + status: CommandState, + has_been_closed_server_side: bool = False, + is_direct_results: bool = False, + results_queue=None, + description=None, + is_staging_operation: bool = False, ): """ A ResultSet manages the results of a single command. - :param connection: The parent connection that was used to execute this command - :param backend: The specialised backend client to be invoked in the fetch phase - :param execute_response: A `ExecuteResponse` class returned by a command execution - :param result_buffer_size_bytes: The size (in bytes) of the internal buffer + max fetch - amount :param arraysize: The max number of rows to fetch at a time (PEP-249) + Parameters: + :param connection: The parent connection + :param backend: The backend client + :param arraysize: The max number of rows to fetch at a time (PEP-249) + :param buffer_size_bytes: The size (in bytes) of the internal buffer + max fetch + :param command_id: The command ID + :param status: The command status + :param has_been_closed_server_side: Whether the command has been closed on the server + :param is_direct_results: Whether the command has more rows + :param results_queue: The results queue + :param description: column description of the results + :param is_staging_operation: Whether the command is a staging operation """ - self.command_id = command_id - self.op_state = op_state - self.has_been_closed_server_side = has_been_closed_server_side + self.connection = connection self.backend = backend self.arraysize = arraysize self.buffer_size_bytes = buffer_size_bytes self._next_row_index = 0 - self.description = None + self.description = description + self.command_id = command_id + self.status = status + self.has_been_closed_server_side = has_been_closed_server_side + self.is_direct_results = is_direct_results + self.results = results_queue + self._is_staging_operation = is_staging_operation def __iter__(self): while True: @@ -74,10 +89,9 @@ def rownumber(self): return self._next_row_index @property - @abstractmethod def is_staging_operation(self) -> bool: """Whether this result set represents a staging operation.""" - pass + return self._is_staging_operation # Define abstract methods that concrete implementations must implement @abstractmethod @@ -119,7 +133,7 @@ def close(self) -> None: """ try: if ( - self.op_state != CommandState.CLOSED + self.status != CommandState.CLOSED and not self.has_been_closed_server_side and self.connection.open ): @@ -129,7 +143,7 @@ def close(self) -> None: logger.info("Operation was canceled by a prior request") finally: self.has_been_closed_server_side = True - self.op_state = CommandState.CLOSED + self.status = CommandState.CLOSED class ThriftResultSet(ResultSet): @@ -138,54 +152,73 @@ class ThriftResultSet(ResultSet): def __init__( self, connection: "Connection", - execute_response: ExecuteResponse, + execute_response: "ExecuteResponse", thrift_client: "ThriftDatabricksClient", buffer_size_bytes: int = 104857600, arraysize: int = 10000, use_cloud_fetch: bool = True, + t_row_set=None, + max_download_threads: int = 10, + ssl_options=None, + is_direct_results: bool = True, ): """ Initialize a ThriftResultSet with direct access to the ThriftDatabricksClient. - Args: - connection: The parent connection - execute_response: Response from the execute command - thrift_client: The ThriftDatabricksClient instance for direct access - buffer_size_bytes: Buffer size for fetching results - arraysize: Default number of rows to fetch - use_cloud_fetch: Whether to use cloud fetch for retrieving results + Parameters: + :param connection: The parent connection + :param execute_response: Response from the execute command + :param thrift_client: The ThriftDatabricksClient instance for direct access + :param buffer_size_bytes: Buffer size for fetching results + :param arraysize: Default number of rows to fetch + :param use_cloud_fetch: Whether to use cloud fetch for retrieving results + :param t_row_set: The TRowSet containing result data (if available) + :param max_download_threads: Maximum number of download threads for cloud fetch + :param ssl_options: SSL options for cloud fetch + :param is_direct_results: Whether there are more rows to fetch """ - super().__init__( - connection, - thrift_client, - execute_response.command_id, - execute_response.status, - execute_response.has_been_closed_server_side, - arraysize, - buffer_size_bytes, - ) - # Initialize ThriftResultSet-specific attributes - self.has_been_closed_server_side = execute_response.has_been_closed_server_side - self.has_more_rows = execute_response.has_more_rows - self.lz4_compressed = execute_response.lz4_compressed - self.description = execute_response.description self._arrow_schema_bytes = execute_response.arrow_schema_bytes self._use_cloud_fetch = use_cloud_fetch - self._is_staging_operation = execute_response.is_staging_operation + self.lz4_compressed = execute_response.lz4_compressed - # Initialize results queue - if execute_response.arrow_queue: - # In this case the server has taken the fast path and returned an initial batch of - # results - self.results = execute_response.arrow_queue - else: - # In this case, there are results waiting on the server so we fetch now for simplicity + # Build the results queue if t_row_set is provided + results_queue = None + if t_row_set and execute_response.result_format is not None: + from databricks.sql.utils import ResultSetQueueFactory + + # Create the results queue using the provided format + results_queue = ResultSetQueueFactory.build_queue( + row_set_type=execute_response.result_format, + t_row_set=t_row_set, + arrow_schema_bytes=execute_response.arrow_schema_bytes or b"", + max_download_threads=max_download_threads, + lz4_compressed=execute_response.lz4_compressed, + description=execute_response.description, + ssl_options=ssl_options, + ) + + # Call parent constructor with common attributes + super().__init__( + connection=connection, + backend=thrift_client, + arraysize=arraysize, + buffer_size_bytes=buffer_size_bytes, + command_id=execute_response.command_id, + status=execute_response.status, + has_been_closed_server_side=execute_response.has_been_closed_server_side, + is_direct_results=is_direct_results, + results_queue=results_queue, + description=execute_response.description, + is_staging_operation=execute_response.is_staging_operation, + ) + + # Initialize results queue if not provided + if not self.results: self._fill_results_buffer() def _fill_results_buffer(self): - # At initialization or if the server does not have cloud fetch result links available - results, has_more_rows = self.backend.fetch_results( + results, is_direct_results = self.backend.fetch_results( command_id=self.command_id, max_rows=self.arraysize, max_bytes=self.buffer_size_bytes, @@ -196,7 +229,7 @@ def _fill_results_buffer(self): use_cloud_fetch=self._use_cloud_fetch, ) self.results = results - self.has_more_rows = has_more_rows + self.is_direct_results = is_direct_results def _convert_columnar_table(self, table): column_names = [c[0] for c in self.description] @@ -280,7 +313,7 @@ def fetchmany_arrow(self, size: int) -> "pyarrow.Table": while ( n_remaining_rows > 0 and not self.has_been_closed_server_side - and self.has_more_rows + and self.is_direct_results ): self._fill_results_buffer() partial_results = self.results.next_n_rows(n_remaining_rows) @@ -305,7 +338,7 @@ def fetchmany_columnar(self, size: int): while ( n_remaining_rows > 0 and not self.has_been_closed_server_side - and self.has_more_rows + and self.is_direct_results ): self._fill_results_buffer() partial_results = self.results.next_n_rows(n_remaining_rows) @@ -320,7 +353,7 @@ def fetchall_arrow(self) -> "pyarrow.Table": results = self.results.remaining_rows() self._next_row_index += results.num_rows - while not self.has_been_closed_server_side and self.has_more_rows: + while not self.has_been_closed_server_side and self.is_direct_results: self._fill_results_buffer() partial_results = self.results.remaining_rows() if isinstance(results, ColumnTable) and isinstance( @@ -346,7 +379,7 @@ def fetchall_columnar(self): results = self.results.remaining_rows() self._next_row_index += results.num_rows - while not self.has_been_closed_server_side and self.has_more_rows: + while not self.has_been_closed_server_side and self.is_direct_results: self._fill_results_buffer() partial_results = self.results.remaining_rows() results = self.merge_columnar(results, partial_results) @@ -389,11 +422,6 @@ def fetchmany(self, size: int) -> List[Row]: else: return self._convert_arrow_table(self.fetchmany_arrow(size)) - @property - def is_staging_operation(self) -> bool: - """Whether this result set represents a staging operation.""" - return self._is_staging_operation - @staticmethod def _get_schema_description(table_schema_message): """ diff --git a/src/databricks/sql/session.py b/src/databricks/sql/session.py index 7c33d9b2d..76aec4675 100644 --- a/src/databricks/sql/session.py +++ b/src/databricks/sql/session.py @@ -10,7 +10,7 @@ from databricks.sql.backend.thrift_backend import ThriftDatabricksClient from databricks.sql.backend.sea.backend import SeaDatabricksClient from databricks.sql.backend.databricks_client import DatabricksClient -from databricks.sql.backend.types import SessionId +from databricks.sql.backend.types import SessionId, BackendType logger = logging.getLogger(__name__) diff --git a/src/databricks/sql/utils.py b/src/databricks/sql/utils.py index 2622b1172..d7b1b74b4 100644 --- a/src/databricks/sql/utils.py +++ b/src/databricks/sql/utils.py @@ -8,7 +8,7 @@ from collections.abc import Iterable from decimal import Decimal from enum import Enum -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Tuple, Union import re import lz4.frame @@ -57,7 +57,7 @@ def build_queue( max_download_threads: int, ssl_options: SSLOptions, lz4_compressed: bool = True, - description: Optional[List[List[Any]]] = None, + description: Optional[List[Tuple]] = None, ) -> ResultSetQueue: """ Factory method to build a result set queue. @@ -206,7 +206,7 @@ def __init__( start_row_offset: int = 0, result_links: Optional[List[TSparkArrowResultLink]] = None, lz4_compressed: bool = True, - description: Optional[List[List[Any]]] = None, + description: Optional[List[Tuple]] = None, ): """ A queue-like wrapper over CloudFetch arrow batches. @@ -349,13 +349,6 @@ def _create_empty_table(self) -> "pyarrow.Table": return create_arrow_table_from_arrow_file(self.schema_bytes, self.description) -ExecuteResponse = namedtuple( - "ExecuteResponse", - "status has_been_closed_server_side has_more_rows description lz4_compressed is_staging_operation " - "command_id arrow_queue arrow_schema_bytes", -) - - def _bound(min_x, max_x, x): """Bound x by [min_x, max_x] diff --git a/tests/e2e/common/retry_test_mixins.py b/tests/e2e/common/retry_test_mixins.py index b5d01a45d..dd509c062 100755 --- a/tests/e2e/common/retry_test_mixins.py +++ b/tests/e2e/common/retry_test_mixins.py @@ -326,7 +326,7 @@ def test_retry_abort_close_operation_on_404(self, caplog): with self.connection(extra_params={**self._retry_policy}) as conn: with conn.cursor() as curs: with patch( - "databricks.sql.utils.ExecuteResponse.has_been_closed_server_side", + "databricks.sql.backend.types.ExecuteResponse.has_been_closed_server_side", new_callable=PropertyMock, return_value=False, ): diff --git a/tests/e2e/test_driver.py b/tests/e2e/test_driver.py index 22897644f..8cfed7c28 100644 --- a/tests/e2e/test_driver.py +++ b/tests/e2e/test_driver.py @@ -933,12 +933,12 @@ def test_result_set_close(self): result_set = cursor.active_result_set assert result_set is not None - initial_op_state = result_set.op_state + initial_op_state = result_set.status result_set.close() - assert result_set.op_state == CommandState.CLOSED - assert result_set.op_state != initial_op_state + 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() diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py index 1a7950870..2054d01d1 100644 --- a/tests/unit/test_client.py +++ b/tests/unit/test_client.py @@ -26,7 +26,7 @@ from databricks.sql.types import Row from databricks.sql.result_set import ResultSet, ThriftResultSet from databricks.sql.backend.types import CommandId, CommandState -from databricks.sql.utils import ExecuteResponse +from databricks.sql.backend.types import ExecuteResponse from tests.unit.test_fetches import FetchTests from tests.unit.test_thrift_backend import ThriftBackendTestSuite @@ -48,7 +48,7 @@ def new(cls): is_staging_operation=False, command_id=None, has_been_closed_server_side=True, - has_more_rows=True, + is_direct_results=True, lz4_compressed=True, arrow_schema_bytes=b"schema", ) @@ -104,6 +104,7 @@ def test_closing_connection_closes_commands(self, mock_thrift_client_class): # Mock the backend that will be used by the real ThriftResultSet mock_backend = Mock(spec=ThriftDatabricksClient) mock_backend.staging_allowed_local_path = None + mock_backend.fetch_results.return_value = (Mock(), False) # Configure the decorator's mock to return our specific mock_backend mock_thrift_client_class.return_value = mock_backend @@ -121,10 +122,10 @@ def test_closing_connection_closes_commands(self, mock_thrift_client_class): # Verify initial state self.assertEqual(real_result_set.has_been_closed_server_side, closed) - expected_op_state = ( + expected_status = ( CommandState.CLOSED if closed else CommandState.SUCCEEDED ) - self.assertEqual(real_result_set.op_state, expected_op_state) + self.assertEqual(real_result_set.status, expected_status) # Mock execute_command to return our real result set cursor.backend.execute_command = Mock(return_value=real_result_set) @@ -146,8 +147,8 @@ def test_closing_connection_closes_commands(self, mock_thrift_client_class): # 1. has_been_closed_server_side should always be True after close() self.assertTrue(real_result_set.has_been_closed_server_side) - # 2. op_state should always be CLOSED after close() - self.assertEqual(real_result_set.op_state, CommandState.CLOSED) + # 2. status should always be CLOSED after close() + self.assertEqual(real_result_set.status, CommandState.CLOSED) # 3. Backend close_command should be called appropriately if not closed: @@ -184,6 +185,7 @@ def test_arraysize_buffer_size_passthrough( def test_closing_result_set_with_closed_connection_soft_closes_commands(self): mock_connection = Mock() mock_backend = Mock() + mock_backend.fetch_results.return_value = (Mock(), False) result_set = ThriftResultSet( connection=mock_connection, @@ -210,6 +212,7 @@ def test_closing_result_set_hard_closes_commands(self): mock_session.open = True type(mock_connection).session = PropertyMock(return_value=mock_session) + mock_thrift_backend.fetch_results.return_value = (Mock(), False) result_set = ThriftResultSet( mock_connection, mock_results_response, mock_thrift_backend ) @@ -254,7 +257,10 @@ def test_closed_cursor_doesnt_allow_operations(self): self.assertIn("closed", e.msg) def test_negative_fetch_throws_exception(self): - result_set = ThriftResultSet(Mock(), Mock(), Mock()) + mock_backend = Mock() + mock_backend.fetch_results.return_value = (Mock(), False) + + result_set = ThriftResultSet(Mock(), Mock(), mock_backend) with self.assertRaises(ValueError) as e: result_set.fetchmany(-1) @@ -472,7 +478,6 @@ def make_fake_row_slice(n_rows): mock_aq = Mock() mock_aq.next_n_rows.side_effect = make_fake_row_slice mock_thrift_backend.execute_command.return_value.arrow_queue = mock_aq - mock_thrift_backend.fetch_results.return_value = (mock_aq, True) cursor = client.Cursor(Mock(), mock_thrift_backend) cursor.execute("foo") @@ -556,7 +561,7 @@ def test_cursor_keeps_connection_alive(self, mock_client_class): self.assertEqual(instance.close_session.call_count, 0) cursor.close() - @patch("%s.utils.ExecuteResponse" % PACKAGE_NAME, autospec=True) + @patch("%s.backend.types.ExecuteResponse" % PACKAGE_NAME) @patch("%s.client.Cursor._handle_staging_operation" % PACKAGE_NAME) @patch("%s.session.ThriftDatabricksClient" % PACKAGE_NAME) def test_staging_operation_response_is_handled( @@ -678,10 +683,10 @@ def test_resultset_close_handles_cursor_already_closed_error(self): """Test that ResultSet.close() handles CursorAlreadyClosedError properly.""" result_set = client.ThriftResultSet.__new__(client.ThriftResultSet) result_set.backend = Mock() - result_set.backend.CLOSED_OP_STATE = "CLOSED" + result_set.backend.CLOSED_OP_STATE = CommandState.CLOSED result_set.connection = Mock() result_set.connection.open = True - result_set.op_state = "RUNNING" + result_set.status = CommandState.RUNNING result_set.has_been_closed_server_side = False result_set.command_id = Mock() @@ -695,7 +700,7 @@ def __init__(self): try: try: if ( - result_set.op_state != result_set.backend.CLOSED_OP_STATE + result_set.status != result_set.backend.CLOSED_OP_STATE and not result_set.has_been_closed_server_side and result_set.connection.open ): @@ -705,7 +710,7 @@ def __init__(self): pass finally: result_set.has_been_closed_server_side = True - result_set.op_state = result_set.backend.CLOSED_OP_STATE + result_set.status = result_set.backend.CLOSED_OP_STATE result_set.backend.close_command.assert_called_once_with( result_set.command_id @@ -713,7 +718,7 @@ def __init__(self): assert result_set.has_been_closed_server_side is True - assert result_set.op_state == result_set.backend.CLOSED_OP_STATE + assert result_set.status == result_set.backend.CLOSED_OP_STATE finally: pass diff --git a/tests/unit/test_fetches.py b/tests/unit/test_fetches.py index 030510a64..a649941e1 100644 --- a/tests/unit/test_fetches.py +++ b/tests/unit/test_fetches.py @@ -8,7 +8,8 @@ pa = None import databricks.sql.client as client -from databricks.sql.utils import ExecuteResponse, ArrowQueue +from databricks.sql.backend.types import ExecuteResponse +from databricks.sql.utils import ArrowQueue from databricks.sql.backend.thrift_backend import ThriftDatabricksClient from databricks.sql.result_set import ThriftResultSet @@ -39,26 +40,30 @@ def make_dummy_result_set_from_initial_results(initial_results): # If the initial results have been set, then we should never try and fetch more schema, arrow_table = FetchTests.make_arrow_table(initial_results) arrow_queue = ArrowQueue(arrow_table, len(initial_results), 0) + + # Create a mock backend that will return the queue when _fill_results_buffer is called + mock_thrift_backend = Mock(spec=ThriftDatabricksClient) + mock_thrift_backend.fetch_results.return_value = (arrow_queue, False) + + num_cols = len(initial_results[0]) if initial_results else 0 + description = [ + (f"col{col_id}", "integer", None, None, None, None, None) + for col_id in range(num_cols) + ] + rs = ThriftResultSet( connection=Mock(), execute_response=ExecuteResponse( + command_id=None, status=None, has_been_closed_server_side=True, - has_more_rows=False, - description=Mock(), - lz4_compressed=Mock(), - command_id=None, - arrow_queue=arrow_queue, - arrow_schema_bytes=schema.serialize().to_pybytes(), + description=description, + lz4_compressed=True, is_staging_operation=False, ), - thrift_client=None, + thrift_client=mock_thrift_backend, + t_row_set=None, ) - num_cols = len(initial_results[0]) if initial_results else 0 - rs.description = [ - (f"col{col_id}", "integer", None, None, None, None, None) - for col_id in range(num_cols) - ] return rs @staticmethod @@ -85,20 +90,19 @@ def fetch_results( mock_thrift_backend.fetch_results = fetch_results num_cols = len(batch_list[0][0]) if batch_list and batch_list[0] else 0 + description = [ + (f"col{col_id}", "integer", None, None, None, None, None) + for col_id in range(num_cols) + ] + rs = ThriftResultSet( connection=Mock(), execute_response=ExecuteResponse( + command_id=None, status=None, has_been_closed_server_side=False, - has_more_rows=True, - description=[ - (f"col{col_id}", "integer", None, None, None, None, None) - for col_id in range(num_cols) - ], - lz4_compressed=Mock(), - command_id=None, - arrow_queue=None, - arrow_schema_bytes=None, + description=description, + lz4_compressed=True, is_staging_operation=False, ), thrift_client=mock_thrift_backend, diff --git a/tests/unit/test_fetches_bench.py b/tests/unit/test_fetches_bench.py index b302c00da..e4a9e5cdd 100644 --- a/tests/unit/test_fetches_bench.py +++ b/tests/unit/test_fetches_bench.py @@ -10,7 +10,8 @@ import pytest import databricks.sql.client as client -from databricks.sql.utils import ExecuteResponse, ArrowQueue +from databricks.sql.backend.types import ExecuteResponse +from databricks.sql.utils import ArrowQueue @pytest.mark.skipif(pa is None, reason="PyArrow is not installed") @@ -35,7 +36,7 @@ def make_dummy_result_set_from_initial_results(arrow_table): execute_response=ExecuteResponse( status=None, has_been_closed_server_side=True, - has_more_rows=False, + is_direct_results=False, description=Mock(), command_id=None, arrow_queue=arrow_queue, diff --git a/tests/unit/test_thrift_backend.py b/tests/unit/test_thrift_backend.py index 57a2a61e3..57b5e9b58 100644 --- a/tests/unit/test_thrift_backend.py +++ b/tests/unit/test_thrift_backend.py @@ -623,7 +623,10 @@ def test_handle_execute_response_sets_compression_in_direct_results( status=Mock(), operationHandle=Mock(), directResults=ttypes.TSparkDirectResults( - operationStatus=Mock(), + operationStatus=ttypes.TGetOperationStatusResp( + status=self.okay_status, + operationState=ttypes.TOperationState.FINISHED_STATE, + ), resultSetMetadata=ttypes.TGetResultSetMetadataResp( status=self.okay_status, resultFormat=ttypes.TSparkRowSetType.ARROW_BASED_SET, @@ -644,7 +647,7 @@ def test_handle_execute_response_sets_compression_in_direct_results( ssl_options=SSLOptions(), ) - execute_response = thrift_backend._handle_execute_response( + execute_response, _ = thrift_backend._handle_execute_response( t_execute_resp, Mock() ) self.assertEqual(execute_response.lz4_compressed, lz4Compressed) @@ -832,9 +835,10 @@ def test_handle_execute_response_checks_direct_results_for_error_statuses(self): thrift_backend._handle_execute_response(error_resp, Mock()) self.assertIn("this is a bad error", str(cm.exception)) + @patch("databricks.sql.backend.thrift_backend.ThriftResultSet") @patch("databricks.sql.backend.thrift_backend.TCLIService.Client", autospec=True) def test_handle_execute_response_can_handle_without_direct_results( - self, tcli_service_class + self, tcli_service_class, mock_result_set ): tcli_service_instance = tcli_service_class.return_value @@ -878,11 +882,12 @@ def test_handle_execute_response_can_handle_without_direct_results( auth_provider=AuthProvider(), ssl_options=SSLOptions(), ) - results_message_response = thrift_backend._handle_execute_response( - execute_resp, Mock() - ) + ( + execute_response, + _, + ) = thrift_backend._handle_execute_response(execute_resp, Mock()) self.assertEqual( - results_message_response.status, + execute_response.status, CommandState.SUCCEEDED, ) @@ -946,8 +951,14 @@ def test_use_arrow_schema_if_available(self, tcli_service_class): tcli_service_instance.GetResultSetMetadata.return_value = ( t_get_result_set_metadata_resp ) + tcli_service_instance.GetOperationStatus.return_value = ( + ttypes.TGetOperationStatusResp( + status=self.okay_status, + operationState=ttypes.TOperationState.FINISHED_STATE, + ) + ) thrift_backend = self._make_fake_thrift_backend() - execute_response = thrift_backend._handle_execute_response( + execute_response, _ = thrift_backend._handle_execute_response( t_execute_resp, Mock() ) @@ -972,8 +983,14 @@ def test_fall_back_to_hive_schema_if_no_arrow_schema(self, tcli_service_class): ) tcli_service_instance.GetResultSetMetadata.return_value = hive_schema_req + tcli_service_instance.GetOperationStatus.return_value = ( + ttypes.TGetOperationStatusResp( + status=self.okay_status, + operationState=ttypes.TOperationState.FINISHED_STATE, + ) + ) thrift_backend = self._make_fake_thrift_backend() - thrift_backend._handle_execute_response(t_execute_resp, Mock()) + _, _ = thrift_backend._handle_execute_response(t_execute_resp, Mock()) self.assertEqual( hive_schema_mock, @@ -987,10 +1004,10 @@ def test_fall_back_to_hive_schema_if_no_arrow_schema(self, tcli_service_class): def test_handle_execute_response_reads_has_more_rows_in_direct_results( self, tcli_service_class, build_queue ): - for has_more_rows, resp_type in itertools.product( + for is_direct_results, resp_type in itertools.product( [True, False], self.execute_response_types ): - with self.subTest(has_more_rows=has_more_rows, resp_type=resp_type): + with self.subTest(is_direct_results=is_direct_results, resp_type=resp_type): tcli_service_instance = tcli_service_class.return_value results_mock = Mock() results_mock.startRowOffset = 0 @@ -1002,7 +1019,7 @@ def test_handle_execute_response_reads_has_more_rows_in_direct_results( resultSetMetadata=self.metadata_resp, resultSet=ttypes.TFetchResultsResp( status=self.okay_status, - hasMoreRows=has_more_rows, + hasMoreRows=is_direct_results, results=results_mock, ), closeOperation=Mock(), @@ -1018,11 +1035,12 @@ def test_handle_execute_response_reads_has_more_rows_in_direct_results( ) thrift_backend = self._make_fake_thrift_backend() - execute_response = thrift_backend._handle_execute_response( - execute_resp, Mock() - ) + ( + execute_response, + has_more_rows_result, + ) = thrift_backend._handle_execute_response(execute_resp, Mock()) - self.assertEqual(has_more_rows, execute_response.has_more_rows) + self.assertEqual(is_direct_results, has_more_rows_result) @patch( "databricks.sql.utils.ResultSetQueueFactory.build_queue", return_value=Mock() @@ -1031,10 +1049,10 @@ def test_handle_execute_response_reads_has_more_rows_in_direct_results( def test_handle_execute_response_reads_has_more_rows_in_result_response( self, tcli_service_class, build_queue ): - for has_more_rows, resp_type in itertools.product( + for is_direct_results, resp_type in itertools.product( [True, False], self.execute_response_types ): - with self.subTest(has_more_rows=has_more_rows, resp_type=resp_type): + with self.subTest(is_direct_results=is_direct_results, resp_type=resp_type): tcli_service_instance = tcli_service_class.return_value results_mock = MagicMock() results_mock.startRowOffset = 0 @@ -1047,7 +1065,7 @@ def test_handle_execute_response_reads_has_more_rows_in_result_response( fetch_results_resp = ttypes.TFetchResultsResp( status=self.okay_status, - hasMoreRows=has_more_rows, + hasMoreRows=is_direct_results, results=results_mock, resultSetMetadata=ttypes.TGetResultSetMetadataResp( resultFormat=ttypes.TSparkRowSetType.ARROW_BASED_SET @@ -1080,7 +1098,7 @@ def test_handle_execute_response_reads_has_more_rows_in_result_response( description=Mock(), ) - self.assertEqual(has_more_rows, has_more_rows_resp) + self.assertEqual(is_direct_results, has_more_rows_resp) @patch("databricks.sql.backend.thrift_backend.TCLIService.Client", autospec=True) def test_arrow_batches_row_count_are_respected(self, tcli_service_class): @@ -1135,9 +1153,10 @@ def test_arrow_batches_row_count_are_respected(self, tcli_service_class): self.assertEqual(arrow_queue.n_valid_rows, 15 * 10) + @patch("databricks.sql.backend.thrift_backend.ThriftResultSet") @patch("databricks.sql.backend.thrift_backend.TCLIService.Client", autospec=True) def test_execute_statement_calls_client_and_handle_execute_response( - self, tcli_service_class + self, tcli_service_class, mock_result_set ): tcli_service_instance = tcli_service_class.return_value response = Mock() @@ -1151,13 +1170,14 @@ def test_execute_statement_calls_client_and_handle_execute_response( ssl_options=SSLOptions(), ) thrift_backend._handle_execute_response = Mock() + thrift_backend._handle_execute_response.return_value = (Mock(), Mock()) cursor_mock = Mock() result = thrift_backend.execute_command( "foo", Mock(), 100, 200, Mock(), cursor_mock ) # Verify the result is a ResultSet - self.assertIsInstance(result, ResultSet) + self.assertEqual(result, mock_result_set.return_value) # Check call to client req = tcli_service_instance.ExecuteStatement.call_args[0][0] @@ -1169,9 +1189,10 @@ def test_execute_statement_calls_client_and_handle_execute_response( response, cursor_mock ) + @patch("databricks.sql.backend.thrift_backend.ThriftResultSet") @patch("databricks.sql.backend.thrift_backend.TCLIService.Client", autospec=True) def test_get_catalogs_calls_client_and_handle_execute_response( - self, tcli_service_class + self, tcli_service_class, mock_result_set ): tcli_service_instance = tcli_service_class.return_value response = Mock() @@ -1185,11 +1206,12 @@ def test_get_catalogs_calls_client_and_handle_execute_response( ssl_options=SSLOptions(), ) thrift_backend._handle_execute_response = Mock() + thrift_backend._handle_execute_response.return_value = (Mock(), Mock()) cursor_mock = Mock() result = thrift_backend.get_catalogs(Mock(), 100, 200, cursor_mock) # Verify the result is a ResultSet - self.assertIsInstance(result, ResultSet) + self.assertEqual(result, mock_result_set.return_value) # Check call to client req = tcli_service_instance.GetCatalogs.call_args[0][0] @@ -1200,9 +1222,10 @@ def test_get_catalogs_calls_client_and_handle_execute_response( response, cursor_mock ) + @patch("databricks.sql.backend.thrift_backend.ThriftResultSet") @patch("databricks.sql.backend.thrift_backend.TCLIService.Client", autospec=True) def test_get_schemas_calls_client_and_handle_execute_response( - self, tcli_service_class + self, tcli_service_class, mock_result_set ): tcli_service_instance = tcli_service_class.return_value response = Mock() @@ -1216,6 +1239,7 @@ def test_get_schemas_calls_client_and_handle_execute_response( ssl_options=SSLOptions(), ) thrift_backend._handle_execute_response = Mock() + thrift_backend._handle_execute_response.return_value = (Mock(), Mock()) cursor_mock = Mock() result = thrift_backend.get_schemas( @@ -1227,7 +1251,7 @@ def test_get_schemas_calls_client_and_handle_execute_response( schema_name="schema_pattern", ) # Verify the result is a ResultSet - self.assertIsInstance(result, ResultSet) + self.assertEqual(result, mock_result_set.return_value) # Check call to client req = tcli_service_instance.GetSchemas.call_args[0][0] @@ -1240,9 +1264,10 @@ def test_get_schemas_calls_client_and_handle_execute_response( response, cursor_mock ) + @patch("databricks.sql.backend.thrift_backend.ThriftResultSet") @patch("databricks.sql.backend.thrift_backend.TCLIService.Client", autospec=True) def test_get_tables_calls_client_and_handle_execute_response( - self, tcli_service_class + self, tcli_service_class, mock_result_set ): tcli_service_instance = tcli_service_class.return_value response = Mock() @@ -1256,6 +1281,7 @@ def test_get_tables_calls_client_and_handle_execute_response( ssl_options=SSLOptions(), ) thrift_backend._handle_execute_response = Mock() + thrift_backend._handle_execute_response.return_value = (Mock(), Mock()) cursor_mock = Mock() result = thrift_backend.get_tables( @@ -1269,7 +1295,7 @@ def test_get_tables_calls_client_and_handle_execute_response( table_types=["type1", "type2"], ) # Verify the result is a ResultSet - self.assertIsInstance(result, ResultSet) + self.assertEqual(result, mock_result_set.return_value) # Check call to client req = tcli_service_instance.GetTables.call_args[0][0] @@ -1284,9 +1310,10 @@ def test_get_tables_calls_client_and_handle_execute_response( response, cursor_mock ) + @patch("databricks.sql.backend.thrift_backend.ThriftResultSet") @patch("databricks.sql.backend.thrift_backend.TCLIService.Client", autospec=True) def test_get_columns_calls_client_and_handle_execute_response( - self, tcli_service_class + self, tcli_service_class, mock_result_set ): tcli_service_instance = tcli_service_class.return_value response = Mock() @@ -1300,6 +1327,7 @@ def test_get_columns_calls_client_and_handle_execute_response( ssl_options=SSLOptions(), ) thrift_backend._handle_execute_response = Mock() + thrift_backend._handle_execute_response.return_value = (Mock(), Mock()) cursor_mock = Mock() result = thrift_backend.get_columns( @@ -1313,7 +1341,7 @@ def test_get_columns_calls_client_and_handle_execute_response( column_name="column_pattern", ) # Verify the result is a ResultSet - self.assertIsInstance(result, ResultSet) + self.assertEqual(result, mock_result_set.return_value) # Check call to client req = tcli_service_instance.GetColumns.call_args[0][0] @@ -2202,14 +2230,23 @@ def test_protocol_v3_fails_if_initial_namespace_set(self, tcli_client_class): str(cm.exception), ) + @patch("databricks.sql.backend.thrift_backend.ThriftResultSet") @patch("databricks.sql.backend.thrift_backend.TCLIService.Client", autospec=True) @patch( "databricks.sql.backend.thrift_backend.ThriftDatabricksClient._handle_execute_response" ) def test_execute_command_sets_complex_type_fields_correctly( - self, mock_handle_execute_response, tcli_service_class + self, mock_handle_execute_response, tcli_service_class, mock_result_set ): tcli_service_instance = tcli_service_class.return_value + # Set up the mock to return a tuple with two values + mock_execute_response = Mock() + mock_arrow_schema = Mock() + mock_handle_execute_response.return_value = ( + mock_execute_response, + mock_arrow_schema, + ) + # Iterate through each possible combination of native types (True, False and unset) for complex, timestamp, decimals in itertools.product( [True, False, None], [True, False, None], [True, False, None]