Merge branch 'metadata-sea' into fetch-json-inline

Author: varun-edachali-dbx

src/databricks/sql/backend/sea/backend.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 import time
 import re
@@ -16,7 +18,7 @@
 
 if TYPE_CHECKING:
     from databricks.sql.client import Cursor
-    from databricks.sql.result_set import ResultSet
+    from databricks.sql.result_set import SeaResultSet
 
 from databricks.sql.backend.databricks_client import DatabricksClient
 from databricks.sql.backend.types import (
@@ -26,7 +28,7 @@
     BackendType,
     ExecuteResponse,
 )
-from databricks.sql.exc import DatabaseError, ServerOperationError
+from databricks.sql.exc import DatabaseError, ProgrammingError, ServerOperationError
 from databricks.sql.backend.sea.utils.http_client import SeaHttpClient
 from databricks.sql.types import SSLOptions
 
@@ -171,7 +173,7 @@ def _extract_warehouse_id(self, http_path: str) -> str:
             f"Note: SEA only works for warehouses."
         )
         logger.error(error_message)
-        raise ValueError(error_message)
+        raise ProgrammingError(error_message)
 
     @property
     def max_download_threads(self) -> int:
@@ -243,14 +245,14 @@ def close_session(self, session_id: SessionId) -> None:
             session_id: The session identifier returned by open_session()
 
         Raises:
-            ValueError: If the session ID is invalid
+            ProgrammingError: If the session ID is invalid
             OperationalError: If there's an error closing the session
         """
 
         logger.debug("SeaDatabricksClient.close_session(session_id=%s)", session_id)
 
         if session_id.backend_type != BackendType.SEA:
-            raise ValueError("Not a valid SEA session ID")
+            raise ProgrammingError("Not a valid SEA session ID")
         sea_session_id = session_id.to_sea_session_id()
 
         request_data = DeleteSessionRequest(
@@ -402,12 +404,12 @@ def execute_command(
         max_rows: int,
         max_bytes: int,
         lz4_compression: bool,
-        cursor: "Cursor",
+        cursor: Cursor,
         use_cloud_fetch: bool,
         parameters: List[ttypes.TSparkParameter],
         async_op: bool,
         enforce_embedded_schema_correctness: bool,
-    ) -> Union["ResultSet", None]:
+    ) -> Union[SeaResultSet, None]:
         """
         Execute a SQL command using the SEA backend.
 
@@ -428,7 +430,7 @@ def execute_command(
         """
 
         if session_id.backend_type != BackendType.SEA:
-            raise ValueError("Not a valid SEA session ID")
+            raise ProgrammingError("Not a valid SEA session ID")
 
         sea_session_id = session_id.to_sea_session_id()
 
@@ -503,11 +505,11 @@ def cancel_command(self, command_id: CommandId) -> None:
             command_id: Command identifier to cancel
 
         Raises:
-            ValueError: If the command ID is invalid
+            ProgrammingError: If the command ID is invalid
         """
 
         if command_id.backend_type != BackendType.SEA:
-            raise ValueError("Not a valid SEA command ID")
+            raise ProgrammingError("Not a valid SEA command ID")
 
         sea_statement_id = command_id.to_sea_statement_id()
 
@@ -526,11 +528,11 @@ def close_command(self, command_id: CommandId) -> None:
             command_id: Command identifier to close
 
         Raises:
-            ValueError: If the command ID is invalid
+            ProgrammingError: If the command ID is invalid
         """
 
         if command_id.backend_type != BackendType.SEA:
-            raise ValueError("Not a valid SEA command ID")
+            raise ProgrammingError("Not a valid SEA command ID")
 
         sea_statement_id = command_id.to_sea_statement_id()
 
@@ -552,7 +554,7 @@ def get_query_state(self, command_id: CommandId) -> CommandState:
             CommandState: The current state of the command
 
         Raises:
-            ValueError: If the command ID is invalid
+            ProgrammingError: If the command ID is invalid
         """
 
         if command_id.backend_type != BackendType.SEA:
@@ -574,8 +576,8 @@ def get_query_state(self, command_id: CommandId) -> CommandState:
     def get_execution_result(
         self,
         command_id: CommandId,
-        cursor: "Cursor",
-    ) -> "ResultSet":
+        cursor: Cursor,
+    ) -> SeaResultSet:
         """
         Get the result of a command execution.
 
@@ -584,14 +586,14 @@ def get_execution_result(
             cursor: Cursor executing the command
 
         Returns:
-            ResultSet: A SeaResultSet instance with the execution results
+            SeaResultSet: A SeaResultSet instance with the execution results
 
         Raises:
             ValueError: If the command ID is invalid
         """
 
         if command_id.backend_type != BackendType.SEA:
