(Improvement) Remove CQL binary protocol v3

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

@@ -422,8 +422,8 @@ def __str__(self):
                                 ', '.join('{0}={1}'.format(k, getattr(self, k)) for k in self._primary_keys.keys()))
 
     @classmethod
-    def _routing_key_from_values(cls, pk_values, protocol_version):
-        return cls._key_serializer(pk_values, protocol_version)
+    def _routing_key_from_values(cls, pk_values):
+        return cls._key_serializer(pk_values)
 
     @classmethod
     def _discover_polymorphic_submodels(cls):
@@ -948,10 +948,10 @@ 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: [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)
+            key_serializer = staticmethod(lambda parts: None)
 
         # setup partition key shortcut
         if len(partition_keys) == 0:

cassandra/cqlengine/query.py

@@ -1523,7 +1523,7 @@ def _execute_statement(model, statement, consistency_level, timeout, connection=
     if model._partition_key_index:
         key_values = statement.partition_key_values(model._partition_key_index)
         if not any(v is None for v in key_values):
-            parts = model._routing_key_from_values(key_values, conn.get_cluster(connection).protocol_version)
+            parts = model._routing_key_from_values(key_values)
             s.routing_key = parts
             s.keyspace = model._get_keyspace()
     connection = connection or model._get_connection()