Skip to content

Commit 5c39fac

Browse files
committed
(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).
1 parent 41741c0 commit 5c39fac

2 files changed

Lines changed: 401 additions & 31 deletions

File tree

cassandra/query.py

Lines changed: 129 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,12 @@
3333
from cassandra.protocol import _UNSET_VALUE
3434
from cassandra.util import OrderedDict, _sanitize_identifiers
3535

36+
try:
37+
from cassandra.serializers import make_serializers as _cython_make_serializers
38+
_HAVE_CYTHON_SERIALIZERS = True
39+
except ImportError:
40+
_HAVE_CYTHON_SERIALIZERS = False
41+
3642
import logging
3743
log = logging.getLogger(__name__)
3844

@@ -522,6 +528,32 @@ def update_result_metadata(self, result_metadata, result_metadata_id):
522528
self._result_metadata_and_id = (result_metadata, result_metadata_id)
523529
self._warned_missing_column_metadata = False
524530

531+
@property
532+
def _serializers(self):
533+
"""Lazily create and cache Cython serializers for column types.
534+
535+
Returns a list of Serializer objects if Cython serializers are available
536+
and there is no column encryption policy, otherwise returns None.
537+
538+
The column_encryption_policy check is performed on every access (not
539+
cached) so that serializers are correctly bypassed if a policy is set
540+
after construction. This means the cache never goes stale: once a CE
541+
policy is present, we always return None and fall through to the
542+
encryption-aware bind path.
543+
"""
544+
if self.column_encryption_policy:
545+
return None
546+
try:
547+
return self._cached_serializers
548+
except AttributeError:
549+
pass
550+
if _HAVE_CYTHON_SERIALIZERS and self.column_metadata:
551+
self._cached_serializers = _cython_make_serializers(
552+
[col.type for col in self.column_metadata])
553+
else:
554+
self._cached_serializers = None
555+
return self._cached_serializers
556+
525557
@classmethod
526558
def from_message(cls, query_id, column_metadata, pk_indexes, cluster_metadata,
527559
query, prepared_keyspace, protocol_version, result_metadata,
@@ -580,6 +612,26 @@ def __str__(self):
580612
__repr__ = __str__
581613

582614

615+
def _raise_bind_serialize_error(col_spec, value, exc):
616+
"""Wrap TypeError, struct.error, or OverflowError with column context.
617+
618+
Called from all three bind loop paths (CE, Cython, plain Python) to
619+
provide a uniform error message that includes the column name and
620+
expected type. struct.error arises from int32 out-of-range values;
621+
OverflowError from float out-of-range values. Other exception types
622+
(e.g. ValueError from VectorType dimension mismatch) propagate
623+
without wrapping.
624+
"""
625+
actual_type = type(value)
626+
if isinstance(exc, (OverflowError, struct.error)):
627+
reason = 'value out of range'
628+
else:
629+
reason = 'invalid type'
630+
message = ('Received an argument with %s for column "%s". '
631+
'Expected: %s, Got: %s; (%s)' % (reason, col_spec.name, col_spec.type, actual_type, exc))
632+
raise TypeError(message) from exc
633+
634+
583635
class BoundStatement(Statement):
584636
"""
585637
A prepared statement that has been bound to a particular set of values.
@@ -683,44 +735,91 @@ def bind(self, values):
683735
(value_len, len(self.prepared_statement.routing_key_indexes)))
684736

685737
self.raw_values = values
686-
self.values = []
687-
for value, col_spec in zip(values, col_meta):
688-
if value is None:
689-
self.values.append(None)
690-
elif value is UNSET_VALUE:
691-
if proto_version >= 4:
692-
self._append_unset_value()
738+
# Pre-allocate to avoid repeated list growth reallocations
739+
self.values = [None] * col_meta_len
740+
idx = 0
741+
if ce_policy:
742+
# Column encryption path: check each column for CE policy
743+
for value, col_spec in zip(values, col_meta):
744+
if value is None:
745+
self.values[idx] = None
746+
elif value is UNSET_VALUE:
747+
if proto_version >= 4:
748+
idx = self._append_unset_value(idx)
749+
continue
750+
else:
751+
raise ValueError("Attempt to bind UNSET_VALUE while using unsuitable protocol version (%d < 4)" % proto_version)
693752
else:
694-
raise ValueError("Attempt to bind UNSET_VALUE while using unsuitable protocol version (%d < 4)" % proto_version)
753+
try:
754+
col_desc = ColDesc(col_spec.keyspace_name, col_spec.table_name, col_spec.name)
755+
uses_ce = ce_policy.contains_column(col_desc)
756+
if uses_ce:
757+
col_type = ce_policy.column_type(col_desc)
758+
col_bytes = col_type.serialize(value, proto_version)
759+
col_bytes = ce_policy.encrypt(col_desc, col_bytes)
760+
else:
761+
col_bytes = col_spec.type.serialize(value, proto_version)
762+
self.values[idx] = col_bytes
763+
# struct.error: int32 out-of-range; OverflowError: float out-of-range
764+
except (TypeError, struct.error, OverflowError) as exc:
765+
_raise_bind_serialize_error(col_spec, value, exc)
766+
idx += 1
767+
else:
768+
# Fast path: no column encryption, use Cython serializers if available
769+
serializers = self.prepared_statement._serializers
770+
if serializers is not None:
771+
for ser, value, col_spec in zip(serializers, values, col_meta):
772+
if value is None:
773+
self.values[idx] = None
774+
elif value is UNSET_VALUE:
775+
if proto_version >= 4:
776+
idx = self._append_unset_value(idx)
777+
continue
778+
else:
779+
raise ValueError("Attempt to bind UNSET_VALUE while using unsuitable protocol version (%d < 4)" % proto_version)
780+
else:
781+
try:
782+
col_bytes = ser.serialize(value, proto_version)
783+
self.values[idx] = col_bytes
784+
# struct.error: int32 out-of-range; OverflowError: float out-of-range
785+
except (TypeError, struct.error, OverflowError) as exc:
786+
_raise_bind_serialize_error(col_spec, value, exc)
787+
idx += 1
695788
else:
696-
try:
697-
col_desc = ColDesc(col_spec.keyspace_name, col_spec.table_name, col_spec.name)
698-
uses_ce = ce_policy and ce_policy.contains_column(col_desc)
699-
col_type = ce_policy.column_type(col_desc) if uses_ce else col_spec.type
700-
col_bytes = col_type.serialize(value, proto_version)
701-
if uses_ce:
702-
col_bytes = ce_policy.encrypt(col_desc, col_bytes)
703-
self.values.append(col_bytes)
704-
except (TypeError, struct.error) as exc:
705-
actual_type = type(value)
706-
message = ('Received an argument of invalid type for column "%s". '
707-
'Expected: %s, Got: %s; (%s)' % (col_spec.name, col_spec.type, actual_type, exc))
708-
raise TypeError(message)
789+
for value, col_spec in zip(values, col_meta):
790+
if value is None:
791+
self.values[idx] = None
792+
elif value is UNSET_VALUE:
793+
if proto_version >= 4:
794+
idx = self._append_unset_value(idx)
795+
continue
796+
else:
797+
raise ValueError("Attempt to bind UNSET_VALUE while using unsuitable protocol version (%d < 4)" % proto_version)
798+
else:
799+
try:
800+
col_bytes = col_spec.type.serialize(value, proto_version)
801+
self.values[idx] = col_bytes
802+
# struct.error: int32 out-of-range; OverflowError: float out-of-range
803+
except (TypeError, struct.error, OverflowError) as exc:
804+
_raise_bind_serialize_error(col_spec, value, exc)
805+
idx += 1
709806

710807
if proto_version >= 4:
711-
diff = col_meta_len - len(self.values)
712-
if diff:
713-
for _ in range(diff):
714-
self._append_unset_value()
808+
# Fill remaining unbound columns with UNSET_VALUE (v4+ feature).
809+
while idx < col_meta_len:
810+
idx = self._append_unset_value(idx)
811+
elif idx < col_meta_len:
812+
# Pre-v4: trim trailing unused slots (no UNSET_VALUE support)
813+
self.values = self.values[:idx]
715814

716815
return self
717816

718-
def _append_unset_value(self):
719-
next_index = len(self.values)
720-
if self.prepared_statement.is_routing_key_index(next_index):
721-
col_meta = self.prepared_statement.column_metadata[next_index]
817+
def _append_unset_value(self, idx):
818+
if self.prepared_statement.is_routing_key_index(idx):
819+
col_meta = self.prepared_statement.column_metadata[idx]
722820
raise ValueError("Cannot bind UNSET_VALUE as a part of the routing key '%s'" % col_meta.name)
723-
self.values.append(UNSET_VALUE)
821+
self.values[idx] = UNSET_VALUE
822+
return idx + 1
724823

725824
@property
726825
def routing_key(self):

0 commit comments

Comments
 (0)