(improvement) query: add Cython-aware serializer path in BoundStatement.bind()

When Cython serializers (from cassandra.serializers) are available and no
column encryption policy is active, BoundStatement.bind() now uses
pre-built Serializer objects cached on the PreparedStatement instead of
calling cqltype classmethods. This avoids per-value Python method dispatch
overhead and enables the ~30x vector serialization speedup from the Cython
serializers module.

The bind loop is split into three paths:
1. Column encryption policy path (unchanged behavior)
2. Cython serializers path (new fast path)
3. Plain Python path (no CE, no Cython -- removes per-value ColDesc/CE check)

Depends on PR scylladb#748 (Cython serializers module) and PR scylladb#630 (CE-policy
bind split).

Author: mykaul

cassandra/query.py

@@ -33,6 +33,12 @@
 from cassandra.protocol import _UNSET_VALUE
 from cassandra.util import OrderedDict, _sanitize_identifiers
 
+try:
+    from cassandra.serializers import make_serializers as _cython_make_serializers
+    _HAVE_CYTHON_SERIALIZERS = True
+except ImportError:
+    _HAVE_CYTHON_SERIALIZERS = False
+
 import logging
 log = logging.getLogger(__name__)
 
@@ -474,6 +480,30 @@ def __init__(self, column_metadata, query_id, routing_key_indexes, query,
         self.is_idempotent = False
         self._is_lwt = is_lwt
 
+    @property
+    def _serializers(self):
+        """Lazily create and cache Cython serializers for column types.
+
+        Returns a list of Serializer objects if Cython serializers are available
+        and there is no column encryption policy, otherwise returns None.
+
+        The column_encryption_policy check is performed on every access (not
+        cached) so that serializers are correctly bypassed if a policy is set
+        after construction.
+        """
+        if self.column_encryption_policy:
+            return None
+        try:
+            return self.__serializers
+        except AttributeError:
+            pass
+        if _HAVE_CYTHON_SERIALIZERS and self.column_metadata:
+            self.__serializers = _cython_make_serializers(
+                [col.type for col in self.column_metadata])
+        else:
+            self.__serializers = None
+        return self.__serializers
+
     @classmethod
     def from_message(cls, query_id, column_metadata, pk_indexes, cluster_metadata,
                      query, prepared_keyspace, protocol_version, result_metadata,
@@ -532,6 +562,14 @@ def __str__(self):
     __repr__ = __str__
 
 
+def _raise_bind_serialize_error(col_spec, value, exc):
+    """Wrap serialization errors with column context for all bind loop paths."""
+    actual_type = type(value)
+    message = ('Received an argument of invalid type for column "%s". '
+               'Expected: %s, Got: %s; (%s)' % (col_spec.name, col_spec.type, actual_type, exc))
+    raise TypeError(message)
+
+
 class BoundStatement(Statement):
     """
     A prepared statement that has been bound to a particular set of values.
@@ -636,28 +674,65 @@ def bind(self, values):
 
         self.raw_values = values
         self.values = []
-        for value, col_spec in zip(values, col_meta):
-            if value is None:
-                self.values.append(None)
-            elif value is UNSET_VALUE:
-                if proto_version >= 4:
-                    self._append_unset_value()
+        if ce_policy:
+            # Column encryption path: check each column for CE policy
+            for value, col_spec in zip(values, col_meta):
+                if value is None:
+                    self.values.append(None)
+                elif value is UNSET_VALUE:
+                    if proto_version >= 4:
+                        self._append_unset_value()
+                    else:
+                        raise ValueError("Attempt to bind UNSET_VALUE while using unsuitable protocol version (%d < 4)" % proto_version)
                 else:
-                    raise ValueError("Attempt to bind UNSET_VALUE while using unsuitable protocol version (%d < 4)" % proto_version)
+                    try:
+                        col_desc = ColDesc(col_spec.keyspace_name, col_spec.table_name, col_spec.name)
+                        uses_ce = ce_policy.contains_column(col_desc)
+                        if uses_ce:
+                            col_type = ce_policy.column_type(col_desc)
+                            col_bytes = col_type.serialize(value, proto_version)
+                            col_bytes = ce_policy.encrypt(col_desc, col_bytes)
+                        else:
+                            col_bytes = col_spec.type.serialize(value, proto_version)
+                        self.values.append(col_bytes)
+                    # OverflowError: Cython int32/float casts may raise on out-of-range values
+                    except (TypeError, struct.error, OverflowError) as exc:
+                        _raise_bind_serialize_error(col_spec, value, exc)
+        else:
+            # Fast path: no column encryption, use Cython serializers if available
+            serializers = self.prepared_statement._serializers
+            if serializers is not None:
+                for ser, value, col_spec in zip(serializers, values, col_meta):
+                    if value is None:
+                        self.values.append(None)
+                    elif value is UNSET_VALUE:
+                        if proto_version >= 4:
+                            self._append_unset_value()
+                        else:
+                            raise ValueError("Attempt to bind UNSET_VALUE while using unsuitable protocol version (%d < 4)" % proto_version)
+                    else:
+                        try:
+                            col_bytes = ser.serialize(value, proto_version)
+                            self.values.append(col_bytes)
+                        # OverflowError: Cython int32/float casts may raise on out-of-range values
+                        except (TypeError, struct.error, OverflowError) as exc:
+                            _raise_bind_serialize_error(col_spec, value, exc)
             else:
-                try:
-                    col_desc = ColDesc(col_spec.keyspace_name, col_spec.table_name, col_spec.name)
-                    uses_ce = ce_policy and ce_policy.contains_column(col_desc)
-                    col_type = ce_policy.column_type(col_desc) if uses_ce else col_spec.type
-                    col_bytes = col_type.serialize(value, proto_version)
-                    if uses_ce:
-                        col_bytes = ce_policy.encrypt(col_desc, col_bytes)
-                    self.values.append(col_bytes)
-                except (TypeError, struct.error) as exc:
-                    actual_type = type(value)
-                    message = ('Received an argument of invalid type for column "%s". '
-                               'Expected: %s, Got: %s; (%s)' % (col_spec.name, col_spec.type, actual_type, exc))
-                    raise TypeError(message)
+                for value, col_spec in zip(values, col_meta):
+                    if value is None:
+                        self.values.append(None)
+                    elif value is UNSET_VALUE:
+                        if proto_version >= 4:
+                            self._append_unset_value()
+                        else:
+                            raise ValueError("Attempt to bind UNSET_VALUE while using unsuitable protocol version (%d < 4)" % proto_version)
+                    else:
+                        try:
+                            col_bytes = col_spec.type.serialize(value, proto_version)
+                            self.values.append(col_bytes)
+                        # OverflowError: Cython int32/float casts may raise on out-of-range values
+                        except (TypeError, struct.error, OverflowError) as exc:
+                            _raise_bind_serialize_error(col_spec, value, exc)
 
         if proto_version >= 4:
             diff = col_meta_len - len(self.values)

tests/unit/test_parameter_binding.py

@@ -216,3 +216,123 @@ def test_unset_value(self):
 
 class BoundStatementTestV5(BoundStatementTestV4):
     protocol_version = 5
+
+
+class StubSerializer:
+    """Stub that mimics a Cython Serializer object for testing the fast path."""
+
+    def __init__(self, cqltype):
+        self.cqltype = cqltype
+
+    def serialize(self, value, protocol_version):
+        return self.cqltype.serialize(value, protocol_version)
+
+
+class OverflowSerializer:
+    """Stub that raises OverflowError, mimicking Cython <int32_t> cast overflow."""
+
+    def __init__(self, cqltype):
+        self.cqltype = cqltype
+
+    def serialize(self, value, protocol_version):
+        raise OverflowError('value too large to convert to int32_t')
+
+
+class CythonBindPathTest(unittest.TestCase):
+    """Tests for the Cython serializer fast path in BoundStatement.bind().
+
+    These tests inject stub serializers via the PreparedStatement's cached
+    __serializers attribute to exercise the Cython bind branch without
+    requiring compiled Cython.
+    """
+
+    protocol_version = 4
+
+    def _make_prepared(self, column_metadata, serializers=None):
+        """Create a PreparedStatement and inject serializers into its cache."""
+        prepared = PreparedStatement(column_metadata=column_metadata,
+                                     query_id=None,
+                                     routing_key_indexes=[],
+                                     query=None,
+                                     keyspace='keyspace',
+                                     protocol_version=self.protocol_version,
+                                     result_metadata=None,
+                                     result_metadata_id=None)
+        # Inject directly into the name-mangled cache attribute used by
+        # the _serializers property, bypassing the lazy initialization.
+        prepared._PreparedStatement__serializers = serializers
+        return prepared
+
+    def test_cython_path_normal_serialization(self):
+        """Cython fast path produces the same result as the plain Python path."""
+        column_metadata = [ColumnMetadata('keyspace', 'cf', 'c0', Int32Type),
+                           ColumnMetadata('keyspace', 'cf', 'c1', Int32Type)]
+        serializers = [StubSerializer(Int32Type), StubSerializer(Int32Type)]
+        prepared = self._make_prepared(column_metadata, serializers)
+
+        bound = BoundStatement(prepared_statement=prepared)
+        bound.bind((42, -1))
+        assert bound.values == [Int32Type.serialize(42, self.protocol_version),
+                                Int32Type.serialize(-1, self.protocol_version)]
+
+    def test_cython_path_none_value(self):
+        """None values pass through the Cython path without serialization."""
+        column_metadata = [ColumnMetadata('keyspace', 'cf', 'c0', Int32Type)]
+        serializers = [StubSerializer(Int32Type)]
+        prepared = self._make_prepared(column_metadata, serializers)
+
+        bound = BoundStatement(prepared_statement=prepared)
+        bound.bind((None,))
+        assert bound.values == [None]
+
+    def test_cython_path_unset_value(self):
+        """UNSET_VALUE is handled correctly in the Cython fast path (v4+)."""
+        column_metadata = [ColumnMetadata('keyspace', 'cf', 'c0', Int32Type),
+                           ColumnMetadata('keyspace', 'cf', 'c1', Int32Type)]
+        serializers = [StubSerializer(Int32Type), StubSerializer(Int32Type)]
+        prepared = self._make_prepared(column_metadata, serializers)
+
+        bound = BoundStatement(prepared_statement=prepared)
+        bound.bind((42, UNSET_VALUE))
+        assert bound.values[0] == Int32Type.serialize(42, self.protocol_version)
+        assert bound.values[1] == UNSET_VALUE
+
+    def test_cython_path_overflow_error_wrapped(self):
+        """OverflowError from Cython cast is caught and wrapped with column context."""
+        column_metadata = [ColumnMetadata('keyspace', 'cf', 'v0', Int32Type)]
+        serializers = [OverflowSerializer(Int32Type)]
+        prepared = self._make_prepared(column_metadata, serializers)
+
+        bound = BoundStatement(prepared_statement=prepared)
+        with pytest.raises(TypeError) as exc:
+            bound.bind((2**31,))
+        msg = str(exc.value)
+        assert 'v0' in msg
+        assert 'Int32Type' in msg
+        assert 'int' in msg
+
+    def test_cython_path_type_error_wrapped(self):
+        """TypeError from serializer is caught and wrapped with column context."""
+        column_metadata = [ColumnMetadata('keyspace', 'cf', 'v0', Int32Type)]
+        serializers = [StubSerializer(Int32Type)]
+        prepared = self._make_prepared(column_metadata, serializers)
+
+        bound = BoundStatement(prepared_statement=prepared)
+        with pytest.raises(TypeError) as exc:
+            bound.bind(('not_an_int',))
+        msg = str(exc.value)
+        assert 'v0' in msg
+        assert 'Int32Type' in msg
+
+    def test_plain_path_overflow_error_wrapped(self):
+        """OverflowError in the plain Python path is also caught and wrapped."""
+        column_metadata = [ColumnMetadata('keyspace', 'cf', 'v0', Int32Type)]
+        # Force the plain Python path (no Cython serializers)
+        prepared = self._make_prepared(column_metadata, serializers=None)
+
+        bound = BoundStatement(prepared_statement=prepared)
+        with pytest.raises(TypeError) as exc:
+            bound.bind((2**31,))
+        msg = str(exc.value)
+        assert 'v0' in msg
+        assert 'Int32Type' in msg