diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 2a02f1ac54..5ad1c04e38 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -7,13 +7,35 @@ Features statements skip re-sending result metadata on EXECUTE, and the driver automatically refreshes cached metadata when the server detects a schema change (DRIVER-153) +Bug Fixes +--------- +* Harden prepared-result metadata caching against concurrent schema/client decoder + changes. Cached metadata, ids, and decoder provenance are published atomically; + UDT descriptors are isolated by registered Python class; UDT registration + invalidates cached ids before publishing mapping changes; and futures refresh + their request snapshot before fetching another page or after re-prepare. +* Restrict ``skip_meta`` to immutable snapshots of built-in protocol handlers. + A Session keeps using the immutable configuration it captured, while modified + built-in handlers and custom handlers cannot create an eligible snapshot and + continue to receive full result metadata. +* Stop retrying a prepared query when re-prepare returns a different query id; + the future now remains failed instead of updating metadata and issuing another + request. +* Restore an evicted prepared-statement cache entry through the Cluster's + lock-protected ``add_prepared()`` path during UNPREPARED recovery. + Others ------ * ``PreparedStatement.result_metadata`` and ``PreparedStatement.result_metadata_id`` are now read-only. They are replaced together by ``PreparedStatement.update_result_metadata()``, so a request can never observe a metadata id paired with result metadata from a different schema version. Code that assigned either - attribute directly must call ``update_result_metadata()`` instead. + attribute directly must call ``update_result_metadata()`` instead. The metadata getter + keeps its historical list shape but returns a defensive copy, so mutating it cannot + change the internal metadata/id pair. +* Client-initiated PREPARE responses now use + ``Session.client_protocol_handler``, matching the handler used for subsequent + EXECUTE responses. This is user-visible for custom protocol handlers. * Message serialization now receives the connection's negotiated ``ProtocolFeatures``: ``Connection.send_msg`` passes ``protocol_features`` to the encoder, and ``_ProtocolHandler.encode_message`` forwards it to each message's ``send_body``. @@ -24,8 +46,10 @@ Others forward it. There is deliberately no compatibility fallback: protocol extensions are negotiated per connection at STARTUP, so an encoder unaware of ``protocol_features`` could silently omit fields a negotiated extension requires. - This release emits no new bytes on the wire; the parameter is groundwork for - upcoming protocol extensions (``SCYLLA_USE_METADATA_ID``, ``TABLETS_ROUTING_V2``). + When ``SCYLLA_USE_METADATA_ID`` is negotiated, EXECUTE messages use this plumbing + to include the extension's metadata id and may request that result metadata be + omitted. Connections that do not negotiate the extension retain their existing + frame layout. 3.29.11 ======= diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 88c8d2707a..e186d64a97 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -31,6 +31,7 @@ import enum import json import logging +from types import MappingProxyType from typing import Any, Dict, Optional, Union, Tuple from warnings import warn from random import random @@ -69,7 +70,10 @@ BatchMessage, RESULT_KIND_PREPARED, RESULT_KIND_SET_KEYSPACE, RESULT_KIND_ROWS, RESULT_KIND_SCHEMA_CHANGE, ProtocolHandler, - RESULT_KIND_VOID, ProtocolException) + RESULT_KIND_VOID, ProtocolException, + _class_attribute, + _prepared_metadata_cache_config, + _prepared_metadata_result_attributes) from cassandra.metadata import Metadata, protect_name, murmur3, _NodeInfo from cassandra.policies import (TokenAwarePolicy, DCAwareRoundRobinPolicy, SimpleConvictionPolicy, ExponentialReconnectionPolicy, HostDistance, @@ -196,6 +200,225 @@ def _connection_reduce_fn(val,import_fn): _NOT_SET = object() +_PREPARED_METADATA_HANDLER_SNAPSHOTS = {} +_PREPARED_METADATA_HANDLER_SNAPSHOT_LOCK = Lock() + + +def _prepared_metadata_protocol_handler_snapshot(protocol_handler): + """ + Freeze the result-decoder configuration of a driver-owned handler. + + The public handler maps are intentionally customizable. Only the canonical + driver-owned configuration is eligible: application changes fall back to + full metadata because arbitrary decoder class state cannot be frozen + safely. An eligible prepared future receives a private snapshot so a later + source mutation cannot produce a hybrid old-metadata/new-decoder request. + + Custom or modified handlers are returned unchanged and remain ineligible + for skip-metadata. + """ + try: + handler_vars = vars(protocol_handler) + except TypeError: + log.debug( + "Cannot inspect protocol handler %r; prepared executions will " + "request full result metadata.", protocol_handler, + exc_info=True) + return protocol_handler + + if handler_vars.get('_prepared_metadata_handler_snapshot'): + return protocol_handler + + handler_token = handler_vars.get('_prepared_metadata_cache_token') + if handler_token is None: + return protocol_handler + + try: + message_types = dict(protocol_handler.message_types_by_opcode) + result_message_type = message_types[ResultMessage.opcode] + if not isinstance(result_message_type, type): + return protocol_handler + type_codes = dict(getattr(result_message_type, 'type_codes', {})) + cqltypes_by_code = dict( + getattr(result_message_type, '_cqltypes_by_code', {})) + decode_message = _class_attribute( + protocol_handler, 'decode_message') + result_attributes = \ + _prepared_metadata_result_attributes(result_message_type) + column_encryption_policy = getattr( + protocol_handler, 'column_encryption_policy', None) + handler_config = _prepared_metadata_cache_config( + protocol_handler, + result_message_type, + message_types, + type_codes, + cqltypes_by_code, + decode_message, + result_attributes, + column_encryption_policy, + ) + canonical_config = handler_vars.get( + '_prepared_metadata_cache_config') + if (canonical_config is None + or handler_config[:-1] != canonical_config[:-1] + or column_encryption_policy is not canonical_config[-1]): + return protocol_handler + cache_key = ( + protocol_handler, + handler_token, + handler_config[:-1], + id(column_encryption_policy), + ) + # A malformed in-place customization should fall back to full metadata, + # not make request construction fail or create an uncached snapshot + # whose identity never matches the PREPARE provenance. + hash(cache_key) + except Exception: + log.debug( + "Cannot validate protocol handler %r; prepared executions will " + "request full result metadata.", protocol_handler, + exc_info=True) + return protocol_handler + + with _PREPARED_METADATA_HANDLER_SNAPSHOT_LOCK: + try: + cached = _PREPARED_METADATA_HANDLER_SNAPSHOTS.get( + protocol_handler) + if cached is not None and cached[0] == cache_key: + return cached[1] + except Exception: + log.debug( + "Cannot compare the cached protocol handler snapshot for %r; " + "prepared executions will request full result metadata.", + protocol_handler, exc_info=True) + return protocol_handler + + try: + result_attrs = dict(result_attributes) + result_attrs.update({ + '__module__': result_message_type.__module__, + 'type_codes': MappingProxyType(type_codes), + '_cqltypes_by_code': MappingProxyType(cqltypes_by_code), + }) + result_snapshot = type( + '_%sPreparedMetadataSnapshot' % + result_message_type.__name__, + (result_message_type,), + result_attrs, + ) + + message_types[ResultMessage.opcode] = result_snapshot + handler_attrs = { + '__module__': protocol_handler.__module__, + 'message_types_by_opcode': MappingProxyType(message_types), + 'column_encryption_policy': column_encryption_policy, + '_prepared_metadata_cache_token': object(), + '_prepared_metadata_handler_snapshot': True, + } + if decode_message is not None: + handler_attrs['decode_message'] = decode_message + snapshot = type( + '%sPreparedMetadataSnapshot' % + protocol_handler.__name__, + (protocol_handler,), + handler_attrs, + ) + except Exception: + log.debug( + "Cannot freeze protocol handler %r; prepared executions will " + "request full result metadata.", protocol_handler, + exc_info=True) + return protocol_handler + # Keep only the latest source configuration in the shared cache. + # Futures retain any older snapshot they are already using. + _PREPARED_METADATA_HANDLER_SNAPSHOTS[protocol_handler] = ( + cache_key, snapshot) + return snapshot + + +def _prepared_metadata_decoder_context(protocol_handler, cluster_context): + """ + Return cache provenance for a frozen driver-owned protocol handler. + + Custom subclasses can mutate ``message_types_by_opcode`` and their result + type maps in place. Class identity alone therefore cannot prove that cached + column descriptors are still valid. Eligible handlers are copied into a + cached snapshot with read-only maps; inherited tokens deliberately do not + opt a custom subclass into skip-metadata. + """ + # ``None`` marks a client-side metadata transition. Responses decoded + # while the transition is in progress must never become eligible for + # skip-metadata, even if another execute observes the same marker. + if cluster_context is None: + return None + + try: + handler_vars = vars(protocol_handler) + except TypeError: + return None + if not handler_vars.get('_prepared_metadata_handler_snapshot'): + return None + + return ( + handler_vars['_prepared_metadata_cache_token'], + protocol_handler, + cluster_context, + ) + + +def _prepared_metadata_session_handler(session, protocol_handler): + """ + Return the Session's immutable snapshot for ``protocol_handler``. + + A Session keeps using the snapshot even if the source class is later + mutated in place. Assigning a different client protocol handler invalidates + the cache by identity and builds (or rejects) a new snapshot. + """ + cached = vars(session).get('_prepared_metadata_handler_cache') + if (cached is not None + and (cached[0] is protocol_handler + or cached[1] is protocol_handler)): + return cached[1] + + snapshot = _prepared_metadata_protocol_handler_snapshot(protocol_handler) + session._prepared_metadata_handler_cache = (protocol_handler, snapshot) + return snapshot + + +def _prepared_metadata_request_state( + prepared_statement, protocol_handler, cluster_context, + requested_metadata_id=_NOT_SET, decoder_context=_NOT_SET): + """ + Resolve one coherent metadata/id/provenance snapshot for an EXECUTE. + + ``requested_metadata_id`` is supplied by the compatibility path where an + extension constructs a ResponseFuture directly. A mismatch with the + PreparedStatement cache forces a self-describing response. + """ + if decoder_context is _NOT_SET: + decoder_context = _prepared_metadata_decoder_context( + protocol_handler, cluster_context) + + result_metadata, cached_metadata_id, cached_decoder_context = \ + prepared_statement._result_metadata_snapshot + result_metadata_id = ( + cached_metadata_id + if requested_metadata_id is _NOT_SET + else requested_metadata_id) + + if (result_metadata_id != cached_metadata_id + or (result_metadata + and (decoder_context is None + or cached_decoder_context != decoder_context))): + return (), None, False, decoder_context + + return ( + result_metadata, + result_metadata_id, + bool(result_metadata) and result_metadata_id is not None, + decoder_context, + ) + class NoHostAvailable(Exception): """ @@ -1208,6 +1431,7 @@ def token_metadata_enabled(self, enabled): _is_setup = False _prepared_statements = None _prepared_statement_lock = None + _prepared_metadata_context = None _idle_heartbeat = None _protocol_version_explicit = False _discount_down_events = True @@ -1549,6 +1773,10 @@ def __init__(self, self.control_connection = None self._prepared_statements = WeakValueDictionary() self._prepared_statement_lock = Lock() + # Opaque and instance-unique: cached decoder metadata must not be + # reused by a Session from another Cluster merely because both have + # registered the same number of UDT mappings. + self._prepared_metadata_context = object() self._user_types = defaultdict(dict) @@ -1636,6 +1864,21 @@ def _create_thread_pool_executor(self, **kwargs): return tpe_class(**kwargs) + def _invalidate_prepared_metadata_context(self, eligible=True): + """ + Rotate this Cluster's decoder context and invalidate cached server ids. + + The token also covers statements not currently present in the weak + prepared cache: their old context will no longer match on execute. + An ineligible context prevents metadata decoded during a client-side + transition from enabling skip-metadata before the transition finishes. + """ + with self._prepared_statement_lock: + self._prepared_metadata_context = object() if eligible else None + prepared_statements = tuple(self._prepared_statements.values()) + for prepared_statement in prepared_statements: + prepared_statement._invalidate_result_metadata_id() + def register_user_type(self, keyspace, user_type, klass): """ Registers a class to use to represent a particular user-defined type. @@ -1689,10 +1932,23 @@ def __init__(self, street, zipcode): "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) + # Enter a transition context before publishing any mapping changes. + # Executes created while Session callbacks run must request full + # metadata rather than retain the old id and decoder descriptors. + self._invalidate_prepared_metadata_context(eligible=False) self._user_types[keyspace][user_type] = klass - for session in tuple(self.sessions): - session.user_type_registered(keyspace, user_type, klass) - UserType.evict_udt_class(keyspace, user_type) + try: + for session in tuple(self.sessions): + session.user_type_registered(keyspace, user_type, klass) + finally: + UserType.evict_udt_class(keyspace, user_type) + + # Publish a distinct stable context after the new mapping and UDT + # descriptor cache are ready. Responses decoded during the + # transition carry the intermediate token and cannot be reused + # afterward. Do this even if a Session rejects registration: the + # shared connection map was already changed above. + self._invalidate_prepared_metadata_context() def add_execution_profile(self, name, profile, pool_wait_timeout=5): """ @@ -2666,6 +2922,7 @@ def default_serial_consistency_level(self, cl): _metrics = None _request_init_callbacks = None _graph_paging_available = False + _prepared_metadata_handler_cache = None def __init__(self, cluster, hosts, keyspace=None): self.cluster = cluster @@ -2677,6 +2934,7 @@ def __init__(self, cluster, hosts, keyspace=None): self._profile_manager = cluster.profile_manager self._metrics = cluster.metrics self._request_init_callbacks = [] + self._prepared_metadata_handler_cache = None self._protocol_version = self.cluster.protocol_version self.encoder = Encoder() @@ -2821,7 +3079,6 @@ def execute_async(self, query, parameters=None, trace=False, custom_payload=None future = self._create_response_future( query, parameters, trace, custom_payload, timeout, execution_profile, paging_state, host) - future._protocol_handler = self.client_protocol_handler self._on_request(future) future.send_request() return future @@ -2887,7 +3144,6 @@ def execute_graph_async(self, query, parameters=None, trace=False, execution_pro timeout=_NOT_SET, execution_profile=execution_profile) future.message.query_params = graph_parameters - future._protocol_handler = self.client_protocol_handler if execution_profile.graph_options.is_analytics_source and \ isinstance(execution_profile.load_balancing_policy, DefaultLoadBalancingPolicy): @@ -2970,9 +3226,12 @@ def _transform_params(self, parameters, graph_options): def _target_analytics_master(self, future): future._start_timer() + # This internal query deliberately retains the default handler; only + # client-initiated requests use Session.client_protocol_handler. master_query_future = self._create_response_future("CALL DseClientTool.getAnalyticsGraphServer()", parameters=None, trace=False, - custom_payload=None, timeout=future.timeout) + custom_payload=None, timeout=future.timeout, + protocol_handler=ProtocolHandler) master_query_future.row_factory = tuple_factory master_query_future.send_request() @@ -2997,10 +3256,17 @@ def _on_analytics_master_result(self, response, master_future, query_future): def _create_response_future(self, query, parameters, trace, custom_payload, timeout, execution_profile=EXEC_PROFILE_DEFAULT, - paging_state=None, host=None): + paging_state=None, host=None, + protocol_handler=_NOT_SET): """ Returns the ResponseFuture before calling send_request() on it """ prepared_statement = None + # A request's decoder and any cached metadata it consumes must belong to + # one handler snapshot. Reading this attribute again after construction + # could pair metadata selected for one handler with another handler's + # decode_message when applications change it concurrently. + if protocol_handler is _NOT_SET: + protocol_handler = self.client_protocol_handler if isinstance(query, str): query = SimpleStatement(query) @@ -3050,6 +3316,7 @@ def _create_response_future(self, query, parameters, trace, custom_payload, # Snapshot passed to the ResponseFuture for decoding skip_meta responses; only # bound statements carry cached result metadata (set in the BoundStatement branch). bound_result_metadata = _NOT_SET + result_metadata_decoder_context = _NOT_SET if isinstance(query, SimpleStatement): query_string = query.query_string @@ -3062,6 +3329,8 @@ def _create_response_future(self, query, parameters, trace, custom_payload, continuous_paging_options, statement_keyspace) elif isinstance(query, BoundStatement): prepared_statement = query.prepared_statement + protocol_handler = _prepared_metadata_session_handler( + self, protocol_handler) # Snapshot metadata and its id as one atomic pair so the message never # carries the id of one schema version alongside a skip_meta decision # made for another. skip_meta is requested only when there is both an @@ -3070,17 +3339,23 @@ def _create_response_future(self, query, parameters, trace, custom_payload, # metadata-less response with, so the server must send it. # Whether skip_meta and the id actually reach the wire is decided per # connection at serialization time (see ExecuteMessage.send_body). - # Continuous paging sessions are excluded: Connection.process_msg hardcodes - # result_metadata=None for every page after the first (it isn't threaded - # through the paging session), so a skip_meta response has nothing to - # decode page 2+ against. - result_metadata, result_metadata_id = prepared_statement.result_metadata_and_id - bound_result_metadata = result_metadata + # Continuous paging sessions are excluded: + # Connection.process_msg hardcodes result_metadata=None for every + # page after the first (it is not threaded through the paging + # session), so a skip_meta response has nothing to decode page 2+ + # against. + (bound_result_metadata, + result_metadata_id, + skip_meta, + result_metadata_decoder_context) = \ + _prepared_metadata_request_state( + prepared_statement, + protocol_handler, + self.cluster._prepared_metadata_context) message = ExecuteMessage( prepared_statement.query_id, query.values, cl, serial_cl, fetch_size, paging_state, timestamp, - skip_meta=bool(result_metadata) and result_metadata_id is not None - and continuous_paging_options is None, + skip_meta=skip_meta and continuous_paging_options is None, continuous_paging_options=continuous_paging_options, result_metadata_id=result_metadata_id) elif isinstance(query, BatchStatement): @@ -3105,11 +3380,16 @@ def _create_response_future(self, query, parameters, trace, custom_payload, message.allow_beta_protocol_version = self.cluster.allow_beta_protocol_version spec_exec_plan = spec_exec_policy.new_plan(query.keyspace or self.keyspace, query) if query.is_idempotent and spec_exec_policy else None - return ResponseFuture( + future = ResponseFuture( self, message, query, timeout, metrics=self._metrics, prepared_statement=prepared_statement, retry_policy=retry_policy, row_factory=row_factory, load_balancer=load_balancing_policy, start_time=start_time, speculative_execution_plan=spec_exec_plan, - continuous_paging_state=None, host=host, bound_result_metadata=bound_result_metadata) + continuous_paging_state=None, host=host, bound_result_metadata=bound_result_metadata, + result_metadata_decoder_context=result_metadata_decoder_context, + protocol_handler=protocol_handler, + prepared_metadata_skip_allowed=( + continuous_paging_options is None)) + return future def get_execution_profile(self, name): """ @@ -3216,7 +3496,19 @@ def prepare(self, query, custom_payload=None, keyspace=None): message. See :ref:`custom_payload`. """ message = PrepareMessage(query=query, keyspace=keyspace) - future = ResponseFuture(self, message, query=None, timeout=self.default_timeout) + # PREPARE is client-initiated too: its bind and result type descriptors + # must be produced by the same handler snapshot that will decode + # EXECUTE. + protocol_handler = _prepared_metadata_session_handler( + self, self.client_protocol_handler) + result_metadata_decoder_context = \ + _prepared_metadata_decoder_context( + protocol_handler, + self.cluster._prepared_metadata_context) + future = ResponseFuture( + self, message, query=None, timeout=self.default_timeout, + result_metadata_decoder_context=result_metadata_decoder_context, + protocol_handler=protocol_handler) try: future.send_request() response = future.result().one() @@ -3227,7 +3519,8 @@ 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) + self._protocol_version, response.column_metadata, response.result_metadata_id, response.is_lwt, + self.cluster.column_encryption_policy, result_metadata_decoder_context) prepared_statement.custom_payload = future.custom_payload self.cluster.add_prepared(response.query_id, prepared_statement) @@ -4737,14 +5030,28 @@ class ResponseFuture(object): _control_connection_query_attempted = False _TABLET_ROUTING_CTYPE = None _bound_result_metadata = None + _result_metadata_decoder_context = None _warned_timeout = False def __init__(self, session, message, query, timeout, metrics=None, prepared_statement=None, retry_policy=RetryPolicy(), row_factory=None, load_balancer=None, start_time=None, speculative_execution_plan=None, continuous_paging_state=None, host=None, - bound_result_metadata=_NOT_SET): + bound_result_metadata=_NOT_SET, + result_metadata_decoder_context=_NOT_SET, + protocol_handler=_NOT_SET, + prepared_metadata_skip_allowed=True): self.session = session + protocol_handler = ( + session.client_protocol_handler + if protocol_handler is _NOT_SET else protocol_handler) + if (prepared_statement is not None + and isinstance(message, ExecuteMessage)): + protocol_handler = _prepared_metadata_session_handler( + session, protocol_handler) + self._protocol_handler = protocol_handler + self._prepared_metadata_skip_allowed = \ + prepared_metadata_skip_allowed # TODO: normalize handling of retry policy and row factory self.row_factory = row_factory or session.row_factory self._load_balancer = load_balancer or session.cluster._default_load_balancing_policy @@ -4754,12 +5061,68 @@ def __init__(self, session, message, query, timeout, metrics=None, prepared_stat self._retry_policy = retry_policy self._metrics = metrics self.prepared_statement = prepared_statement - # Metadata snapshotted alongside the message's result_metadata_id at construction - # time (see Session._create_response_future). Decoding a skip_meta response uses - # this so the metadata decoded-with always pairs with the id the message sent, - # even if a concurrent METADATA_CHANGED replaces the prepared statement's cache in - # between. Defaults to [] for unprepared statements (no cached metadata). - self._bound_result_metadata = [] if bound_result_metadata is _NOT_SET else bound_result_metadata + # Session._create_response_future passes an already resolved snapshot. + # Extension code constructing an Execute ResponseFuture directly may + # omit it; resolve the same state here without mutating the caller's + # message, which could already be shared by speculative attempts. + if (bound_result_metadata is _NOT_SET + and prepared_statement is not None + and isinstance(message, ExecuteMessage)): + (bound_result_metadata, + result_metadata_id, + skip_meta, + result_metadata_decoder_context) = \ + _prepared_metadata_request_state( + prepared_statement, + self._protocol_handler, + session.cluster._prepared_metadata_context, + requested_metadata_id=message.result_metadata_id, + decoder_context=result_metadata_decoder_context) + message = copy(message) + message.result_metadata_id = result_metadata_id + message.skip_meta = ( + skip_meta and self._prepared_metadata_skip_allowed) + self.message = message + elif (prepared_statement is not None + and isinstance(message, ExecuteMessage)): + decoder_context = _prepared_metadata_decoder_context( + self._protocol_handler, + session.cluster._prepared_metadata_context) + if not bound_result_metadata: + if message.skip_meta: + message = copy(message) + message.skip_meta = False + self.message = message + result_metadata_decoder_context = decoder_context + elif (decoder_context is None + or result_metadata_decoder_context != decoder_context): + # An explicit snapshot with unknown or stale provenance cannot + # safely decode a NO_METADATA response. Keep the caller's + # message immutable in case another attempt already shares it. + bound_result_metadata = () + message = copy(message) + message.skip_meta = False + message.result_metadata_id = None + self.message = message + result_metadata_decoder_context = decoder_context + elif (message.skip_meta + and not self._prepared_metadata_skip_allowed): + message = copy(message) + message.skip_meta = False + self.message = message + elif bound_result_metadata is _NOT_SET: + bound_result_metadata = [] + self._bound_result_metadata = bound_result_metadata + self._result_metadata_decoder_context = ( + None if result_metadata_decoder_context is _NOT_SET + else result_metadata_decoder_context) + # Requests may be sent concurrently by speculative execution. Publish + # the message and the metadata handed to its decoder as one immutable + # snapshot so refreshes for paging/reprepare cannot tear the pair. + self._request_snapshot = ( + self.message, + self._bound_result_metadata, + self._result_metadata_decoder_context) self._callback_lock = Lock() self._start_time = start_time or time.time() self._host = host @@ -4965,7 +5328,10 @@ def _query_control_connection(self, message=None, cb=None, connection=None, host self._control_connection_query_attempted = True if message is None: - message = self.message + message, result_meta, decoder_context = self._request_snapshot + else: + result_meta = self._bound_result_metadata + decoder_context = self._result_metadata_decoder_context if connection is None: control_connection = self.session.cluster.control_connection @@ -4983,9 +5349,10 @@ def _query_control_connection(self, message=None, cb=None, connection=None, host try: request_id = self._borrow_control_connection(connection) self._connection = connection - result_meta = self._bound_result_metadata if cb is None: - cb = partial(self._set_result, host, connection, None) + cb = partial( + self._set_result, host, connection, None, + result_metadata_decoder_context=decoder_context) cb = partial(self._handle_control_connection_response, connection, cb) log.debug("No usable node pools; falling back to control connection for host %s", host) @@ -5015,7 +5382,10 @@ def _query_control_connection(self, message=None, cb=None, connection=None, host def _query(self, host, message=None, cb=None): if message is None: - message = self.message + message, result_meta, decoder_context = self._request_snapshot + else: + result_meta = self._bound_result_metadata + decoder_context = self._result_metadata_decoder_context self._control_connection_query_attempted = False @@ -5037,10 +5407,10 @@ def _query(self, host, message=None, cb=None): else: connection, request_id = pool.borrow_connection(timeout=2.0) self._connection = connection - result_meta = self._bound_result_metadata - if cb is None: - cb = partial(self._set_result, host, connection, pool) + cb = partial( + self._set_result, host, connection, pool, + result_metadata_decoder_context=decoder_context) self.request_encoded_size = connection.send_msg(message, request_id, cb=cb, encoder=self._protocol_handler.encode_message, @@ -5124,8 +5494,10 @@ def start_fetching_next_page(self): if not self._paging_state: raise QueryExhausted() + if not self._refresh_prepared_result_metadata( + paging_state=self._paging_state): + self.message.paging_state = self._paging_state self._make_query_plan() - self.message.paging_state = self._paging_state self._event.clear() self._final_result = _NOT_SET self._final_exception = None @@ -5133,8 +5505,49 @@ def start_fetching_next_page(self): self._start_timer() self.send_request() + def _refresh_prepared_result_metadata(self, paging_state=_NOT_SET): + """ + Refresh this future's metadata/id snapshot before reusing its EXECUTE. + + A future is reused for manual paging and after UNPREPARED recovery. In + both cases the shared PreparedStatement may have acquired fresher + metadata since the original message was built. + """ + if (self.prepared_statement is None + or not isinstance(self.message, ExecuteMessage)): + return False + + (result_metadata, + result_metadata_id, + skip_meta, + decoder_context) = _prepared_metadata_request_state( + self.prepared_statement, + self._protocol_handler, + self.session.cluster._prepared_metadata_context) + + message = copy(self.message) + message.result_metadata_id = result_metadata_id + message.skip_meta = ( + skip_meta and self._prepared_metadata_skip_allowed) + if paging_state is not _NOT_SET: + message.paging_state = paging_state + + self.message = message + self._bound_result_metadata = result_metadata + self._result_metadata_decoder_context = decoder_context + self._request_snapshot = ( + message, result_metadata, decoder_context) + return True + def _reprepare(self, prepare_message, host, connection, pool): - cb = partial(self.session.submit, self._execute_after_prepare, host, connection, pool) + decoder_context = _prepared_metadata_decoder_context( + self._protocol_handler, + self.session.cluster._prepared_metadata_context) + cb = partial( + self.session.submit, + self._execute_after_prepare, + host, connection, pool, + result_metadata_decoder_context=decoder_context) if pool is None and connection is not None and connection.is_control_connection: request_id = self._query_control_connection(prepare_message, cb=cb, connection=connection, host=host) @@ -5144,8 +5557,12 @@ def _reprepare(self, prepare_message, host, connection, pool): # try to submit the original prepared statement on some other host self.send_request() - def _set_result(self, host, connection, pool, response): + def _set_result(self, host, connection, pool, response, + result_metadata_decoder_context=_NOT_SET): try: + if result_metadata_decoder_context is _NOT_SET: + result_metadata_decoder_context = \ + self._result_metadata_decoder_context self.coordinator_host = host if pool and not pool.is_shutdown: pool.return_connection(connection) @@ -5211,8 +5628,10 @@ def _set_result(self, host, connection, pool, response): # skip sending metadata and rows would be decoded # against stale columns, with no recovery). # (this also re-arms the anomaly warning below) - self.prepared_statement.update_result_metadata( - response.column_metadata, new_result_metadata_id) + self.prepared_statement._update_result_metadata( + response.column_metadata, + new_result_metadata_id, + result_metadata_decoder_context) elif not self.prepared_statement._warned_missing_column_metadata: # Anomalous response: a new id without the metadata it # describes. Cache neither — adopting the id alone would @@ -5284,7 +5703,8 @@ def _set_result(self, host, connection, pool, response): return else: prepared_statement = self.prepared_statement - self.session.cluster._prepared_statements[query_id] = prepared_statement + self.session.cluster.add_prepared( + query_id, prepared_statement) current_keyspace = self._connection.keyspace prepared_keyspace = prepared_statement.keyspace @@ -5356,7 +5776,9 @@ def _set_keyspace_completed(self, errors): self._set_final_exception(ConnectionException( "Failed to set keyspace on all hosts: %s" % (errors,))) - def _execute_after_prepare(self, host, connection, pool, response): + def _execute_after_prepare( + self, host, connection, pool, response, + result_metadata_decoder_context=_NOT_SET): """ Handle the response to our attempt to prepare a statement. If it succeeded, run the original query again against the same host. @@ -5366,6 +5788,9 @@ def _execute_after_prepare(self, host, connection, pool, response): if self._final_exception: return + if result_metadata_decoder_context is _NOT_SET: + result_metadata_decoder_context = \ + self._result_metadata_decoder_context if isinstance(response, ResultMessage): if response.kind == RESULT_KIND_PREPARED: @@ -5379,6 +5804,7 @@ def _execute_after_prepare(self, host, connection, pool, response): expected=hexlify(self.prepared_statement.query_id), got=hexlify(response.query_id) ) )) + return # Update the metadata/id pair atomically from exactly what this # reprepare response carries. Falling back to the previously # cached id when this response has none would risk pairing it @@ -5388,9 +5814,12 @@ def _execute_after_prepare(self, host, connection, pool, response): # id-aware execute could send without the server detecting the # mismatch. Dropping it instead triggers the same self-healing # b'' sentinel path a never-prepared id would. - self.prepared_statement.update_result_metadata( - response.column_metadata, response.result_metadata_id) - + self.prepared_statement._update_result_metadata( + response.column_metadata, + response.result_metadata_id, + result_metadata_decoder_context) + self._refresh_prepared_result_metadata() + # use self._query to re-use the same host and # at the same time properly borrow the connection if pool is None and connection is not None and connection.is_control_connection: diff --git a/cassandra/cqltypes.py b/cassandra/cqltypes.py index 99018eef03..6f3ad10c9b 100644 --- a/cassandra/cqltypes.py +++ b/cassandra/cqltypes.py @@ -41,6 +41,7 @@ import time import struct import sys +from threading import Lock from uuid import UUID from cassandra.marshal import (int8_pack, int8_unpack, int16_pack, int16_unpack, @@ -977,36 +978,66 @@ class UserType(TupleType): typename = "org.apache.cassandra.db.marshal.UserType" _cache = {} + _cache_lock = Lock() _module = sys.modules[__name__] @classmethod - def make_udt_class(cls, keyspace, udt_name, field_names, field_types): + def make_udt_class(cls, keyspace, udt_name, field_names, field_types, + mapped_class=None): assert len(field_names) == len(field_types) - instance = cls._cache.get((keyspace, udt_name)) - if not instance or instance.fieldnames != field_names or instance.subtypes != field_types: - instance = type(udt_name, (cls,), {'subtypes': field_types, - 'cassname': cls.cassname, - 'typename': udt_name, - 'fieldnames': field_names, - 'keyspace': keyspace, - 'mapped_class': None, - 'tuple_type': cls._make_registered_udt_namedtuple(keyspace, udt_name, field_names)}) - cls._cache[(keyspace, udt_name)] = instance + # A UDT type descriptor can outlive the response that created it (for + # example, in a PreparedStatement's cached result metadata). Keep a + # distinct descriptor for each registered Python mapping so parsing + # metadata for another Cluster cannot mutate an existing descriptor. + # + # Use identity rather than the class itself in the key. Registered + # classes normally are hashable, but a custom metaclass may opt out of + # hashing. The generated descriptor holds a strong reference to + # mapped_class, so its id cannot be reused while this cache entry exists. + cache_key = (keyspace, udt_name, id(mapped_class)) + with cls._cache_lock: + instance = cls._cache.get(cache_key) + if not instance or instance.fieldnames != field_names or instance.subtypes != field_types: + instance = type(udt_name, (cls,), {'subtypes': field_types, + 'cassname': cls.cassname, + 'typename': udt_name, + 'fieldnames': field_names, + 'keyspace': keyspace, + 'mapped_class': mapped_class, + # Mapped values deserialize + # directly into mapped_class. + # Only unmapped variants need + # a module-registered tuple + # type for pickle support. + 'tuple_type': ( + None if mapped_class is not None + else cls._make_registered_udt_namedtuple( + keyspace, udt_name, field_names))}) + cls._cache[cache_key] = instance return instance @classmethod def evict_udt_class(cls, keyspace, udt_name): - try: - del cls._cache[(keyspace, udt_name)] - except KeyError: - pass + # Registration changes must evict mapped and unmapped variants alike. + # Slicing also recognizes two-element keys left by older driver code in + # a long-running process that reloads this module. + with cls._cache_lock: + cache_keys = [ + cache_key for cache_key in cls._cache + if cache_key[:2] == (keyspace, udt_name) + ] + for cache_key in cache_keys: + del cls._cache[cache_key] @classmethod def apply_parameters(cls, subtypes, names): keyspace = subtypes[0].cass_parameterized_type() # when parsed from cassandra type, the keyspace is created as an unrecognized cass type; This gets the name back udt_name = _name_from_hex_string(subtypes[1].cassname) field_names = tuple(_name_from_hex_string(encoded_name) for encoded_name in names[2:]) # using tuple here to match what comes into make_udt_class from other sources (for caching equality test) + # Cassandra type strings carry no per-Cluster user-type map. Keep this + # legacy lookup explicitly unmapped instead of borrowing a descriptor + # registered for some other Cluster. return cls.make_udt_class(keyspace, udt_name, field_names, tuple(subtypes[2:])) @classmethod diff --git a/cassandra/protocol.py b/cassandra/protocol.py index 9dfdbf3022..e60fa6da62 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -730,6 +730,7 @@ class ResultMessage(_MessageType): new_keyspace = None column_metadata = None query_id = None + result_metadata_id = None bind_metadata = None pk_indexes = None schema_change_event = None @@ -891,9 +892,9 @@ def read_type(cls, f, user_type_map): num_fields = read_short(f) names, types = zip(*((read_string(f), cls.read_type(f, user_type_map)) for _ in range(num_fields))) - specialized_type = typeclass.make_udt_class(ks, udt_name, names, types) - specialized_type.mapped_class = user_type_map.get(ks, {}).get(udt_name) - typeclass = specialized_type + mapped_class = user_type_map.get(ks, {}).get(udt_name) + typeclass = typeclass.make_udt_class( + ks, udt_name, names, types, mapped_class=mapped_class) elif typeclass == CUSTOM_TYPE: classname = read_string(f) typeclass = lookup_casstype(classname) @@ -1100,6 +1101,95 @@ def send_body(self, f, protocol_version, protocol_features=None): "Continuous paging backpressure is not supported.") +def _class_attribute(cls, name): + for base in cls.__mro__: + if name in vars(base): + return vars(base)[name] + return None + + +_PREPARED_METADATA_RESULT_ATTRIBUTE_EXCLUSIONS = frozenset(( + '__dict__', + '__doc__', + '__module__', + '__weakref__', + 'type_codes', + '_cqltypes_by_code', + # Retained by FastResultMessage but unused by the Python and Cython + # decoders. It therefore is not part of prepared-metadata provenance. + 'code_to_type', +)) + + +def _prepared_metadata_result_attributes(result_message_type): + """ + Capture every effective result-message attribute except decoder type maps. + + Deriving this from the MRO avoids a hand-maintained allowlist silently + falling behind when ResultMessage gains another decoding attribute. + """ + attributes = {} + for base in reversed(result_message_type.__mro__[:-1]): + for name, value in vars(base).items(): + if name not in _PREPARED_METADATA_RESULT_ATTRIBUTE_EXCLUSIONS: + attributes[name] = value + return tuple(attributes.items()) + + +_PREPARED_METADATA_CONFIG_UNSET = object() + + +def _prepared_metadata_cache_config( + handler, result_message_type, + message_types=_PREPARED_METADATA_CONFIG_UNSET, + type_codes=_PREPARED_METADATA_CONFIG_UNSET, + cqltypes_by_code=_PREPARED_METADATA_CONFIG_UNSET, + decode_message=_PREPARED_METADATA_CONFIG_UNSET, + result_attributes=_PREPARED_METADATA_CONFIG_UNSET, + column_encryption_policy=_PREPARED_METADATA_CONFIG_UNSET): + """ + Capture the canonical driver-owned result-decoder configuration. + + Cluster-side snapshots compare against this exact configuration. Installing + an application ResultMessage or mutating a decoding attribute therefore + makes the handler ineligible for skip-metadata instead of attempting to + freeze arbitrary application class state. + """ + message_types = ( + handler.message_types_by_opcode + if message_types is _PREPARED_METADATA_CONFIG_UNSET + else message_types) + type_codes = ( + result_message_type.type_codes + if type_codes is _PREPARED_METADATA_CONFIG_UNSET + else type_codes) + cqltypes_by_code = ( + result_message_type._cqltypes_by_code + if cqltypes_by_code is _PREPARED_METADATA_CONFIG_UNSET + else cqltypes_by_code) + decode_message = ( + _class_attribute(handler, 'decode_message') + if decode_message is _PREPARED_METADATA_CONFIG_UNSET + else decode_message) + result_attributes = ( + _prepared_metadata_result_attributes(result_message_type) + if result_attributes is _PREPARED_METADATA_CONFIG_UNSET + else result_attributes) + column_encryption_policy = ( + handler.column_encryption_policy + if column_encryption_policy is _PREPARED_METADATA_CONFIG_UNSET + else column_encryption_policy) + return ( + tuple(message_types.items()), + result_message_type, + tuple(type_codes.items()), + tuple(cqltypes_by_code.items()), + decode_message, + result_attributes, + column_encryption_policy, + ) + + class _ProtocolHandler(object): """ _ProtocolHander handles encoding and decoding messages. @@ -1121,6 +1211,11 @@ class _ProtocolHandler(object): column_encryption_policy = None """Instance of :class:`cassandra.policies.ColumnEncryptionPolicy` in use by this handler""" + # Marks handlers whose result-decoder configuration the driver may copy + # into an immutable per-request snapshot. Subclasses do not inherit + # eligibility because their opcode and type maps are application-owned. + _prepared_metadata_cache_token = object() + @classmethod def encode_message(cls, msg, stream_id, protocol_version, compressor, allow_beta_protocol_version, protocol_features): @@ -1245,6 +1340,10 @@ def decode_message(cls, protocol_version, protocol_features, user_type_map, stre return msg +_ProtocolHandler._prepared_metadata_cache_config = \ + _prepared_metadata_cache_config(_ProtocolHandler, ResultMessage) + + def cython_protocol_handler(colparser): """ Given a column parser to deserialize ResultMessages, return a suitable @@ -1284,7 +1383,11 @@ class CythonProtocolHandler(_ProtocolHandler): message_types_by_opcode = my_opcodes col_parser = colparser + _prepared_metadata_cache_token = object() + CythonProtocolHandler._prepared_metadata_cache_config = \ + _prepared_metadata_cache_config( + CythonProtocolHandler, FastResultMessage) return CythonProtocolHandler diff --git a/cassandra/query.py b/cassandra/query.py index 39b9fdb0ad..9dd3eaddf7 100644 --- a/cassandra/query.py +++ b/cassandra/query.py @@ -426,15 +426,13 @@ class PreparedStatement(object): A :class:`.PreparedStatement` should be prepared only once. Re-preparing a statement may affect performance (as the operation requires a network roundtrip). - |prepared_stmt_head|: Do not use ``*`` in prepared statements if you might - change the schema of the table being queried. The driver and server each - maintain a map between metadata for a schema and statements that were - prepared against that schema. When a user changes a schema, e.g. by adding - or removing a column, the server invalidates its mappings involving that - schema. However, there is currently no way to propagate that invalidation - to drivers. Thus, after a schema change, the driver will incorrectly - interpret the results of ``SELECT *`` queries prepared before the schema - change. This is currently being addressed in `CASSANDRA-10786 + |prepared_stmt_head|: Take care when using ``*`` in a prepared statement if + the queried table's schema may change. On connections that negotiate + ``SCYLLA_USE_METADATA_ID``, the server reports a result-metadata change and + the driver refreshes its cached column definitions before decoding rows. + Other connections do not use this driver's skip-metadata optimization and + continue to receive column definitions with each result. The general + prepared-metadata invalidation problem is tracked by `CASSANDRA-10786 `_. .. |prepared_stmt_head| raw:: html @@ -451,7 +449,10 @@ class PreparedStatement(object): protocol_version = None query_id = None query_string = None - _result_metadata_and_id = (None, None) + # Cached result metadata, its server-provided id, and the decoder context + # that produced the type descriptors. Keep these in one tuple: execute paths + # may read the state while response callbacks replace it. + _result_metadata_state = (None, None, None) column_encryption_policy = None routing_key_indexes = None _routing_key_index_set = None @@ -464,39 +465,78 @@ class PreparedStatement(object): def __init__(self, column_metadata, query_id, routing_key_indexes, query, keyspace, protocol_version, result_metadata, result_metadata_id, - is_lwt=False, column_encryption_policy=None): + is_lwt=False, column_encryption_policy=None, + result_metadata_decoder_context=None): self.column_metadata = column_metadata self.query_id = query_id self.routing_key_indexes = routing_key_indexes self.query_string = query self.keyspace = keyspace self.protocol_version = protocol_version - self._result_metadata_and_id = (result_metadata, result_metadata_id) + self._result_metadata_state = ( + self._freeze_result_metadata(result_metadata), + result_metadata_id, + result_metadata_decoder_context) self.column_encryption_policy = column_encryption_policy self.is_idempotent = False self._is_lwt = is_lwt + @staticmethod + def _freeze_result_metadata(result_metadata): + """ + Return immutable column-definition sequences. + + Type descriptor classes are intentionally left untouched, but neither + the ordered result sequence nor a list-valued column entry may remain + aliased to mutable application state after publication. + """ + if result_metadata is None: + return None + frozen_metadata = [] + for column in result_metadata: + column = tuple(column) + if len(column) != 4: + raise ValueError( + "Result metadata columns must contain exactly four " + "items: keyspace, table, name, and type") + frozen_metadata.append(column) + return tuple(frozen_metadata) + + @property + def _result_metadata_snapshot(self): + """ + Atomically snapshot ``(metadata, metadata_id, decoder_context)``. + + ``decoder_context`` is ``None`` when its provenance is unknown. + """ + return self._result_metadata_state + @property def result_metadata_and_id(self): """ - The cached result metadata and its metadata id as one immutable - ``(result_metadata, result_metadata_id)`` pair. + A defensive copy of the cached result metadata and its metadata id as + one ``(result_metadata, result_metadata_id)`` pair. - Read this property when both values are needed together: the tuple is - replaced atomically by :meth:`update_result_metadata`, so a single read - can never observe the metadata of one schema version paired with the - metadata id of another. + Read this property when both values are needed together. A single state + snapshot supplies both values, and mutating the returned metadata list + cannot alter the cached metadata associated with the id. """ - return self._result_metadata_and_id + result_metadata_state = self._result_metadata_state + result_metadata = result_metadata_state[0] + return ( + list(result_metadata) if result_metadata is not None else None, + result_metadata_state[1]) @property def result_metadata(self): """ - Cached result metadata (column definitions) from PREPARE. Read-only: - :meth:`update_result_metadata` is the only way to replace it, so it can - never be assigned separately from the id it belongs to. + A defensive list copy of cached result metadata from PREPARE. + + The property cannot be assigned, and mutating the returned list does + not change the metadata cached with :attr:`result_metadata_id`. """ - return self._result_metadata_and_id[0] + result_metadata = self._result_metadata_state[0] + return list(result_metadata) if result_metadata is not None else None @property def result_metadata_id(self): @@ -505,7 +545,7 @@ def result_metadata_id(self): :meth:`update_result_metadata` is the only way to replace it, so it can never be assigned separately from the metadata it describes. """ - return self._result_metadata_and_id[1] + return self._result_metadata_state[1] def update_result_metadata(self, result_metadata, result_metadata_id): """ @@ -518,18 +558,94 @@ def update_result_metadata(self, result_metadata, result_metadata_id): Also re-arms :attr:`_warned_missing_column_metadata`, so an anomaly that recurs after the metadata was recovered is logged again. + + This compatibility method cannot establish which decoder context + produced ``result_metadata``, so its provenance is cleared. Internal + response paths should use :meth:`_update_result_metadata` instead. + """ + self._update_result_metadata(result_metadata, result_metadata_id, None) + + def _update_result_metadata(self, result_metadata, result_metadata_id, + decoder_context): """ - self._result_metadata_and_id = (result_metadata, result_metadata_id) + Replace metadata, id, and decoder context atomically. + """ + self._result_metadata_state = ( + self._freeze_result_metadata(result_metadata), + result_metadata_id, + decoder_context) + self._warned_missing_column_metadata = False + + def _invalidate_result_metadata_id(self): + """ + Invalidate the server metadata id without losing cached metadata or its + decoder context. + + The next execute must request fresh definitions rather than allowing + the server to omit them based on a no-longer-safe id. + """ + result_metadata, _, decoder_context = self._result_metadata_state + self._result_metadata_state = ( + result_metadata, None, decoder_context) self._warned_missing_column_metadata = False + def __getstate__(self): + """ + Do not persist runtime-only decoder provenance. + + A protocol handler may not be picklable, and a Cluster context token is + meaningful only to the process that created it. Clearing the context + makes a restored statement request fresh metadata before skipping it. + """ + state = self.__dict__.copy() + # This also lets an object reconstructed with a legacy instance layout + # retain that layout until __setstate__ migrates it. + if '_result_metadata_state' not in state: + return state + result_metadata, result_metadata_id, _ = \ + self._result_metadata_state + state['_result_metadata_state'] = ( + result_metadata, result_metadata_id, None) + return state + + def __setstate__(self, state): + """ + Restore pickles created before ``_result_metadata_state`` was added. + + Pair-based pickles predate decoder-context provenance, while still + older pickles stored the metadata and id as independent attributes. + Both therefore migrate with unknown provenance. + """ + state = state.copy() + result_metadata_state = state.get('_result_metadata_state') + if result_metadata_state is None: + result_metadata_and_id = state.pop( + '_result_metadata_and_id', (None, None)) + result_metadata = state.pop( + 'result_metadata', result_metadata_and_id[0]) + result_metadata_id = state.pop( + 'result_metadata_id', result_metadata_and_id[1]) + decoder_context = None + else: + result_metadata, result_metadata_id, decoder_context = \ + result_metadata_state + + state['_result_metadata_state'] = ( + self._freeze_result_metadata(result_metadata), + result_metadata_id, + decoder_context) + self.__dict__.update(state) + @classmethod def from_message(cls, query_id, column_metadata, pk_indexes, cluster_metadata, query, prepared_keyspace, protocol_version, result_metadata, - result_metadata_id, is_lwt, column_encryption_policy=None): + result_metadata_id, is_lwt, column_encryption_policy=None, + result_metadata_decoder_context=None): if not column_metadata: return PreparedStatement(column_metadata, query_id, None, query, prepared_keyspace, protocol_version, result_metadata, - result_metadata_id, is_lwt, column_encryption_policy) + result_metadata_id, is_lwt, column_encryption_policy, + result_metadata_decoder_context) if pk_indexes: routing_key_indexes = pk_indexes @@ -555,7 +671,8 @@ def from_message(cls, query_id, column_metadata, pk_indexes, cluster_metadata, return PreparedStatement(column_metadata, query_id, routing_key_indexes, query, prepared_keyspace, protocol_version, result_metadata, - result_metadata_id, is_lwt, column_encryption_policy) + result_metadata_id, is_lwt, column_encryption_policy, + result_metadata_decoder_context) def bind(self, values): """ diff --git a/docs/scylla-specific.rst b/docs/scylla-specific.rst index 4f61846b4c..614d69a6c1 100644 --- a/docs/scylla-specific.rst +++ b/docs/scylla-specific.rst @@ -162,12 +162,12 @@ Prepared Statement Metadata Caching (``SCYLLA_USE_METADATA_ID``) ---------------------------------------------------------------- When the ``SCYLLA_USE_METADATA_ID`` extension is negotiated, the driver requests the -server to skip sending full result metadata with each prepared SELECT's EXECUTE -response (the ``skip_meta`` optimization), relying instead on the metadata cached -from the initial ``PREPARE`` call. Without change detection this would be unsafe: if -the table schema changes after a statement is prepared (e.g., a column is added, -removed, or its type is altered), the cached metadata becomes stale — leading to -decoding errors or incorrect data. +server to skip sending full result metadata with eligible prepared statements' +EXECUTE responses (the ``skip_meta`` optimization), relying instead on the metadata +cached from the initial ``PREPARE`` call. Without change detection this would be +unsafe: if the table schema changes after a statement is prepared (e.g., a column is +added, removed, or its type is altered), the cached metadata becomes stale — leading +to decoding errors or incorrect data. ScyllaDB solves this by backporting the ``metadata_id`` mechanism from CQL native protocol v5 as a v4 extension: ``SCYLLA_USE_METADATA_ID``. When this extension is @@ -178,20 +178,34 @@ new metadata hash together with the updated column definitions. The driver automatically updates its cache and uses the new metadata to decode the current response — all transparently, with no application code change required. +The metadata id describes the server-side result shape. It does not represent +client-side decoding configuration, such as a newly registered Python class for a +user-defined type. The driver therefore binds cached metadata to an immutable snapshot +of the built-in protocol handler and the Cluster-side UDT mapping that decoded it. A +change in either context forces a full-metadata response before ``skip_meta`` can be +used again. Custom protocol handlers always receive full result metadata because +their opcode and type maps are application-mutable. A Session that has captured +a built-in handler snapshot keeps using that immutable configuration; a modified +built-in configuration cannot create a new eligible snapshot. + **Behaviour summary:** - Automatically negotiated at connection time when the ScyllaDB node supports it. -- ``skip_meta`` is enabled (metadata omitted from EXECUTE responses) only when it - is safe: the prepared statement must carry both a ``result_metadata_id`` and - usable cached result metadata from PREPARE, *and* the connection serving the - request must have negotiated ``SCYLLA_USE_METADATA_ID`` — decided per - connection when the request is serialized. +- A prepared statement is eligible for ``skip_meta`` (metadata omitted from EXECUTE + responses) when it carries both a ``result_metadata_id`` and usable cached result + metadata from PREPARE, uses an unmodified built-in protocol handler, *and* the + connection serving the request has negotiated + ``SCYLLA_USE_METADATA_ID`` — decided per connection when the request is + serialized. - Plain CQL v5 connections are unaffected: the metadata id is part of the native v5 EXECUTE frame layout and is still sent, but the driver does not request skip-metadata there, so such connections keep receiving full result metadata. - When a schema change is detected by the server, the driver refreshes both the cached column metadata and the metadata hash for that prepared statement so that all subsequent executions benefit immediately. +- Registering a Python class for a user-defined type invalidates prepared metadata + ids in that Cluster. The next eligible execution requests fresh definitions so + values are decoded with the new mapping. - Statements prepared before the extension was negotiated (e.g., during a rolling upgrade) start without a metadata hash, but acquire one automatically: on their first execution over a connection with the extension, the driver sends an empty @@ -200,11 +214,12 @@ response — all transparently, with no application code change required. ``skip_meta`` optimization — no re-prepare or client restart is needed. **Current scope:** the optimization applies to any prepared statement that has -non-empty cached result columns — in practice, SELECT queries. -UPDATE/INSERT/DELETE statements naturally return no result columns, so -their ``result_metadata`` is always empty and ``skip_meta`` is never set for -them. There is no code-level restriction to SELECT; the behaviour follows -directly from the data. +non-empty cached result columns. This includes SELECT queries and conditional +INSERT, UPDATE, and DELETE statements (lightweight transactions), whose results +include an ``[applied]`` column and may include values from the existing row. +Ordinary DML statements that produce no result columns remain ineligible. There is +no code-level restriction by statement type; the behaviour follows directly from +the result metadata. For full protocol details see the ScyllaDB CQL protocol extensions documentation: https://github.com/scylladb/scylladb/blob/master/docs/dev/protocol-extensions.md diff --git a/tests/unit/test_prepared_metadata_cache_regressions.py b/tests/unit/test_prepared_metadata_cache_regressions.py new file mode 100644 index 0000000000..29d7fac5e8 --- /dev/null +++ b/tests/unit/test_prepared_metadata_cache_regressions.py @@ -0,0 +1,569 @@ +# Copyright DataStax, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import io +import struct +from collections import defaultdict +from threading import Lock +from unittest.mock import Mock, patch + +import pytest + +from cassandra import ConsistencyLevel, type_codes as type_code_constants +from cassandra.cluster import ( + Cluster, + Session, + _PREPARED_METADATA_HANDLER_SNAPSHOTS, + _prepared_metadata_decoder_context, + _prepared_metadata_protocol_handler_snapshot, +) +from cassandra.cqltypes import Int32Type, UserType +from cassandra.protocol import ( + ProtocolHandler, + RESULT_KIND_ROWS, + ResultMessage, +) +from cassandra.protocol_features import ProtocolFeatures +from cassandra.query import BoundStatement, PreparedStatement + + +_KEYSPACE = 'prepared_metadata_cache_regressions' +_TABLE = 'values_by_id' + + +@pytest.fixture(autouse=True) +def clear_prepared_metadata_handler_snapshots(): + with patch.dict( + _PREPARED_METADATA_HANDLER_SNAPSHOTS, {}, clear=True): + yield + + +def _pack_string(value): + encoded = value.encode('utf-8') + return struct.pack('>H', len(encoded)) + encoded + + +def _pack_value(value): + return struct.pack('>i', len(value)) + value + + +def _rows_body(column_name, type_description, value, no_metadata=False): + flags = ResultMessage._NO_METADATA_FLAG if no_metadata else 0 + body = [ + struct.pack('>i', RESULT_KIND_ROWS), + struct.pack('>i', flags), + struct.pack('>i', 1), + ] + if not no_metadata: + body.extend(( + _pack_string(_KEYSPACE), + _pack_string(_TABLE), + _pack_string(column_name), + type_description, + )) + body.extend(( + struct.pack('>i', 1), + _pack_value(value), + )) + return b''.join(body) + + +def _int_rows_body(value, no_metadata=False): + return _rows_body( + 'value', + struct.pack('>H', type_code_constants.Int32Type), + struct.pack('>i', value), + no_metadata, + ) + + +def _udt_rows_body(value, no_metadata=False): + udt_description = b''.join(( + struct.pack('>H', type_code_constants.UserType), + _pack_string(_KEYSPACE), + _pack_string('address'), + struct.pack('>H', 1), + _pack_string('number'), + struct.pack('>H', type_code_constants.Int32Type), + )) + udt_value = _pack_value(struct.pack('>i', value)) + return _rows_body('address', udt_description, udt_value, no_metadata) + + +def _decode_rows(message_class, body, user_type_map=None, result_metadata=None): + return message_class.recv_body( + io.BytesIO(body), + protocol_version=4, + protocol_features=ProtocolFeatures(), + user_type_map=user_type_map or {}, + result_metadata=result_metadata, + column_encryption_policy=None, + ) + + +class _TimesTenInt32Type(Int32Type): + + @staticmethod + def deserialize(byts, protocol_version): + return Int32Type.deserialize(byts, protocol_version) * 10 + + +class _CustomResultMessage(ResultMessage): + type_codes = ResultMessage.type_codes.copy() + type_codes[type_code_constants.Int32Type] = _TimesTenInt32Type + + +class _CustomProtocolHandler(ProtocolHandler): + message_types_by_opcode = ProtocolHandler.message_types_by_opcode.copy() + message_types_by_opcode[ResultMessage.opcode] = _CustomResultMessage + + +class Address: + + def __init__(self, number): + self.number = number + + +def _prepared_statement(result_metadata, result_metadata_id): + return PreparedStatement( + column_metadata=[], + query_id=b'query-id', + routing_key_indexes=None, + query='SELECT address FROM values_by_id', + keyspace=_KEYSPACE, + protocol_version=4, + result_metadata=result_metadata, + result_metadata_id=result_metadata_id, + ) + + +def _configure_execute_session(session): + session.cluster._config_mode = object() + session.cluster.allow_beta_protocol_version = False + session._protocol_version = 4 + session.default_fetch_size = 5000 + session.use_client_timestamp = False + session._metrics = None + session.keyspace = _KEYSPACE + + profile = Mock() + profile.consistency_level = ConsistencyLevel.ONE + profile.serial_consistency_level = None + profile.continuous_paging_options = None + profile.retry_policy = Mock() + profile.row_factory = Mock() + profile.load_balancing_policy = Mock() + profile.speculative_execution_policy = None + session._maybe_get_execution_profile.return_value = profile + + +def _capture_execute(session, prepared_statement): + _configure_execute_session(session) + bound = BoundStatement(prepared_statement).bind(()) + with patch('cassandra.cluster.ResponseFuture') as response_future: + Session._create_response_future( + session, + bound, + parameters=None, + trace=False, + custom_payload=None, + timeout=1, + ) + + call = response_future.call_args + return call.args[1], call.kwargs['bound_result_metadata'] + + +def test_late_udt_registration_forces_result_metadata_refresh(): + UserType.evict_udt_class(_KEYSPACE, 'address') + try: + initial = _decode_rows(ResultMessage, _udt_rows_body(7)) + cached_metadata = initial.column_metadata + prepared = _prepared_statement(cached_metadata, b'metadata-id') + + cluster = object.__new__(Cluster) + cluster.protocol_version = 4 + cluster._user_types = defaultdict(dict) + cluster.sessions = () + cluster._prepared_statements = {prepared.query_id: prepared} + cluster._prepared_statement_lock = Lock() + + cluster.register_user_type(_KEYSPACE, 'address', Address) + + # A metadata-less response would still deserialize through the UDT class + # cached before registration, which has no mapped_class. + stale = _decode_rows( + ResultMessage, + _udt_rows_body(7, no_metadata=True), + user_type_map=cluster._user_types, + result_metadata=cached_metadata, + ) + assert not isinstance(stale.parsed_rows[0][0], Address) + + # Registering a mapping is a client-side metadata change: the server's + # metadata id cannot detect it. Invalidate the id so the next execute + # requests definitions and rebuilds the UDT type with the new mapping. + assert prepared.result_metadata_id is None + + session = Mock(spec=Session) + session.cluster = cluster + session.client_protocol_handler = ProtocolHandler + message, _ = _capture_execute(session, prepared) + assert message.skip_meta is False + assert message.result_metadata_id is None + + refreshed = _decode_rows( + ResultMessage, + _udt_rows_body(7), + user_type_map=cluster._user_types, + result_metadata=cached_metadata, + ) + assert isinstance(refreshed.parsed_rows[0][0], Address) + assert refreshed.parsed_rows[0][0].number == 7 + finally: + UserType.evict_udt_class(_KEYSPACE, 'address') + + +def test_failed_udt_registration_still_invalidates_prepared_metadata(): + cached_metadata = [('ks', 'tbl', 'address', Mock())] + prepared = _prepared_statement(cached_metadata, b'metadata-id') + old_context = object() + prepared._update_result_metadata( + cached_metadata, + b'metadata-id', + _prepared_metadata_decoder_context( + _prepared_metadata_protocol_handler_snapshot(ProtocolHandler), + old_context), + ) + + session = Mock() + session.user_type_registered.side_effect = RuntimeError('registration failed') + cluster = object.__new__(Cluster) + cluster.protocol_version = 4 + cluster._user_types = defaultdict(dict) + cluster.sessions = (session,) + cluster._prepared_statements = {prepared.query_id: prepared} + cluster._prepared_statement_lock = Lock() + cluster._prepared_metadata_context = old_context + + with pytest.raises(RuntimeError, match='registration failed'): + cluster.register_user_type(_KEYSPACE, 'address', Address) + + assert cluster._user_types[_KEYSPACE]['address'] is Address + assert cluster._prepared_metadata_context is not old_context + assert prepared.result_metadata_id is None + + +def test_udt_registration_invalidates_before_session_callbacks(): + cached_metadata = [('ks', 'tbl', 'address', Mock())] + prepared = _prepared_statement(cached_metadata, b'metadata-id') + old_context = object() + prepared._update_result_metadata( + cached_metadata, + b'metadata-id', + _prepared_metadata_decoder_context( + _prepared_metadata_protocol_handler_snapshot(ProtocolHandler), + old_context), + ) + + cluster = object.__new__(Cluster) + cluster.protocol_version = 4 + cluster._user_types = defaultdict(dict) + cluster._prepared_statements = {prepared.query_id: prepared} + cluster._prepared_statement_lock = Lock() + cluster._prepared_metadata_context = old_context + + execute_session = Mock(spec=Session) + execute_session.cluster = cluster + execute_session.client_protocol_handler = ProtocolHandler + observed = {} + registration_session = Mock() + + def inspect_during_registration(*args): + observed['transition_context'] = \ + cluster._prepared_metadata_context + transition_decoder_context = \ + _prepared_metadata_decoder_context( + _prepared_metadata_protocol_handler_snapshot( + ProtocolHandler), + observed['transition_context'], + ) + # Simulate a full-metadata response racing with registration and + # publishing descriptors under the transition context. + prepared._update_result_metadata( + cached_metadata, + b'transition-metadata-id', + transition_decoder_context, + ) + observed['message'], _ = _capture_execute( + execute_session, prepared) + + registration_session.user_type_registered.side_effect = \ + inspect_during_registration + cluster.sessions = (registration_session,) + + cluster.register_user_type(_KEYSPACE, 'address', Address) + + assert observed['transition_context'] is None + assert observed['message'].skip_meta is False + assert observed['message'].result_metadata_id is None + assert cluster._prepared_metadata_context is not None + + +def test_protocol_handler_change_forces_result_metadata_refresh(): + default_result = _decode_rows(ResultMessage, _int_rows_body(7)) + cached_metadata = default_result.column_metadata + + session = Mock(spec=Session) + session.client_protocol_handler = ProtocolHandler + session._protocol_version = 4 + session.default_timeout = 1 + session.cluster.metadata = Mock() + session.cluster.column_encryption_policy = None + session.cluster.prepare_on_all_hosts = False + + response = Mock() + response.query_id = b'query-id' + response.bind_metadata = [] + response.pk_indexes = None + response.column_metadata = cached_metadata + response.result_metadata_id = b'metadata-id' + response.is_lwt = False + + prepare_future = Mock() + prepare_future.result.return_value.one.return_value = response + prepare_future.custom_payload = None + with patch('cassandra.cluster.ResponseFuture', return_value=prepare_future): + prepared = Session.prepare(session, 'SELECT value FROM values_by_id') + + # With NO_METADATA, even the custom ResultMessage must use the default type + # cached by PREPARE and therefore cannot apply its custom decoder. + stale = _decode_rows( + _CustomResultMessage, + _int_rows_body(7, no_metadata=True), + result_metadata=cached_metadata, + ) + assert stale.parsed_rows == [(7,)] + + session.client_protocol_handler = _CustomProtocolHandler + message, bound_result_metadata = _capture_execute(session, prepared) + + # A cache produced by another handler is not safe to use for NO_METADATA. + # Send the empty id sentinel and suppress skip so the server returns column + # definitions for the current handler to interpret. + assert message.skip_meta is False + assert message.result_metadata_id is None + assert not bound_result_metadata + + refreshed = _decode_rows(_CustomResultMessage, _int_rows_body(7)) + assert refreshed.parsed_rows == [(70,)] + + # Custom handlers expose mutable opcode/type maps. Even after one full + # response, class identity cannot prove that their decoding configuration + # is unchanged, so they remain ineligible for skip-metadata. + prepared._update_result_metadata( + refreshed.column_metadata, b'custom-metadata-id', None) + message, bound_result_metadata = _capture_execute(session, prepared) + assert message.skip_meta is False + assert message.result_metadata_id is None + assert not bound_result_metadata + + +def test_builtin_protocol_handler_map_change_forces_result_metadata_refresh( + monkeypatch): + default_result = _decode_rows(ResultMessage, _int_rows_body(7)) + cached_metadata = default_result.column_metadata + cluster_context = object() + prepared = _prepared_statement(cached_metadata, b'metadata-id') + handler_snapshot = \ + _prepared_metadata_protocol_handler_snapshot(ProtocolHandler) + prepared._update_result_metadata( + cached_metadata, + b'metadata-id', + _prepared_metadata_decoder_context( + handler_snapshot, + cluster_context), + ) + + session = Mock(spec=Session) + session.client_protocol_handler = ProtocolHandler + session.cluster._prepared_metadata_context = cluster_context + + monkeypatch.setitem( + ProtocolHandler.message_types_by_opcode, + ResultMessage.opcode, + _CustomResultMessage, + ) + modified_handler = \ + _prepared_metadata_protocol_handler_snapshot(ProtocolHandler) + assert modified_handler is ProtocolHandler + assert _prepared_metadata_decoder_context( + modified_handler, cluster_context) is None + + message, bound_result_metadata = _capture_execute(session, prepared) + + assert message.skip_meta is False + assert message.result_metadata_id is None + assert not bound_result_metadata + + # A future that already captured the old built-in configuration remains + # internally consistent even after the public source map changes. + frozen_response = handler_snapshot.decode_message( + protocol_version=4, + protocol_features=ProtocolFeatures(), + user_type_map={}, + stream_id=0, + flags=0, + opcode=ResultMessage.opcode, + body=_int_rows_body(7, no_metadata=True), + decompressor=None, + result_metadata=cached_metadata, + ) + current_response = _CustomProtocolHandler.decode_message( + protocol_version=4, + protocol_features=ProtocolFeatures(), + user_type_map={}, + stream_id=0, + flags=0, + opcode=ResultMessage.opcode, + body=_int_rows_body(7), + decompressor=None, + result_metadata=None, + ) + assert frozen_response.parsed_rows == [(7,)] + assert current_response.parsed_rows == [(70,)] + + # Even metadata decoded after the customization cannot make an arbitrary + # application ResultMessage eligible for metadata-less responses. + prepared._update_result_metadata( + current_response.column_metadata, + b'custom-metadata-id', + None, + ) + message, bound_result_metadata = _capture_execute(session, prepared) + assert message.skip_meta is False + assert message.result_metadata_id is None + assert not bound_result_metadata + + +def test_session_reuses_its_protocol_handler_snapshot(): + cached_metadata = [('ks', 'tbl', 'value', Mock())] + cluster_context = object() + prepared = _prepared_statement(cached_metadata, b'metadata-id') + prepared._update_result_metadata( + cached_metadata, + b'metadata-id', + _prepared_metadata_decoder_context( + _prepared_metadata_protocol_handler_snapshot(ProtocolHandler), + cluster_context), + ) + session = Mock(spec=Session) + session.client_protocol_handler = ProtocolHandler + session.cluster._prepared_metadata_context = cluster_context + + with patch( + 'cassandra.cluster.' + '_prepared_metadata_protocol_handler_snapshot', + wraps=_prepared_metadata_protocol_handler_snapshot) as snapshot: + first_message, _ = _capture_execute(session, prepared) + second_message, _ = _capture_execute(session, prepared) + + assert snapshot.call_count == 1 + assert first_message.skip_meta is True + assert second_message.skip_meta is True + + +def test_full_metadata_response_recovers_after_udt_transition(): + cached_metadata = [('ks', 'tbl', 'address', Mock())] + prepared = _prepared_statement(cached_metadata, None) + cluster_context = object() + handler_snapshot = \ + _prepared_metadata_protocol_handler_snapshot(ProtocolHandler) + prepared._update_result_metadata( + cached_metadata, + b'new-metadata-id', + _prepared_metadata_decoder_context( + handler_snapshot, cluster_context), + ) + session = Mock(spec=Session) + session.client_protocol_handler = ProtocolHandler + session.cluster._prepared_metadata_context = cluster_context + + message, bound_result_metadata = _capture_execute(session, prepared) + + assert message.skip_meta is True + assert message.result_metadata_id == b'new-metadata-id' + assert bound_result_metadata == tuple(cached_metadata) + + +def test_non_subclassable_result_customization_falls_back_to_full_metadata( + monkeypatch): + class _NonSubclassableResultMessage(ResultMessage): + + def __init_subclass__(cls, **kwargs): + raise TypeError('must not be subclassed') + + monkeypatch.setitem( + ProtocolHandler.message_types_by_opcode, + ResultMessage.opcode, + _NonSubclassableResultMessage, + ) + + assert _prepared_metadata_protocol_handler_snapshot( + ProtocolHandler) is ProtocolHandler + + +def test_custom_result_comparison_error_falls_back_to_full_metadata( + monkeypatch): + class RaisingComparisonMeta(type(ResultMessage)): + + def __eq__(cls, other): + raise ValueError('comparison is application code') + + __hash__ = type.__hash__ + + class _RaisingComparisonResultMessage( + ResultMessage, metaclass=RaisingComparisonMeta): + pass + + monkeypatch.setitem( + ProtocolHandler.message_types_by_opcode, + ResultMessage.opcode, + _RaisingComparisonResultMessage, + ) + + assert _prepared_metadata_protocol_handler_snapshot( + ProtocolHandler) is ProtocolHandler + + +def test_handler_snapshot_does_not_register_its_internal_result_class(): + with patch.dict( + _PREPARED_METADATA_HANDLER_SNAPSHOTS, {}, clear=True), \ + patch('cassandra.protocol.register_class') as register_class: + _prepared_metadata_protocol_handler_snapshot(ProtocolHandler) + + register_class.assert_not_called() + + +def test_handler_snapshot_freezes_all_result_message_attributes(monkeypatch): + handler_snapshot = \ + _prepared_metadata_protocol_handler_snapshot(ProtocolHandler) + result_snapshot = handler_snapshot.message_types_by_opcode[ + ResultMessage.opcode] + original_metadata_id_default = result_snapshot.result_metadata_id + + monkeypatch.setattr(ResultMessage, 'result_metadata_id', object()) + + assert result_snapshot.result_metadata_id is \ + original_metadata_id_default diff --git a/tests/unit/test_protocol.py b/tests/unit/test_protocol.py index 75dc69bca5..1575ccaa3d 100644 --- a/tests/unit/test_protocol.py +++ b/tests/unit/test_protocol.py @@ -314,7 +314,7 @@ def test_recv_results_metadata_no_metadata_flag_skips_metadata_id(self): # recv_results_metadata returns early on NO_METADATA; result_metadata_id # must never be set as an instance attribute (it is not a class default). # column_metadata is a class attribute defaulting to None and must remain so. - assert not hasattr(msg, 'result_metadata_id') + assert msg.result_metadata_id is None assert msg.column_metadata is None def test_query_message(self): diff --git a/tests/unit/test_query.py b/tests/unit/test_query.py index 1bbe069667..b7e2fc1c98 100644 --- a/tests/unit/test_query.py +++ b/tests/unit/test_query.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import pickle import unittest import pytest @@ -138,7 +139,8 @@ def _make_statement(result_metadata, result_metadata_id): def test_constructor_sets_pair(self): meta = [('ks', 'tb', 'col', None)] ps = self._make_statement(meta, b'hash') - assert ps.result_metadata is meta + assert ps.result_metadata == meta + assert isinstance(ps.result_metadata, list) assert ps.result_metadata_id == b'hash' assert ps.result_metadata_and_id == (meta, b'hash') @@ -153,6 +155,99 @@ def test_update_replaces_pair_atomically(self): assert snapshot_before == ([('ks', 'tb', 'old', None)], b'old') assert ps.result_metadata_and_id == (new_meta, b'new') + def test_constructor_does_not_retain_mutable_metadata_sequence(self): + meta = [ + ('ks', 'tb', 'first', None), + ('ks', 'tb', 'second', None), + ] + ps = self._make_statement(meta, b'hash') + + meta.reverse() + meta.append(('ks', 'tb', 'third', None)) + + assert ps.result_metadata == [ + ('ks', 'tb', 'first', None), + ('ks', 'tb', 'second', None), + ] + + def test_update_does_not_retain_mutable_metadata_sequence(self): + ps = self._make_statement([], None) + meta = [('ks', 'tb', 'col', None)] + + ps.update_result_metadata(meta, b'hash') + meta.clear() + + assert ps.result_metadata == [('ks', 'tb', 'col', None)] + assert ps.result_metadata_id == b'hash' + + def test_public_metadata_is_a_defensive_copy(self): + ps = self._make_statement( + [('ks', 'tb', 'col', None)], b'hash') + + returned_metadata = ps.result_metadata + returned_metadata.clear() + + assert ps.result_metadata == [('ks', 'tb', 'col', None)] + assert ps.result_metadata_id == b'hash' + + def test_update_freezes_mutable_column_definition(self): + ps = self._make_statement([], None) + column = ['ks', 'tb', 'col', None] + ps.update_result_metadata([column], b'hash') + + column[2] = 'changed' + column[3] = object() + + assert ps.result_metadata == [('ks', 'tb', 'col', None)] + + def test_rejects_malformed_column_definition(self): + ps = self._make_statement([], None) + + with pytest.raises(ValueError, match='exactly four'): + ps.update_result_metadata([('ks', 'tb', 'col')], b'hash') + + def test_internal_update_tracks_decoder_context_atomically(self): + ps = self._make_statement([], None) + decoder_context = object() + meta = [('ks', 'tb', 'col', None)] + + ps._update_result_metadata(meta, b'hash', decoder_context) + snapshot = ps._result_metadata_snapshot + + assert snapshot == (tuple(meta), b'hash', decoder_context) + + ps._update_result_metadata([], b'new-hash', object()) + assert snapshot == (tuple(meta), b'hash', decoder_context) + + def test_public_update_clears_decoder_context_provenance(self): + ps = self._make_statement([], None) + ps._update_result_metadata([], b'old-hash', object()) + + ps.update_result_metadata([], b'new-hash') + + assert ps._result_metadata_snapshot == ((), b'new-hash', None) + + def test_invalidate_id_preserves_metadata_and_decoder_context(self): + ps = self._make_statement([], None) + decoder_context = object() + meta = [('ks', 'tb', 'col', None)] + ps._update_result_metadata(meta, b'hash', decoder_context) + + ps._invalidate_result_metadata_id() + + assert ps._result_metadata_snapshot == ( + tuple(meta), None, decoder_context) + + def test_pickle_drops_runtime_decoder_context(self): + ps = self._make_statement([], None) + ps._update_result_metadata( + [('ks', 'tb', 'col', None)], b'hash', lambda: None) + + restored = pickle.loads(pickle.dumps(ps)) + + assert restored._result_metadata_snapshot == ( + (('ks', 'tb', 'col', None),), b'hash', None) + def test_halves_of_the_pair_cannot_be_assigned_individually(self): # Assigning one half alone would leave the other stale, which is exactly # the torn state update_result_metadata() exists to prevent, so neither @@ -167,3 +262,31 @@ def test_halves_of_the_pair_cannot_be_assigned_individually(self): ps.result_metadata = [] assert ps.result_metadata_and_id == (meta, b'hash') + + def test_migrates_pickle_with_metadata_pair(self): + ps = self._make_statement([], None) + ps.__dict__.pop('_result_metadata_state') + ps.__dict__['_result_metadata_and_id'] = ( + [('ks', 'tb', 'col', None)], b'hash') + + restored = pickle.loads(pickle.dumps(ps)) + + assert restored._result_metadata_snapshot == ( + (('ks', 'tb', 'col', None),), b'hash', None) + assert '_result_metadata_and_id' not in restored.__dict__ + + def test_migrates_pickle_with_independent_metadata_attributes(self): + ps = self._make_statement([], None) + state = ps.__dict__.copy() + state.pop('_result_metadata_state') + state['result_metadata'] = [ + ('ks', 'tb', 'col', None)] + state['result_metadata_id'] = b'hash' + + restored = object.__new__(PreparedStatement) + restored.__setstate__(state) + + assert restored._result_metadata_snapshot == ( + (('ks', 'tb', 'col', None),), b'hash', None) + assert 'result_metadata' not in restored.__dict__ + assert 'result_metadata_id' not in restored.__dict__ diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index cf1194a91f..72643e2f66 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -16,10 +16,14 @@ from collections import deque from threading import RLock -from unittest.mock import Mock, MagicMock, ANY - -from cassandra import ConsistencyLevel, Unavailable, SchemaTargetType, SchemaChangeType, OperationTimedOut -from cassandra.cluster import Session, ResponseFuture, NoHostAvailable, ProtocolVersion, ControlConnectionQueryFallback +from unittest.mock import Mock, MagicMock, ANY, patch + +from cassandra import (ConsistencyLevel, DriverException, OperationTimedOut, + SchemaChangeType, SchemaTargetType, Unavailable) +from cassandra.cluster import (Session, ResponseFuture, NoHostAvailable, + ProtocolVersion, ControlConnectionQueryFallback, + _prepared_metadata_decoder_context, + _prepared_metadata_protocol_handler_snapshot) from cassandra.connection import Connection, ConnectionException from cassandra.protocol import (ReadTimeoutErrorMessage, WriteTimeoutErrorMessage, UnavailableErrorMessage, ResultMessage, QueryMessage, @@ -36,11 +40,24 @@ import pytest +_PREPARED_METADATA_CONTEXT = object() + + +def _decoder_context(protocol_handler=ProtocolHandler, + cluster_context=_PREPARED_METADATA_CONTEXT): + protocol_handler = _prepared_metadata_protocol_handler_snapshot( + protocol_handler) + return _prepared_metadata_decoder_context( + protocol_handler, cluster_context) + + class ResponseFutureTests(unittest.TestCase): def make_basic_session(self): s = Mock(spec=Session) s.row_factory = lambda col_names, rows: [(col_names, rows)] + s.client_protocol_handler = ProtocolHandler + s.cluster._prepared_metadata_context = _PREPARED_METADATA_CONTEXT s.cluster.control_connection._tablets_routing_v1 = False s.cluster.allow_control_connection_query_fallback = ControlConnectionQueryFallback.Disabled return s @@ -884,6 +901,31 @@ def test_prepared_query_not_found(self): assert isinstance(args[-4], PrepareMessage) assert args[-4].query == "SELECT * FROM foobar" + def test_prepared_query_not_found_restores_cache_through_cluster(self): + session = self.make_session() + pool = session._pools.get.return_value + connection = Mock(spec=Connection) + pool.borrow_connection.return_value = (connection, 1) + + query_id = b'a' * 16 + prepared_statement = Mock( + query_id=query_id, + query_string="SELECT * FROM foobar", + keyspace="FooKeyspace") + rf = self.make_response_future(session) + rf.prepared_statement = prepared_statement + rf.send_request() + + session.cluster.protocol_version = ProtocolVersion.V4 + session.cluster._prepared_statements = {} + rf._connection.keyspace = "FooKeyspace" + + result = Mock(spec=PreparedQueryNotFound, info=query_id) + rf._set_result(None, None, None, result) + + session.cluster.add_prepared.assert_called_once_with( + query_id, prepared_statement) + def test_prepared_query_not_found_bad_keyspace(self): session = self.make_session() pool = session._pools.get.return_value @@ -915,6 +957,7 @@ def test_repeat_orig_query_after_succesful_reprepare(self): result_metadata_id=b'foo') response.results = (None, None, None, None, None) response.query_id = query_id + response.column_metadata = [] rf._query = Mock(return_value=True) rf._execute_after_prepare('host', None, None, response) @@ -929,6 +972,35 @@ def test_repeat_orig_query_after_succesful_reprepare(self): rf._query.assert_called_once_with('host') assert rf.prepared_statement.result_metadata_id == b'foo' + def test_execute_after_prepare_id_mismatch_preserves_metadata_snapshot(self): + session = self.make_session() + connection = Mock(spec=Connection) + old_metadata = [('ks', 'tb', 'old_col', Mock())] + prepared_statement = self._make_prepared_statement( + old_metadata, b'old_hash', query_id=b'expected_query_id') + rf = self._make_execute_response_future( + session, connection, prepared_statement) + prepared_snapshot = prepared_statement._result_metadata_snapshot + request_snapshot = rf._request_snapshot + + response = Mock( + spec=ResultMessage, + kind=RESULT_KIND_PREPARED, + query_id=b'unexpected_query_id', + column_metadata=[('ks', 'tb', 'foreign_col', Mock())], + result_metadata_id=b'foreign_hash') + rf._query = Mock(return_value=True) + + rf._execute_after_prepare('host', None, None, response) + + assert isinstance(rf._final_exception, DriverException) + assert prepared_statement._result_metadata_snapshot == prepared_snapshot + assert rf._request_snapshot == request_snapshot + assert rf._bound_result_metadata == request_snapshot[1] + assert rf.message.result_metadata_id == b'old_hash' + assert rf.message.skip_meta is True + rf._query.assert_not_called() + def test_execute_after_prepare_updates_result_metadata_id(self): """ After a PreparedQueryNotFound triggers a reprepare, _execute_after_prepare @@ -956,7 +1028,7 @@ def test_execute_after_prepare_updates_result_metadata_id(self): rf._execute_after_prepare('host', None, None, response) # Both metadata fields must be refreshed from the reprepare response. - assert rf.prepared_statement.result_metadata is new_meta + assert rf.prepared_statement.result_metadata == new_meta assert rf.prepared_statement.result_metadata_id == b'new_hash' assert rf.prepared_statement.result_metadata_and_id == (new_meta, b'new_hash') # Recovering the metadata re-arms the anomaly warning, on this path too. @@ -995,7 +1067,7 @@ def test_execute_after_prepare_no_metadata_id_in_response_clears_id(self): # result_metadata is refreshed (always); result_metadata_id is cleared, not # carried forward from the old pair. - assert rf.prepared_statement.result_metadata is new_meta + assert rf.prepared_statement.result_metadata == new_meta assert rf.prepared_statement.result_metadata_id is None def test_timeout_does_not_release_stream_id(self): @@ -1105,36 +1177,48 @@ def _make_rows_response(self, result_metadata_id=None, column_metadata=None): return response def _make_prepared_statement(self, result_metadata, result_metadata_id, query_id=b'qid'): - return PreparedStatement( + prepared_statement = PreparedStatement( column_metadata=[], query_id=query_id, routing_key_indexes=None, query="SELECT * FROM foo", keyspace='ks', protocol_version=4, result_metadata=result_metadata, result_metadata_id=result_metadata_id) + prepared_statement._update_result_metadata( + result_metadata, result_metadata_id, + _decoder_context()) + return prepared_statement def _make_execute_response_future(self, session, connection, prepared_statement): """ Return a ResponseFuture whose message is an ExecuteMessage and which has a prepared_statement set, as _create_response_future would build it. """ - execute_msg = ExecuteMessage(b'qid', [], ConsistencyLevel.ONE) + metadata, metadata_id, decoder_context = \ + prepared_statement._result_metadata_snapshot + execute_msg = ExecuteMessage( + prepared_statement.query_id, [], ConsistencyLevel.ONE, + skip_meta=bool(metadata) and metadata_id is not None, + result_metadata_id=metadata_id) query = SimpleStatement("SELECT * FROM foo") rf = ResponseFuture( session, execute_msg, query, timeout=1, prepared_statement=prepared_statement, # mirror _create_response_future: snapshot the metadata paired with the id - bound_result_metadata=prepared_statement.result_metadata, + bound_result_metadata=metadata, + result_metadata_decoder_context=decoder_context, + protocol_handler=ProtocolHandler, ) pool = session._pools.get.return_value pool.borrow_connection.return_value = (connection, 1) return rf - def _create_execute_future(self, prepared_statement, continuous_paging_options=None): + def _create_execute_future(self, prepared_statement, + continuous_paging_options=None, session=None): """ Drive the real Session._create_response_future (with a mock session) for a statement bound to `prepared_statement`, returning the ResponseFuture. This exercises the ExecuteMessage construction path where skip_meta and result_metadata_id are decided from the statement's metadata pair. """ - session = self.make_session() + session = session or self.make_session() profile = session._maybe_get_execution_profile.return_value profile.consistency_level = ConsistencyLevel.ONE profile.serial_consistency_level = None @@ -1179,7 +1263,7 @@ def test_set_result_updates_metadata_when_metadata_changed(self): ) rf._set_result(None, None, None, response) - assert ps.result_metadata is new_meta + assert ps.result_metadata == new_meta assert ps.result_metadata_id == b'new_id' # the pair is replaced as one unit — a snapshot can never be torn assert ps.result_metadata_and_id == (new_meta, b'new_id') @@ -1212,7 +1296,7 @@ def test_set_result_does_not_update_metadata_when_metadata_id_absent(self): ) rf._set_result(None, None, None, response) - assert ps.result_metadata is old_meta + assert ps.result_metadata == old_meta assert ps.result_metadata_id == b'old_id' def test_set_result_warns_when_metadata_id_but_no_column_metadata(self): @@ -1316,8 +1400,9 @@ def test_set_result_no_metadata_statement_adopts_metadata_changed(self): ) # Ordinary METADATA_CHANGED handling, so no anomaly warning either. - with self.assertNoLogs('cassandra.cluster', level='WARNING'): + with patch('cassandra.cluster.log.warning') as warning: rf._set_result(None, None, None, response) + warning.assert_not_called() assert ps.result_metadata_and_id == (new_meta, b'new_id') @@ -1351,8 +1436,9 @@ def test_set_result_anomalous_metadata_id_warns_once_and_rearms(self): assert sum('result_metadata_id' in msg for msg in first.output) == 1 # Second identical anomalous response: no new warning (deduped). - with self.assertNoLogs('cassandra.cluster', level='WARNING'): + with patch('cassandra.cluster.log.warning') as warning: rf._set_result(None, None, None, anomalous) + warning.assert_not_called() assert ps.result_metadata_and_id == (old_meta, b'old_id') # A genuine METADATA_CHANGED recovers the metadata and re-arms the warning. @@ -1381,6 +1467,20 @@ def test_create_execute_message_with_metadata_and_id(self): assert rf.message.skip_meta is True assert rf.message.result_metadata_id == b'meta_hash' + def test_session_snapshot_validation_is_not_repeated_per_execute(self): + session = self.make_session() + ps = self._make_prepared_statement( + [('ks', 'tbl', 'col', Mock())], b'meta_hash') + + with patch( + 'cassandra.cluster.' + '_prepared_metadata_protocol_handler_snapshot', + wraps=_prepared_metadata_protocol_handler_snapshot) as snapshot: + self._create_execute_future(ps, session=session) + self._create_execute_future(ps, session=session) + + assert snapshot.call_count == 1 + def test_create_execute_message_without_metadata_id(self): """ A statement prepared before the extension was active (result_metadata_id @@ -1424,13 +1524,11 @@ def test_create_execute_message_result_metadata_empty(self): assert rf.message.skip_meta is False assert rf.message.result_metadata_id == b'meta_hash' - def test_create_execute_message_continuous_paging_disables_skip_meta(self): + def test_continuous_paging_option_disables_skip_meta(self): """ - Continuous paging sessions must never get skip_meta=True, even with a - statement that otherwise qualifies (valid cached metadata + id): - Connection.process_msg hardcodes result_metadata=None for every page - after the first (it isn't threaded through the paging session), so a - skip_meta response would leave nothing to decode page 2+ against. + Continuous paging cannot decode later metadata-less pages because its + connection path does not thread cached result metadata through the + paging session. """ ps = self._make_prepared_statement([('ks', 'tbl', 'col', Mock())], b'meta_hash') @@ -1439,6 +1537,165 @@ def test_create_execute_message_continuous_paging_disables_skip_meta(self): assert rf.message.skip_meta is False assert rf.message.result_metadata_id == b'meta_hash' + def test_cross_cluster_execute_rejects_foreign_decoder_metadata(self): + """ + Decoder revisions are cluster-unique. Two clusters with no UDT + registrations must not consider their cached metadata interchangeable. + """ + ps = self._make_prepared_statement( + [('ks', 'tbl', 'col', Mock())], b'meta_hash') + other_session = self.make_session() + other_session.cluster._prepared_metadata_context = object() + + rf = self._create_execute_future(ps, session=other_session) + + assert rf.message.skip_meta is False + assert rf.message.result_metadata_id is None + assert not rf._bound_result_metadata + + def test_protocol_handler_is_set_before_response_future_timer_starts(self): + """ + A speculative timer may fire from ResponseFuture.__init__, so the + handler must be installed before the timer is created. + """ + class CustomProtocolHandler(ProtocolHandler): + pass + + session = self.make_session() + session.client_protocol_handler = CustomProtocolHandler + metadata = [('ks', 'tbl', 'col', Mock())] + ps = self._make_prepared_statement(metadata, b'meta_hash') + ps._update_result_metadata( + metadata, b'meta_hash', + _decoder_context(CustomProtocolHandler)) + handlers_at_timer_start = [] + + def capture_handler(future): + handlers_at_timer_start.append(future._protocol_handler) + + with patch.object(ResponseFuture, '_start_timer', + new=capture_handler): + rf = self._create_execute_future(ps, session=session) + + assert handlers_at_timer_start == [CustomProtocolHandler] + assert rf._protocol_handler is CustomProtocolHandler + + def test_direct_constructor_uses_matching_metadata_snapshot(self): + session = self.make_session() + metadata = [('ks', 'tbl', 'col', Mock())] + ps = self._make_prepared_statement(metadata, b'meta_hash') + message = ExecuteMessage( + ps.query_id, [], ConsistencyLevel.ONE, + skip_meta=True, result_metadata_id=b'meta_hash') + + rf = ResponseFuture( + session, message, SimpleStatement("SELECT * FROM foo"), 1, + prepared_statement=ps) + + assert rf._bound_result_metadata == tuple(metadata) + assert rf.message.skip_meta is True + assert rf.message.result_metadata_id == b'meta_hash' + + def test_direct_constructor_disables_skip_for_mismatched_snapshot(self): + session = self.make_session() + ps = self._make_prepared_statement( + [('ks', 'tbl', 'new_col', Mock())], b'new_hash') + message = ExecuteMessage( + ps.query_id, [], ConsistencyLevel.ONE, + skip_meta=True, result_metadata_id=b'old_hash') + + rf = ResponseFuture( + session, message, SimpleStatement("SELECT * FROM foo"), 1, + prepared_statement=ps) + + assert not rf._bound_result_metadata + assert rf.message is not message + assert rf.message.skip_meta is False + assert rf.message.result_metadata_id is None + assert message.skip_meta is True + assert message.result_metadata_id == b'old_hash' + + def test_direct_constructor_rejects_explicit_snapshot_without_context(self): + session = self.make_session() + metadata = [('ks', 'tbl', 'old_col', Mock())] + ps = self._make_prepared_statement(metadata, b'meta_hash') + message = ExecuteMessage( + ps.query_id, [], ConsistencyLevel.ONE, + skip_meta=True, result_metadata_id=b'meta_hash') + + rf = ResponseFuture( + session, message, SimpleStatement("SELECT * FROM foo"), 1, + prepared_statement=ps, + bound_result_metadata=metadata) + + assert not rf._bound_result_metadata + assert rf._result_metadata_decoder_context == _decoder_context() + assert rf.message is not message + assert rf.message.skip_meta is False + assert rf.message.result_metadata_id is None + assert message.skip_meta is True + assert message.result_metadata_id == b'meta_hash' + + def test_noop_continuous_paging_option_uses_normal_result_path(self): + metadata = [('ks', 'tbl', 'col', Mock())] + ps = self._make_prepared_statement(metadata, b'meta_hash') + rf = self._create_execute_future( + ps, continuous_paging_options=Mock()) + rf._handle_continuous_paging_first_response = Mock() + rf._set_final_result = Mock() + response = self._make_rows_response( + column_metadata=metadata) + + rf._set_result(None, None, None, response) + + assert not hasattr(rf.message, 'continuous_paging_options') + rf._handle_continuous_paging_first_response.assert_not_called() + rf._set_final_result.assert_called_once() + + def test_next_page_refreshes_prepared_metadata_snapshot(self): + old_metadata = [('ks', 'tbl', 'old_col', Mock())] + new_metadata = [('ks', 'tbl', 'new_col', Mock())] + ps = self._make_prepared_statement(old_metadata, b'old_hash') + rf = self._create_execute_future(ps) + rf._paging_state = b'next-page' + ps._update_result_metadata( + new_metadata, b'new_hash', + _decoder_context()) + rf._make_query_plan = Mock() + rf._start_timer = Mock() + rf.send_request = Mock() + + rf.start_fetching_next_page() + + assert rf._bound_result_metadata == tuple(new_metadata) + assert rf.message.result_metadata_id == b'new_hash' + assert rf.message.skip_meta is True + rf.send_request.assert_called_once_with() + + def test_reprepare_refreshes_rerun_metadata_snapshot(self): + session = self.make_session() + connection = Mock(spec=Connection) + old_metadata = [('ks', 'tbl', 'old_col', Mock())] + new_metadata = [('ks', 'tbl', 'new_col', Mock())] + ps = self._make_prepared_statement( + old_metadata, b'old_hash', query_id=b'reprepare_qid') + rf = self._make_execute_response_future(session, connection, ps) + response = Mock( + spec=ResultMessage, + kind=RESULT_KIND_PREPARED, + query_id=ps.query_id, + column_metadata=new_metadata, + result_metadata_id=b'new_hash') + rf._query = Mock(return_value=True) + + rf._execute_after_prepare('host', None, None, response) + + assert rf._bound_result_metadata == tuple(new_metadata) + assert rf.message.result_metadata_id == b'new_hash' + assert rf.message.skip_meta is True + assert rf._result_metadata_decoder_context == _decoder_context() + rf._query.assert_called_once_with('host') + def test_query_does_not_mutate_execute_message(self): """ _query() must send the ExecuteMessage exactly as constructed: all @@ -1471,6 +1728,42 @@ def test_query_does_not_mutate_execute_message(self): assert rf.message.result_metadata_id is original_id assert not hasattr(rf.message, 'use_metadata_id') + def test_prepared_future_freezes_builtin_decoder_maps(self): + """ + Mutating the public built-in handler map after future construction must + not turn an old-metadata request into a new-decoder response. + """ + class _ReplacementResultMessage(ResultMessage): + pass + + session = self.make_basic_session() + session.cluster._default_load_balancing_policy.make_query_plan.return_value = ['ip1'] + pool = self.make_pool() + session._pools.get.return_value = pool + connection = Mock(spec=Connection) + pool.borrow_connection.return_value = (connection, 1) + ps = self._make_prepared_statement( + [('ks', 'tbl', 'col', Mock())], b'meta_hash') + rf = self._make_execute_response_future( + session, connection, ps) + frozen_handler = rf._protocol_handler + frozen_result_message = \ + frozen_handler.message_types_by_opcode[ResultMessage.opcode] + + with patch.dict( + ProtocolHandler.message_types_by_opcode, + {ResultMessage.opcode: _ReplacementResultMessage}): + rf.send_request() + + sent_decoder = connection.send_msg.call_args.kwargs['decoder'] + assert sent_decoder.__self__ is frozen_handler + assert frozen_handler.message_types_by_opcode[ + ResultMessage.opcode] is frozen_result_message + assert frozen_result_message is not _ReplacementResultMessage + with pytest.raises(TypeError): + frozen_handler.message_types_by_opcode[ + ResultMessage.opcode] = _ReplacementResultMessage + def test_query_decodes_with_construction_snapshot_not_live_cache(self): """ The metadata handed to the decoder must be the snapshot taken when the message @@ -1493,14 +1786,65 @@ def test_query_decodes_with_construction_snapshot_not_live_cache(self): ps = self._make_prepared_statement(meta_v1, b'id1') rf = self._make_execute_response_future(session, connection, ps) # snapshot captured at construction, independent of the live pair - assert rf._bound_result_metadata is meta_v1 + assert rf._bound_result_metadata == tuple(meta_v1) # a concurrent METADATA_CHANGED replaces the statement's cached pair ps.update_result_metadata([('ks', 'tbl', 'col_v2', Mock())], b'id2') - assert rf._bound_result_metadata is meta_v1 + assert rf._bound_result_metadata == tuple(meta_v1) rf.send_request() connection.send_msg.assert_called_once() # _query decodes with the construction snapshot, not the mutated cache - assert connection.send_msg.call_args.kwargs['result_metadata'] is meta_v1 + assert connection.send_msg.call_args.kwargs['result_metadata'] == tuple(meta_v1) + + def test_concurrent_refresh_publishes_message_and_metadata_together(self): + """ + A speculative send that has captured a request snapshot must retain its + matching message/metadata pair if paging or reprepare publishes a new + snapshot while that send is borrowing a connection. + """ + session = self.make_basic_session() + session.cluster._default_load_balancing_policy.make_query_plan.return_value = ['ip1'] + pool = self.make_pool() + session._pools.get.return_value = pool + connection = Mock(spec=Connection) + old_metadata = [('ks', 'tbl', 'old_col', Mock())] + new_metadata = [('ks', 'tbl', 'new_col', Mock())] + ps = self._make_prepared_statement(old_metadata, b'old_hash') + rf = self._make_execute_response_future(session, connection, ps) + old_message = rf.message + new_decoder_context = object() + + def refresh_while_borrowing(**kwargs): + session.cluster._prepared_metadata_context = new_decoder_context + ps._update_result_metadata( + new_metadata, b'new_hash', + _decoder_context(cluster_context=new_decoder_context)) + rf._refresh_prepared_result_metadata() + return connection, 1 + + pool.borrow_connection.side_effect = refresh_while_borrowing + + rf.send_request() + + sent_message = connection.send_msg.call_args.args[0] + sent_metadata = connection.send_msg.call_args.kwargs['result_metadata'] + assert sent_message is old_message + assert sent_message.result_metadata_id == b'old_hash' + assert sent_metadata == tuple(old_metadata) + assert rf.message.result_metadata_id == b'new_hash' + assert rf._bound_result_metadata == tuple(new_metadata) + + late_metadata = [('ks', 'tbl', 'late_col', Mock())] + response = self._make_rows_response( + result_metadata_id=b'late_hash', + column_metadata=late_metadata) + connection.send_msg.call_args.kwargs['cb'](response) + + # The late callback is tagged with the context captured by its own + # request, not the future's newer mutable context. + assert ps._result_metadata_snapshot == ( + tuple(late_metadata), + b'late_hash', + _decoder_context()) diff --git a/tests/unit/test_udt_descriptor_isolation.py b/tests/unit/test_udt_descriptor_isolation.py new file mode 100644 index 0000000000..fc175367c0 --- /dev/null +++ b/tests/unit/test_udt_descriptor_isolation.py @@ -0,0 +1,214 @@ +# Copyright DataStax, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import io +import struct +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from cassandra import type_codes +from cassandra.cqltypes import Int32Type, UserType +from cassandra.protocol import ResultMessage + + +KEYSPACE = 'udt_descriptor_isolation' +UDT_NAME = 'address' +FIELD_NAMES = ('number',) +FIELD_TYPES = (Int32Type,) + + +class AddressA: + + def __init__(self, number): + self.number = number + + +class AddressB: + + def __init__(self, number): + self.number = number + + +class UnhashableMappingMeta(type): + __hash__ = None + + +class UnhashableAddress(metaclass=UnhashableMappingMeta): + + def __init__(self, number): + self.number = number + + +@pytest.fixture(autouse=True) +def clear_udt_descriptor_variants(): + UserType.evict_udt_class(KEYSPACE, UDT_NAME) + yield + UserType.evict_udt_class(KEYSPACE, UDT_NAME) + + +def _pack_string(value): + value = value.encode('utf-8') + return struct.pack('>H', len(value)) + value + + +def _udt_type_description(): + return b''.join(( + struct.pack('>H', type_codes.UserType), + _pack_string(KEYSPACE), + _pack_string(UDT_NAME), + struct.pack('>H', 1), + _pack_string(FIELD_NAMES[0]), + struct.pack('>H', type_codes.Int32Type), + )) + + +def _read_udt_type(mapped_class): + user_type_map = {KEYSPACE: {UDT_NAME: mapped_class}} + return ResultMessage.read_type( + io.BytesIO(_udt_type_description()), user_type_map) + + +def _udt_binary_value(number): + encoded_number = struct.pack('>i', number) + return struct.pack('>i', len(encoded_number)) + encoded_number + + +def test_protocol_udt_descriptors_do_not_mutate_across_mappings(): + descriptor_a = _read_udt_type(AddressA) + descriptor_b = _read_udt_type(AddressB) + + assert descriptor_a is not descriptor_b + assert descriptor_a.mapped_class is AddressA + assert descriptor_b.mapped_class is AddressB + + value_a = descriptor_a.from_binary(_udt_binary_value(1), 4) + value_b = descriptor_b.from_binary(_udt_binary_value(2), 4) + assert isinstance(value_a, AddressA) + assert isinstance(value_b, AddressB) + assert value_a.number == 1 + assert value_b.number == 2 + + +def test_same_mapping_reuses_its_immutable_descriptor(): + first = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=AddressA) + second = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=AddressA) + + assert first is second + assert first.mapped_class is AddressA + + +def test_unhashable_mapping_class_has_an_isolated_descriptor(): + descriptor = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=UnhashableAddress) + + assert descriptor.mapped_class is UnhashableAddress + assert isinstance( + descriptor.from_binary(_udt_binary_value(3), 4), + UnhashableAddress) + + +def test_evict_removes_every_mapping_variant(): + descriptor_a = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=AddressA) + descriptor_b = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=AddressB) + descriptor_unmapped = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES) + + UserType.evict_udt_class(KEYSPACE, UDT_NAME) + + assert UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=AddressA) is not descriptor_a + assert UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=AddressB) is not descriptor_b + assert UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES) is not descriptor_unmapped + + +def test_mapped_variants_do_not_replace_picklable_unmapped_tuple(): + unmapped = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES) + registered_tuple = unmapped.tuple_type + + mapped_a = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=AddressA) + mapped_b = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=AddressB) + + assert mapped_a.tuple_type is None + assert mapped_b.tuple_type is None + assert getattr( + UserType._module, + '{}_{}'.format(KEYSPACE, UDT_NAME)) is registered_tuple + + +def test_legacy_casstype_lookup_does_not_borrow_registered_mapping(): + class KeyspaceType: + + @classmethod + def cass_parameterized_type(cls): + return KEYSPACE + + class UdtNameType: + cassname = UDT_NAME.encode('ascii').hex() + + mapped = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=AddressA) + field_name = FIELD_NAMES[0].encode('ascii').hex() + + parsed = UserType.apply_parameters( + (KeyspaceType, UdtNameType, Int32Type), + (None, None, field_name)) + + assert parsed is not mapped + assert parsed.mapped_class is None + + +def test_concurrent_mapping_creation_and_eviction_is_safe(): + mappings = (AddressA, AddressB, UnhashableAddress, None) + + def create_variants(): + for _ in range(100): + for mapped_class in mappings: + descriptor = UserType.make_udt_class( + KEYSPACE, UDT_NAME, FIELD_NAMES, FIELD_TYPES, + mapped_class=mapped_class) + assert descriptor.mapped_class is mapped_class + + def evict_variants(): + for _ in range(100): + UserType.evict_udt_class(KEYSPACE, UDT_NAME) + + with ThreadPoolExecutor(max_workers=4) as executor: + futures = [ + executor.submit(create_variants), + executor.submit(create_variants), + executor.submit(evict_variants), + executor.submit(evict_variants), + ] + for future in futures: + future.result()