.description attribute

This PR implements the `description` attribute. To quote PEP-249:

```
This read-only attribute is a sequence of 7-item sequences.

Each of these sequences contains information describing one result column:

name
type_code
display_size
internal_size
precision
scale
null_ok
The first two items (name and type_code) are mandatory, the other five are optional and are set to None if no meaningful values can be provided.

This attribute will be None for operations that do not return rows or if the cursor has not had an operation invoked via the .execute*() method yet.

The type_code can be interpreted by comparing it to the Type Objects specified in the section below.
```

I only implemented the first two attributes for now as the others are optional, we can revisit the others later.

For now, the actual `description` `type_code` will return the name of the Scala types, as this is what the cmdexec server returns. However, once we put the translation layer in place, it will return the appropriate value in the Thrift type enum. This will bring it inline with the PyHive client.

* Runs against the ODBC smoke test
* New integration test to check comparison to type objects

Author: Niall Egan <niall.egan@databricks.com>

Author: Niall Egan

cmdexec/clients/python/src/dbsql/connector/__init__.py

@@ -1,5 +1,21 @@
 from .client import Connection
 
 
+class _DBAPITypeObject(object):
+    def __init__(self, *values):
+        self.values = values
+
+    def __eq__(self, other):
+        return other in self.values
+
+
+STRING = _DBAPITypeObject('string')
+BINARY = _DBAPITypeObject('binary')
+NUMBER = _DBAPITypeObject('boolean', 'byte', 'short', 'integer', 'long', 'double', 'decimal')
+DATETIME = _DBAPITypeObject('timestamp')
+DATE = _DBAPITypeObject('date')
+ROWID = _DBAPITypeObject()
+
+
 def connect(**kwargs):
     return Connection(**kwargs)

cmdexec/clients/python/src/dbsql/connector/client.py

@@ -64,7 +64,6 @@ def close(self):
 class Cursor:
     def __init__(self, connection, result_buffer_size_bytes=DEFAULT_RESULT_BUFFER_SIZE_BYTES):
         self.connection = connection
-        self.description = None
         self.rowcount = -1
         self.arraysize = 1
         self.buffer_size_bytes = result_buffer_size_bytes
@@ -88,12 +87,15 @@ def __iter__(self):
     def _response_to_result_set(self, execute_command_response, status):
         command_id = execute_command_response.command_id
         arrow_results = execute_command_response.results.arrow_ipc_stream
+        schema_message = execute_command_response.results.metadata \
+                         and execute_command_response.results.metadata.schema
         has_been_closed_server_side = execute_command_response.closed
         has_more_rows = execute_command_response.results.has_more_rows
         num_valid_rows = execute_command_response.results.num_valid_rows
 
         return ResultSet(self.connection, command_id, status, has_been_closed_server_side,
-                         has_more_rows, arrow_results, num_valid_rows, self.buffer_size_bytes)
+                         has_more_rows, arrow_results, num_valid_rows, schema_message,
+                         self.buffer_size_bytes)
 
     def _close_and_clear_active_result_set(self):
         try:
@@ -145,8 +147,10 @@ def execute(self, operation, query_params=None, metadata=None):
             command=operation,
             conf_overlay=None,
             row_limit=None,
-            result_options=None,
-        )
+            result_options=messages_pb2.CommandResultOptions(
+                max_bytes=self.buffer_size_bytes,
+                include_metadata=True,
+            ))
 
         execute_command_response = self.connection.base_client.make_request(
             self.connection.base_client.stub.ExecuteCommand, execute_command_request)
@@ -186,6 +190,13 @@ def close(self):
         if self.active_result_set:
             self._close_and_clear_active_result_set()
 