-            raise ValueError("Not a valid SEA command ID")
+            raise ProgrammingError("Not a valid SEA command ID")
 
         sea_statement_id = command_id.to_sea_statement_id()
 
@@ -628,8 +630,8 @@ def get_catalogs(
         session_id: SessionId,
         max_rows: int,
         max_bytes: int,
-        cursor: "Cursor",
-    ) -> "ResultSet":
+        cursor: Cursor,
+    ) -> SeaResultSet:
         """Get available catalogs by executing 'SHOW CATALOGS'."""
         result = self.execute_command(
             operation=MetadataCommands.SHOW_CATALOGS.value,
@@ -651,13 +653,13 @@ def get_schemas(
         session_id: SessionId,
         max_rows: int,
         max_bytes: int,
-        cursor: "Cursor",
+        cursor: Cursor,
         catalog_name: Optional[str] = None,
         schema_name: Optional[str] = None,
-    ) -> "ResultSet":
+    ) -> SeaResultSet:
         """Get schemas by executing 'SHOW SCHEMAS IN catalog [LIKE pattern]'."""
         if not catalog_name:
-            raise ValueError("Catalog name is required for get_schemas")
+            raise DatabaseError("Catalog name is required for get_schemas")
 
         operation = MetadataCommands.SHOW_SCHEMAS.value.format(catalog_name)
 
@@ -684,12 +686,12 @@ def get_tables(
         session_id: SessionId,
         max_rows: int,
         max_bytes: int,
-        cursor: "Cursor",
+        cursor: Cursor,
         catalog_name: Optional[str] = None,
         schema_name: Optional[str] = None,
         table_name: Optional[str] = None,
         table_types: Optional[List[str]] = None,
-    ) -> "ResultSet":
+    ) -> SeaResultSet:
         """Get tables by executing 'SHOW TABLES IN catalog [SCHEMA LIKE pattern] [LIKE pattern]'."""
         operation = (
             MetadataCommands.SHOW_TABLES_ALL_CATALOGS.value
@@ -719,12 +721,6 @@ def get_tables(
         )
         assert result is not None, "execute_command returned None in synchronous mode"
 
-        from databricks.sql.result_set import SeaResultSet
-
-        assert isinstance(
-            result, SeaResultSet
-        ), "execute_command returned a non-SeaResultSet"
-
         # Apply client-side filtering by table_types
         from databricks.sql.backend.sea.utils.filters import ResultSetFilter
 
@@ -737,15 +733,15 @@ def get_columns(
         session_id: SessionId,
         max_rows: int,
         max_bytes: int,
-        cursor: "Cursor",
+        cursor: Cursor,
         catalog_name: Optional[str] = None,
         schema_name: Optional[str] = None,
         table_name: Optional[str] = None,
         column_name: Optional[str] = None,
-    ) -> "ResultSet":
+    ) -> SeaResultSet:
         """Get columns by executing 'SHOW COLUMNS IN CATALOG catalog [SCHEMA LIKE pattern] [TABLE LIKE pattern] [LIKE pattern]'."""
         if not catalog_name:
