(Improvement) Remove CQL binary protocol v3

No one needs it anymore. Any reasonable version of Cassandra and ScyllaDB support
CQL binary protocol v4 (or higher).

This also assume that we test against Cassandra 2.2 or above, so removed some <= 2.1 items.

Signed-off-by: Yaniv Kaul <yaniv.kaul@scylladb.com>

Author: mykaul

CONTRIBUTING.rst

@@ -89,12 +89,11 @@ The test will start the appropriate Scylla clusters when necessary  but if you d
 Specify a Protocol Version for Tests
 ------------------------------------
 The protocol version defaults to:
-- 4 for Scylla >= 3.0 and Scylla Enterprise > 2019.
-- 3 for older versions of Scylla
-- 5 for Cassandra >= 4.0, 4 for Cassandra >= 2.2, 3 for Cassandra >= 2.1, 2 for Cassandra >= 2.0
+- 4 for Scylla.
+- 5 for Cassandra >= 4.0, 4 for Cassandra >= 2.2
 You can overwrite it with the ``PROTOCOL_VERSION`` environment variable::
 
-    PROTOCOL_VERSION=3 SCYLLA_VERSION="release:5.1" uv run pytest tests/integration/standard tests/integration/cqlengine/
+    PROTOCOL_VERSION=4 SCYLLA_VERSION="release:5.1" uv run pytest tests/integration/standard tests/integration/cqlengine/
 
 Seeing Test Logs in Real Time
 -----------------------------

cassandra/__init__.py

@@ -136,13 +136,6 @@ class ProtocolVersion(object):
     Defines native protocol versions supported by this driver.
     """
 
-    V3 = 3
-    """
-    v3, supported in Cassandra 2.1-->3.x+;
-    added support for protocol-level client-side timestamps (see :attr:`.Session.use_client_timestamp`),
-    serial consistency levels for :class:`~.BatchStatement`, and an improved connection pool.
-    """
-
     V4 = 4
     """
     v4, supported in Cassandra 2.2-->3.x+;
@@ -170,9 +163,9 @@ class ProtocolVersion(object):
     DSE private protocol v2, supported in DSE 6.0+
     """
 
-    SUPPORTED_VERSIONS = (V5, V4, V3)
+    SUPPORTED_VERSIONS = (V5, V4)
     """
-    A tuple of all supported protocol versions for ScyllaDB, including future v5 version.
+    A tuple of all supported protocol versions for ScyllaDB.
     """
 
     BETA_VERSIONS = (V6,)

cassandra/cluster.py

@@ -717,11 +717,9 @@ def application_info(self) -> Optional[ApplicationInfoBase]:
     @property
     def auth_provider(self):
         """
-        When :attr:`~.Cluster.protocol_version` is 2 or higher, this should
-        be an instance of a subclass of :class:`~cassandra.auth.AuthProvider`,
+        This should be an instance of a subclass of :class:`~cassandra.auth.AuthProvider`,
         such as :class:`~.PlainTextAuthProvider`.
 
-
         When not using authentication, this should be left as :const:`None`.
         """
         return self._auth_provider
@@ -735,12 +733,7 @@ def auth_provider(self, value):
         try:
             self._auth_provider_callable = value.new_authenticator
         except AttributeError:
-            if self.protocol_version > 1:
-                raise TypeError("auth_provider must implement the cassandra.auth.AuthProvider "
-                                "interface when protocol_version >= 2")
-            elif not callable(value):
-                raise TypeError("auth_provider must be callable when protocol_version == 1")
-            self._auth_provider_callable = value
+            raise TypeError("auth_provider must implement the cassandra.auth.AuthProvider interface")
 
         self._auth_provider = value
 
@@ -1557,7 +1550,7 @@ def register_user_type(self, keyspace, user_type, klass):
 
         Example::
 
-            cluster = Cluster(protocol_version=3)
+            cluster = Cluster(protocol_version=4)
             session = cluster.connect()
             session.set_keyspace('mykeyspace')
             session.execute("CREATE TYPE address (street text, zipcode int)")
@@ -1582,11 +1575,6 @@ def __init__(self, street, zipcode):
             print(row.id, row.location.street, row.location.zipcode)
 
         """
-        if self.protocol_version < 3:
-            log.warning("User Type serialization is only supported in native protocol version 3+ (%d in use). "
-                        "CQL encoding for simple statements will still work, but named tuples will "
-                        "be returned when reading type %s.%s.", self.protocol_version, keyspace, user_type)
-
         self._user_types[keyspace][user_type] = klass
         for session in tuple(self.sessions):
             session.user_type_registered(keyspace, user_type, klass)
@@ -2445,8 +2433,6 @@ def default_serial_consistency_level(self):
         The default :class:`~ConsistencyLevel` for serial phase of  conditional updates executed through
         this session.  This default may be overridden by setting the
         :attr:`~.Statement.serial_consistency_level` on individual statements.
