Add NO_COMPACT version to CQL protocol, patch for CASSANDRA-10857. (apache#846)

ifesdjeen · mambocab · commit ee515ac26e52 · 2017-10-27T13:00:12.000-04:00
* Add `NO_COMPACT` version to CQL protocol, patch for CASSANDRA-10857.

* add changelog entry; preserve Connection.__init__ API
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
@@ -9,6 +9,7 @@ Features
 * Warn on Cluster init if contact points are specified but LBP isn't (legacy mode) (PYTHON-812)
 * Warn on Cluster init if contact points are specified but LBP isn't (exection profile mode) (PYTHON-838)
 * Include hash of result set metadata in prepared stmt id (PYTHON-808)
+* Add NO_COMPACT startup option (PYTHON-839)
 
 Bug Fixes
 ---------
diff --git a/cassandra/cluster.py b/cassandra/cluster.py
@@ -395,6 +395,9 @@ class Cluster(object):
     """
 
     allow_beta_protocol_version = False
+
+    no_compact = False
+
     """
     Setting true injects a flag in all messages that makes the server accept and use "beta" protocol version.
     Used for testing new protocol features incrementally before the new version is complete.
@@ -788,7 +791,8 @@ def __init__(self,
                  execution_profiles=None,
                  allow_beta_protocol_version=False,
                  timestamp_generator=None,
-                 idle_heartbeat_timeout=30):
+                 idle_heartbeat_timeout=30,
+                 no_compact=False):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
         extablishing connection pools or refreshing metadata.
@@ -821,6 +825,8 @@ def __init__(self,
             self._protocol_version_explicit = True
         self.allow_beta_protocol_version = allow_beta_protocol_version
 
+        self.no_compact = no_compact
+
         self.auth_provider = auth_provider
 
         if load_balancing_policy is not None:
@@ -1204,6 +1210,7 @@ def _make_connection_kwargs(self, address, kwargs_dict):
         kwargs_dict.setdefault('protocol_version', self.protocol_version)
         kwargs_dict.setdefault('user_type_map', self._user_types)
         kwargs_dict.setdefault('allow_beta_protocol_version', self.allow_beta_protocol_version)
+        kwargs_dict.setdefault('no_compact', self.no_compact)
 
         return kwargs_dict
 
diff --git a/cassandra/connection.py b/cassandra/connection.py
@@ -203,6 +203,7 @@ class Connection(object):
     out_buffer_size = 4096
 
     cql_version = None
+    no_compact = False
     protocol_version = ProtocolVersion.MAX_SUPPORTED
 
     keyspace = None
@@ -259,7 +260,7 @@ class Connection(object):
     def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
                  ssl_options=None, sockopts=None, compression=True,
                  cql_version=None, protocol_version=ProtocolVersion.MAX_SUPPORTED, is_control_connection=False,
-                 user_type_map=None, connect_timeout=None, allow_beta_protocol_version=False):
+                 user_type_map=None, connect_timeout=None, allow_beta_protocol_version=False, no_compact=False):
         self.host = host
         self.port = port
         self.authenticator = authenticator
@@ -272,6 +273,7 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
         self.user_type_map = user_type_map
         self.connect_timeout = connect_timeout
         self.allow_beta_protocol_version = allow_beta_protocol_version
+        self.no_compact = no_compact
         self._push_watchers = defaultdict(set)
         self._requests = {}
         self._iobuf = io.BytesIO()
@@ -637,7 +639,7 @@ def _send_options_message(self):
                       "specified", id(self), self.host)
             self._compressor = None
             self.cql_version = DEFAULT_CQL_VERSION
-            self._send_startup_message()
+            self._send_startup_message(no_compact=self.no_compact)
         else:
             log.debug("Sending initial options message for new connection (%s) to %s", id(self), self.host)
             self.send_msg(OptionsMessage(), self.get_request_id(), self._handle_options_response)
@@ -703,14 +705,16 @@ def _handle_options_response(self, options_response):
                 self._compressor, self.decompressor = \
                     locally_supported_compressions[compression_type]
 
-        self._send_startup_message(compression_type)
+        self._send_startup_message(compression_type, no_compact=self.no_compact)
 
     @defunct_on_error
-    def _send_startup_message(self, compression=None):
+    def _send_startup_message(self, compression=None, no_compact=False):
         log.debug("Sending StartupMessage on %s", self)
         opts = {}
         if compression:
             opts['COMPRESSION'] = compression
+        if no_compact:
+            opts['NO_COMPACT'] = 'true'
         sm = StartupMessage(cqlversion=self.cql_version, options=opts)
         self.send_msg(sm, self.get_request_id(), cb=self._handle_startup_response)
         log.debug("Sent StartupMessage on %s", self)
diff --git a/cassandra/protocol.py b/cassandra/protocol.py
@@ -394,6 +394,7 @@ class StartupMessage(_MessageType):
     KNOWN_OPTION_KEYS = set((
         'CQL_VERSION',
         'COMPRESSION',
+        'NO_COMPACT'
     ))
 
     def __init__(self, cqlversion, options):