+    @property
+    def description(self):
+        if self.active_result_set:
+            return self.active_result_set.description
+        else:
+            return None
+
 
 class ResultSet:
     def __init__(self,
@@ -196,6 +207,7 @@ def __init__(self,
                  has_more_rows,
                  arrow_ipc_stream=None,
                  num_valid_rows=None,
+                 schema_message=None,
                  result_buffer_size_bytes=DEFAULT_RESULT_BUFFER_SIZE_BYTES):
         self.connection = connection
         self.command_id = command_id
@@ -204,14 +216,16 @@ def __init__(self,
         self.has_more_rows = has_more_rows
         self.buffer_size_bytes = result_buffer_size_bytes
         self._row_index = 0
+        self.description = None
 
         assert (self.status not in [messages_pb2.PENDING, messages_pb2.RUNNING])
 
         if arrow_ipc_stream:
-            # In the case we are passed in an initial result set, the server has taken the
-            # fast path and has no more rows to send
+            # In this case the server has taken the fast path and returned an initial batch of
+            # results
             self.results = ArrowQueue(
                 pyarrow.ipc.open_stream(arrow_ipc_stream).read_all(), num_valid_rows, 0)
+            self.description = self._get_schema_description(schema_message)
         else:
             # In this case, there are results waiting on the server so we fetch now for simplicity
             self._fill_results_buffer()
@@ -239,17 +253,19 @@ def _fetch_and_deserialize_results(self):
         arrow_table = pyarrow.ipc.open_stream(result_message.arrow_ipc_stream).read_all()
         results = ArrowQueue(arrow_table, num_valid_rows,
                              self._row_index - result_message.start_row_offset)
-        return results, result_message.has_more_rows
+        description = self._get_schema_description(result_message.metadata.schema)
+        return results, result_message.has_more_rows, description
 
     def _fill_results_buffer(self):
         if self.status == messages_pb2.CLOSED:
             raise Error("Can't fetch results on closed command %s" % self.command_id)
         elif self.status == messages_pb2.ERROR:
             raise DatabaseError("Command %s failed" % self.command_id)
         else:
-            results, has_more_rows = self._fetch_and_deserialize_results()
+            results, has_more_rows, description = self._fetch_and_deserialize_results()
             self.results = results
             self.has_more_rows = has_more_rows
+            self.description = description
 
     @staticmethod
     def _convert_arrow_table(table):
@@ -334,6 +350,21 @@ def close(self):
             self.has_been_closed_server_side = True
             self.status = messages_pb2.CLOSED
 
+    @staticmethod
+    def _get_schema_description(table_schema_message):
+        """
+        Takes a TableSchema message and returns a description 7-tuple as specified by PEP-249
+        """
+
+        def map_col_type(type_):
+            if type_.startswith('decimal'):
+                return 'decimal'
+            else:
+                return type_
+
+        return [(column.name, map_col_type(column.type), None, None, None, None, None)
+                for column in table_schema_message.columns]
+
 
 class ArrowQueue:
     def __init__(self, arrow_table, n_valid_rows, start_row_index):

cmdexec/clients/python/tests/test_fetches.py

@@ -1,4 +1,5 @@
 import unittest
+from unittest.mock import MagicMock
 from collections import deque
 
 import pyarrow as pa
@@ -40,13 +41,14 @@ 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
         arrow_ipc_stream = FetchTests.make_arrow_ipc_stream(initial_results)
         return client.ResultSet(
-            None,
-            None,
-            None,
-            True,
+            connection=None,
+            command_id=None,
+            status=None,
+            has_been_closed_server_side=True,
+            has_more_rows=False,
             arrow_ipc_stream=arrow_ipc_stream,
             num_valid_rows=len(initial_results),
-            has_more_rows=False)
+            schema_message=MagicMock())
 
     @staticmethod
     def make_dummy_result_set_from_batch_list(batch_list):
@@ -58,7 +60,8 @@ def _fetch_and_deserialize_results(self):
                 results = FetchTests.make_arrow_queue(batch_list[batch_index])
                 batch_index += 1
 
-                return results, batch_index < len(batch_list)
+                return results, batch_index < len(batch_list), \
+                       [('id', 'integer', None, None, None, None, None)]
 
         return SemiFakeResultSet(None, None, None, False, False)
 

cmdexec/clients/python/tests/tests.py

@@ -39,7 +39,7 @@ def test_missing_params_throws_interface_exception(self):
     @patch("%s.client.CmdExecBaseHttpClient" % PACKAGE_NAME)
     def test_close_uses_the_correct_session_id(self, mock_client_class):
         instance = mock_client_class.return_value
-        mock_response = Mock()
+        mock_response = MagicMock()
         mock_response.id = b'\x22'
         instance.make_request.return_value = mock_response
         good_connection_args = {"HOST": 1, "PORT": 1}
@@ -58,7 +58,7 @@ def test_closing_connection_closes_commands(self, mock_result_set_class, mock_cl
         for closed in (True, False):
             with self.subTest(closed=closed):
                 instance = mock_client_class.return_value
-                mock_response = Mock()
+                mock_response = MagicMock()
                 mock_response.id = b'\x22'
                 instance.make_request.return_value = mock_response
                 instance.stub.CloseCommand = Mock()
@@ -79,7 +79,7 @@ def test_closing_connection_closes_commands(self, mock_result_set_class, mock_cl
     @patch("%s.client.CmdExecBaseHttpClient" % PACKAGE_NAME)
     def test_cant_open_cursor_on_closed_connection(self, mock_client_class):
         instance = mock_client_class.return_value
-        mock_response = Mock()
+        mock_response = MagicMock()
         mock_response.id = b'\x22'
         instance.make_request.return_value = mock_response
         good_connection_args = {"HOST": 1, "PORT": 1}
@@ -95,17 +95,18 @@ def test_cant_open_cursor_on_closed_connection(self, mock_client_class):
     def test_closing_result_set_with_closed_connection_soft_closes_commands(
             self, pyarrow_ipc_open_stream):
         mock_connection = Mock()
-        mock_response = Mock()
+        mock_response = MagicMock()
         mock_response.id = b'\x22'
         mock_connection.base_client.make_request.return_value = mock_response
         result_set = client.ResultSet(
-            mock_connection,
-            b'\x10',
-            command_pb2.SUCCESS,
-            False,
+            connection=mock_connection,
+            command_id=b'\x10',
+            status=command_pb2.SUCCESS,
+            has_been_closed_server_side=False,
             arrow_ipc_stream=Mock(),
             num_valid_rows=0,
-            has_more_rows=False)
+            has_more_rows=False,
+            schema_message=MagicMock())
         mock_connection.open = False
 
         result_set.close()
@@ -118,7 +119,7 @@ def test_closing_result_set_with_closed_connection_soft_closes_commands(
     @patch("pyarrow.ipc.open_stream")
     def test_closing_result_set_hard_closes_commands(self, pyarrow_ipc_open_stream):
         mock_connection = Mock()
-        mock_response = Mock()
+        mock_response = MagicMock()
         mock_response.id = b'\x22'
         mock_response.results.start_row_offset = 0
         mock_connection.base_client.make_request.return_value = mock_response
@@ -135,7 +136,7 @@ def test_closing_result_set_hard_closes_commands(self, pyarrow_ipc_open_stream):
     @patch("%s.client.ResultSet" % PACKAGE_NAME)
     def test_executing_multiple_commands_uses_the_most_recent_command(self, mock_result_set_class):
         mock_client = Mock()
-        mock_response = Mock()
+        mock_response = MagicMock()
         mock_connection = Mock()
         mock_response.id = b'\x22'
         mock_response.status.state = command_pb2.SUCCESS
@@ -159,7 +160,7 @@ def test_executing_multiple_commands_uses_the_most_recent_command(self, mock_res
 
     def test_closed_cursor_doesnt_allow_operations(self):
         mock_connection = Mock()
-        mock_response = Mock()
+        mock_response = MagicMock()
         mock_response.id = b'\x22'
         mock_response.status.state = command_pb2.SUCCESS
         mock_connection.base_client.make_request.return_value = mock_response
@@ -178,7 +179,7 @@ def test_closed_cursor_doesnt_allow_operations(self):
     @patch("pyarrow.ipc.open_stream")
     def test_negative_fetch_throws_exception(self, pyarrow_ipc_open_stream_mock):
         mock_connection = Mock()
-        mock_response = Mock()
+        mock_response = MagicMock()
         mock_response.id = b'\x22'
         mock_response.results.start_row_offset = 0
         mock_response.status.state = command_pb2.SUCCESS
@@ -199,7 +200,7 @@ def test_context_manager_closes_cursor(self):
     @patch("%s.client.CmdExecBaseHttpClient" % PACKAGE_NAME)
     def test_context_manager_closes_connection(self, mock_client_class):
         instance = mock_client_class.return_value
-        mock_response = Mock()
+        mock_response = MagicMock()
         mock_response.id = b'\x22'
         instance.make_request.return_value = mock_response
         good_connection_args = {"HOST": 1, "PORT": 1}