(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,24 @@ 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.
+        """
+        try:
+            return self.__serializers
+        except AttributeError:
+            pass
+        if _HAVE_CYTHON_SERIALIZERS and not self.column_encryption_policy 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,
@@ -636,28 +660,71 @@ 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)
+                    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)
+        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)
+                        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)
             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)
+                        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)
 
         if proto_version >= 4:
             diff = col_meta_len - len(self.values)