Merge branch 'pr-665'

cassandra/cluster.py

@@ -3264,6 +3264,7 @@ class ResponseFuture(object):
     _req_id = None
     _final_result = _NOT_SET
     _col_names = None
+    _col_types = None
     _final_exception = None
     _query_traces = None
     _callbacks = None
@@ -3531,6 +3532,7 @@ class ResponseFuture(object):
                     results = getattr(response, 'results', None)
                     if results is not None and response.kind == RESULT_KIND_ROWS:
                         self._paging_state = response.paging_state
+                        self._col_types = response.col_types
                         self._col_names = results[0]
                         results = self.row_factory(*results)
                     self._set_final_result(results)
@@ -3954,6 +3956,7 @@ class ResultSet(object):
     def __init__(self, response_future, initial_response):
         self.response_future = response_future
         self.column_names = response_future._col_names
+        self.column_types = response_future._col_types
         self._set_current_rows(initial_response)
         self._page_iter = None
         self._list_mode = False

cassandra/protocol.py

@@ -601,19 +601,21 @@ class ResultMessage(_MessageType):
     _HAS_MORE_PAGES_FLAG = 0x0002
     _NO_METADATA_FLAG = 0x0004
 
-    def __init__(self, kind, results, paging_state=None):
+    def __init__(self, kind, results, paging_state=None, col_types=None):
         self.kind = kind
         self.results = results
         self.paging_state = paging_state
+        self.col_types = col_types
 
     @classmethod
     def recv_body(cls, f, protocol_version, user_type_map, result_metadata):
         kind = read_int(f)
         paging_state = None
+        col_types = None
         if kind == RESULT_KIND_VOID:
             results = None
         elif kind == RESULT_KIND_ROWS:
-            paging_state, results = cls.recv_results_rows(
+            paging_state, col_types, results = cls.recv_results_rows(
                 f, protocol_version, user_type_map, result_metadata)
         elif kind == RESULT_KIND_SET_KEYSPACE:
             ksname = read_string(f)
@@ -624,7 +626,7 @@ class ResultMessage(_MessageType):
             results = cls.recv_results_schema_change(f, protocol_version)
         else:
             raise DriverException("Unknown RESULT kind: %d" % kind)
-        return cls(kind, results, paging_state)
+        return cls(kind, results, paging_state, col_types)
 
     @classmethod
     def recv_results_rows(cls, f, protocol_version, user_type_map, result_metadata):
@@ -647,7 +649,7 @@ class ResultMessage(_MessageType):
                     raise DriverException('Failed decoding result column "%s" of type %s: %s' % (colnames[i],
                                                                                                  coltypes[i].cql_parameterized_type(),
                                                                                                  e.message))
-        return paging_state, (colnames, parsed_rows)
+        return paging_state, coltypes, (colnames, parsed_rows)
 
     @classmethod
     def recv_results_prepared(cls, f, protocol_version, user_type_map):

cassandra/row_parser.pyx

@@ -35,6 +35,6 @@ def make_recv_results_rows(ColumnParser colparser):
         reader = BytesIOReader(f.read())
         parsed_rows = colparser.parse_rows(reader, desc)
 
-        return (paging_state, (colnames, parsed_rows))
+        return (paging_state, coltypes, (colnames, parsed_rows))
 
     return recv_results_rows

tests/integration/standard/test_custom_protocol_handler.py

@@ -130,8 +130,9 @@ class CustomResultMessageRaw(ResultMessage):
             paging_state, column_metadata = cls.recv_results_metadata(f, user_type_map)
             rowcount = read_int(f)
             rows = [cls.recv_row(f, len(column_metadata)) for _ in range(rowcount)]
+            colnames = [c[2] for c in column_metadata]
             coltypes = [c[3] for c in column_metadata]
-            return (paging_state, (coltypes, rows))
+            return paging_state, coltypes, (colnames, rows)
 
 
 class CustomTestRawRowType(ProtocolHandler):
@@ -166,7 +167,7 @@ class CustomResultMessageTracked(ResultMessage):
             tuple(ctype.from_binary(val, protocol_version)
                   for ctype, val in zip(coltypes, row))
             for row in rows]
-        return (paging_state, (colnames, parsed_rows))
+        return paging_state, coltypes, (colnames, parsed_rows)
 
 
 class CustomProtocolHandlerResultMessageTracked(ProtocolHandler):

tests/unit/test_concurrent.py

@@ -37,6 +37,7 @@ class MockResponseResponseFuture():
 
     _query_trace = None
     _col_names = None
+    _col_types = None
 
     # a list pending callbacks, these will be prioritized in reverse or normal orderd
     pending_callbacks = PriorityQueue()

tests/unit/test_response_future.py

@@ -50,7 +50,7 @@ class ResponseFutureTests(unittest.TestCase):
         return ResponseFuture(session, message, query, 1)
 
     def make_mock_response(self, results):
-        return Mock(spec=ResultMessage, kind=RESULT_KIND_ROWS, results=results, paging_state=None)
+        return Mock(spec=ResultMessage, kind=RESULT_KIND_ROWS, results=results, paging_state=None, col_types=None)
 
     def test_result_message(self):
         session = self.make_basic_session()