-            raise ValueError("Catalog name is required for get_columns")
+            raise DatabaseError("Catalog name is required for get_columns")
 
         operation = MetadataCommands.SHOW_COLUMNS.value.format(catalog_name)
 

src/databricks/sql/backend/sea/utils/filters.py

@@ -112,7 +112,6 @@ def filter_by_column_values(
             result_set,
             lambda row: (
                 len(row) > column_index
-                and isinstance(row[column_index], str)
                 and (
                     row[column_index].upper()
                     if not case_sensitive

tests/unit/test_sea_backend.py

@@ -18,6 +18,7 @@
 from databricks.sql.exc import (
     Error,
     NotSupportedError,
+    ProgrammingError,
     ServerOperationError,
     DatabaseError,
 )
@@ -129,7 +130,7 @@ def test_initialization(self, mock_http_client):
         assert client3.max_download_threads == 5
 
         # Test with invalid HTTP path
-        with pytest.raises(ValueError) as excinfo:
+        with pytest.raises(ProgrammingError) as excinfo:
             SeaDatabricksClient(
                 server_hostname="test-server.databricks.com",
                 port=443,
@@ -195,7 +196,7 @@ def test_session_management(self, sea_client, mock_http_client, thrift_session_i
         )
 
         # Test close_session with invalid ID type
-        with pytest.raises(ValueError) as excinfo:
+        with pytest.raises(ProgrammingError) as excinfo:
             sea_client.close_session(thrift_session_id)
         assert "Not a valid SEA session ID" in str(excinfo.value)
 
@@ -244,7 +245,7 @@ def test_command_execution_sync(
             assert cmd_id_arg.guid == "test-statement-123"
 
         # Test with invalid session ID
-        with pytest.raises(ValueError) as excinfo:
+        with pytest.raises(ProgrammingError) as excinfo:
             mock_thrift_handle = MagicMock()
             mock_thrift_handle.sessionId.guid = b"guid"
             mock_thrift_handle.sessionId.secret = b"secret"
@@ -452,7 +453,7 @@ def test_command_management(
         )
 
         # Test cancel_command with invalid ID
-        with pytest.raises(ValueError) as excinfo:
+        with pytest.raises(ProgrammingError) as excinfo:
             sea_client.cancel_command(thrift_command_id)
         assert "Not a valid SEA command ID" in str(excinfo.value)
 
@@ -466,7 +467,7 @@ def test_command_management(
         )
 
         # Test close_command with invalid ID
-        with pytest.raises(ValueError) as excinfo:
+        with pytest.raises(ProgrammingError) as excinfo:
             sea_client.close_command(thrift_command_id)
         assert "Not a valid SEA command ID" in str(excinfo.value)
 
@@ -525,7 +526,7 @@ def test_command_management(
         assert result.status == CommandState.SUCCEEDED
 
         # Test get_execution_result with invalid ID
-        with pytest.raises(ValueError) as excinfo:
+        with pytest.raises(ProgrammingError) as excinfo:
             sea_client.get_execution_result(thrift_command_id, mock_cursor)
         assert "Not a valid SEA command ID" in str(excinfo.value)
 
@@ -721,7 +722,7 @@ def test_get_schemas(self, sea_client, sea_session_id, mock_cursor):
             )
 
             # Case 3: Without catalog name (should raise ValueError)
-            with pytest.raises(ValueError) as excinfo:
+            with pytest.raises(DatabaseError) as excinfo:
                 sea_client.get_schemas(
                     session_id=sea_session_id,
                     max_rows=100,
@@ -872,7 +873,7 @@ def test_get_columns(self, sea_client, sea_session_id, mock_cursor):
             )
 
             # Case 3: Without catalog name (should raise ValueError)
-            with pytest.raises(ValueError) as excinfo:
+            with pytest.raises(DatabaseError) as excinfo:
                 sea_client.get_columns(
                     session_id=sea_session_id,
                     max_rows=100,