Normalise Execution Response (clean backend interfaces) (#587)

* [squash from exec-sea] bring over execution phase changes

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove excess test

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* add docstring

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remvoe exec func in sea backend

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove excess files

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove excess models

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove excess sea backend tests

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* cleanup

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* re-introduce get_schema_desc

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove SeaResultSet

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* clean imports and attributes

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* pass CommandId to ExecResp

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove changes in types

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* add back essential types (ExecResponse, from_sea_state)

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* fix fetch types

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* excess imports

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* reduce diff by maintaining logs

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* fix int test types

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* move guid_to_hex_id import to utils

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* reduce diff in guid utils import

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* move arrow_schema_bytes back into ExecuteResult

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* maintain log

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove un-necessary assignment

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove un-necessary tuple response

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove un-ncessary verbose mocking

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* move Queue construction to ResultSert

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* move description to List[Tuple]

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* frmatting (black)

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* reduce diff (remove explicit tuple conversion)

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove has_more_rows from ExecuteResponse

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove un-necessary has_more_rows aclc

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* default has_more_rows to True

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* return has_more_rows from ExecResponse conversion during GetRespMetadata

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove unnecessary replacement

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* better mocked backend naming

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove has_more_rows test in ExecuteResponse

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* introduce replacement of original has_more_rows read test

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* call correct method in test_use_arrow_schema

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* call correct method in test_fall_back_to_hive_schema

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* re-introduce result response read test

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* simplify test

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove excess fetch_results mocks

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* more minimal changes to thrift_backend tests

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* move back to old table types

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove outdated arrow_schema_bytes return

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* remove duplicate import

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* rephrase model docstrings to explicitly denote that they are representations and not used over the wire

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* has_more_rows -> is_direct_results

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

* switch docstring format to align with Connection class

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

---------

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

Author: varun-edachali-dbx

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

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

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
 

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

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):