-
-        Only valid for ``protocol_version >= 2``.
         """
         return self._default_serial_consistency_level
 
@@ -2957,11 +2943,6 @@ def _create_response_future(self, query, parameters, trace, custom_payload,
                 continuous_paging_options=continuous_paging_options,
                 result_metadata_id=prepared_statement.result_metadata_id)
         elif isinstance(query, BatchStatement):
-            if self._protocol_version < 2:
-                raise UnsupportedOperation(
-                    "BatchStatement execution is only supported with protocol version "
-                    "2 or higher (supported in Cassandra 2.0 and higher).  Consider "
-                    "setting Cluster.protocol_version to 2 to support this operation.")
             statement_keyspace = query.keyspace if ProtocolVersion.uses_keyspace_flag(self._protocol_version) else None
             message = BatchMessage(
                 query.batch_type, query._statements_and_parameters, cl,
@@ -3100,7 +3081,7 @@ def prepare(self, query, custom_payload=None, keyspace=None):
         prepared_keyspace = keyspace if keyspace else None
         prepared_statement = PreparedStatement.from_message(
             response.query_id, response.bind_metadata, response.pk_indexes, self.cluster.metadata, query, prepared_keyspace,
-            self._protocol_version, response.column_metadata, response.result_metadata_id, response.is_lwt, self.cluster.column_encryption_policy)
+            response.column_metadata, response.result_metadata_id, response.is_lwt, self.cluster.column_encryption_policy)
         prepared_statement.custom_payload = future.custom_payload
 
         self.cluster.add_prepared(response.query_id, prepared_statement)
@@ -4640,10 +4621,9 @@ def _set_result(self, host, connection, pool, response):
             self._custom_payload = getattr(response, 'custom_payload', None)
 
             if self._custom_payload and self.session.cluster.control_connection._tablets_routing_v1 and 'tablets-routing-v1' in self._custom_payload:
-                protocol = self.session.cluster.protocol_version
                 info = self._custom_payload.get('tablets-routing-v1')
                 ctype = types.lookup_casstype('TupleType(LongType, LongType, ListType(TupleType(UUIDType, Int32Type)))')
-                tablet_routing_info = ctype.from_binary(info, protocol)
+                tablet_routing_info = ctype.from_binary(info)
                 first_token = tablet_routing_info[0]
                 last_token = tablet_routing_info[1]
                 tablet_replicas = tablet_routing_info[2]

cassandra/connection.py

@@ -41,7 +41,7 @@
 from cassandra import ConsistencyLevel, AuthenticationFailed, OperationTimedOut, ProtocolVersion
 from cassandra.marshal import int32_pack
 from cassandra.protocol import (ReadyMessage, AuthenticateMessage, OptionsMessage,
-                                StartupMessage, ErrorMessage, CredentialsMessage,
+                                StartupMessage, ErrorMessage,
                                 QueryMessage, ResultMessage, ProtocolHandler,
                                 InvalidRequestException, SupportedMessage,
                                 AuthResponseMessage, AuthChallengeMessage,
@@ -1477,18 +1477,12 @@ def _handle_startup_response(self, startup_response, did_authenticate=False):
             if ProtocolVersion.has_checksumming_support(self.protocol_version):
                 self._enable_checksumming()
 
-            if isinstance(self.authenticator, dict):
-                log.debug("Sending credentials-based auth response on %s", self)
-                cm = CredentialsMessage(creds=self.authenticator)
-                callback = partial(self._handle_startup_response, did_authenticate=True)
-                self.send_msg(cm, self.get_request_id(), cb=callback)
-            else:
-                log.debug("Sending SASL-based auth response on %s", self)
-                self.authenticator.server_authenticator_class = startup_response.authenticator
-                initial_response = self.authenticator.initial_response()
-                initial_response = "" if initial_response is None else initial_response
-                self.send_msg(AuthResponseMessage(initial_response), self.get_request_id(),
-                              self._handle_auth_response)
+            log.debug("Sending SASL-based auth response on %s", self)
+            self.authenticator.server_authenticator_class = startup_response.authenticator
+            initial_response = self.authenticator.initial_response()
+            initial_response = "" if initial_response is None else initial_response
+            self.send_msg(AuthResponseMessage(initial_response), self.get_request_id(),
+                          self._handle_auth_response)
         elif isinstance(startup_response, ErrorMessage):
             log.debug("Received ErrorMessage on new connection (%s) from %s: %s",
                       id(self), self.endpoint, startup_response.summary_msg())

cassandra/cqlengine/models.py

@@ -948,7 +948,7 @@ def _transform_column(col_name, col_obj):
             key_cols = [c for c in partition_keys.values()]
             partition_key_index = dict((col.db_field_name, col._partition_key_index) for col in key_cols)
             key_cql_types = [c.cql_type for c in key_cols]
-            key_serializer = staticmethod(lambda parts, proto_version: [t.to_binary(p, proto_version) for t, p in zip(key_cql_types, parts)])
+            key_serializer = staticmethod(lambda parts, proto_version: [t.to_binary(p) for t, p in zip(key_cql_types, parts)])
         else:
             partition_key_index = {}
             key_serializer = staticmethod(lambda parts, proto_version: None)