From b4ccfb4a9aba88ba56d0bf718bbe851553c8c712 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 16 Jul 2026 21:51:39 +0200 Subject: [PATCH 1/2] DRIVER-153: negotiate and implement SCYLLA_USE_METADATA_ID extension Implement the SCYLLA_USE_METADATA_ID protocol extension, which backports the CQL v5 prepared-statement metadata-id mechanism to earlier protocol versions. When negotiated, the server includes a hash of the result metadata in the PREPARE response; the driver sends it back with every EXECUTE, allowing the server to omit result metadata from responses (skip_meta) and to report schema changes with METADATA_CHANGED plus fresh metadata, which the driver adopts automatically. protocol_features.py: parse the extension from SUPPORTED, echo it in STARTUP, expose it as ProtocolFeatures.use_metadata_id. protocol.py: ExecuteMessage carries connection-independent request data (skip_meta, result_metadata_id) fixed at construction; serialization decides the wire format from the (protocol_version, protocol_features) that Connection.send_msg supplies for the serving connection: - The metadata-id field is written iff the connection speaks CQL v5+ or negotiated the extension - always, on such connections. An empty sentinel (b'') is written when the statement has no id (prepared before the extension was active, e.g. during a rolling upgrade, or an LWT statement): the sentinel mismatch makes the server respond with METADATA_CHANGED plus the current id and metadata, so such statements acquire an id on their first execution. This also fixes a TypeError on v5 when result_metadata_id was None. - _SKIP_METADATA_FLAG is written only when the SCYLLA_USE_METADATA_ID extension is negotiated on the connection; without the metadata-id mechanism a schema change after PREPARE would leave the driver decoding rows with stale cached metadata. This is deliberately narrower than the metadata-id field above: on native CQL v5 the field is part of the frame layout, but the driver does not request skip there. Upstream never emitted _SKIP_METADATA_FLAG on any version (_write_query_params never wrote it), and enabling the skip optimization for native v5 is a separate change kept out of scope for this Scylla extension. Because messages are immutable after construction, every send path is correct without per-path setup - including the control-connection fallback - and concurrent sends of the same message (speculative executions) cannot race on per-connection state. query.py: PreparedStatement stores (result_metadata, result_metadata_id) as one tuple replaced in a single attribute assignment, read through compatibility properties and updated via update_result_metadata(). Response callbacks update statements while request threads read them; a torn pair (fresh id + stale metadata) would make the server skip sending metadata while rows are decoded against the wrong columns, with no recovery. The compatibility setters are documented as non-atomic relative to each other - update_result_metadata() is the atomic path; the setters exist only for callers assigning the old individual attributes. cluster.py: _create_response_future snapshots the pair once and requests skip_meta only when the statement has both an id and usable cached metadata (result_metadata is None for NO_METADATA/LWT statements and [] for zero-column statements; neither can nor needs to skip metadata). The same snapshot is handed to the ResponseFuture, so a skip_meta response is decoded against the metadata that pairs with the id the message sent - not a later re-read of the statement cache, which a concurrent METADATA_CHANGED could have replaced between construction and send (and which also keeps speculative sends of one message internally consistent). _set_result adopts a METADATA_CHANGED response by replacing the pair atomically; a response carrying a new id without column metadata is ignored with a warning, since adopting the id alone would create the unrecoverable stale-decode state. skip_meta additionally stays off for continuous paging (@dkropachev): Connection.process_msg hardcodes result_metadata=None for every page after the first, since it isn't threaded through the paging session - a skip_meta response has nothing to decode page 2+ against, and would crash on it. _execute_after_prepare refreshes the pair from exactly what the reprepare response carries, including the id (@dkropachev): falling back to the previously cached id when the response has none risks pairing it with metadata from a different schema version than the one that id was computed for - e.g. if the schema changed and then reverted between the two PREPAREs, the old id can become valid again for the current schema while paired locally with an intermediate version's metadata, with no server-side mismatch to catch it. Dropping it instead lets the next id-aware execute re-acquire a correctly paired id through the same b'' sentinel self-healing path a never-prepared statement uses. docs/scylla-specific.rst: documents the extension and its behaviour, worded so the skip_meta optimization reads as conditional on the extension being negotiated rather than pre-existing default behaviour. CHANGELOG.rst: add a Features entry for the extension. --- CHANGELOG.rst | 6 +++ cassandra/cluster.py | 84 ++++++++++++++++++++++++++++++---- cassandra/protocol.py | 49 ++++++++++++++++++-- cassandra/protocol_features.py | 17 ++++++- cassandra/query.py | 56 +++++++++++++++++++++-- docs/scylla-specific.rst | 49 ++++++++++++++++++++ 6 files changed, 242 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index bebb27c82e..87b861454e 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -1,6 +1,12 @@ Unreleased ========== +Features +-------- +* Negotiate and implement the ``SCYLLA_USE_METADATA_ID`` protocol extension: prepared + statements skip re-sending result metadata on EXECUTE, and the driver automatically + refreshes cached metadata when the server detects a schema change (DRIVER-153) + Others ------ * Message serialization now receives the connection's negotiated ``ProtocolFeatures``: diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 12ade2018f..8b8523849d 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -3047,6 +3047,10 @@ def _create_response_future(self, query, parameters, trace, custom_payload, else: timestamp = None + # 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 + if isinstance(query, SimpleStatement): query_string = query.query_string statement_keyspace = query.keyspace if ProtocolVersion.uses_keyspace_flag(self._protocol_version) else None @@ -3058,12 +3062,28 @@ def _create_response_future(self, query, parameters, trace, custom_payload, continuous_paging_options, statement_keyspace) elif isinstance(query, BoundStatement): prepared_statement = query.prepared_statement + # 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 + # id to validate it with and cached metadata to decode against: + # result_metadata is None for statements prepared with NO_METADATA + # (conditional/LWT statements) and [] for statements returning zero + # columns (plain INSERT/UPDATE/DELETE) — neither can nor needs to skip. + # 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 message = ExecuteMessage( prepared_statement.query_id, query.values, cl, serial_cl, fetch_size, paging_state, timestamp, - skip_meta=bool(prepared_statement.result_metadata), + skip_meta=bool(result_metadata) and result_metadata_id is not None + and continuous_paging_options is None, continuous_paging_options=continuous_paging_options, - result_metadata_id=prepared_statement.result_metadata_id) + result_metadata_id=result_metadata_id) elif isinstance(query, BatchStatement): if self._protocol_version < 2: raise UnsupportedOperation( @@ -3090,7 +3110,7 @@ def _create_response_future(self, query, parameters, trace, custom_payload, 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) + continuous_paging_state=None, host=host, bound_result_metadata=bound_result_metadata) def get_execution_profile(self, name): """ @@ -4717,12 +4737,14 @@ class ResponseFuture(object): _host = None _control_connection_query_attempted = False _TABLET_ROUTING_CTYPE = None + _bound_result_metadata = [] _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): + speculative_execution_plan=None, continuous_paging_state=None, host=None, + bound_result_metadata=_NOT_SET): self.session = session # TODO: normalize handling of retry policy and row factory self.row_factory = row_factory or session.row_factory @@ -4733,6 +4755,12 @@ 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 self._callback_lock = Lock() self._start_time = start_time or time.time() self._host = host @@ -4956,7 +4984,7 @@ 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.prepared_statement.result_metadata if self.prepared_statement else [] + result_meta = self._bound_result_metadata if cb is None: cb = partial(self._set_result, host, connection, None) cb = partial(self._handle_control_connection_response, connection, cb) @@ -5010,7 +5038,7 @@ def _query(self, host, message=None, cb=None): else: connection, request_id = pool.borrow_connection(timeout=2.0) self._connection = connection - result_meta = self.prepared_statement.result_metadata if self.prepared_statement else [] + result_meta = self._bound_result_metadata if cb is None: cb = partial(self._set_result, host, connection, pool) @@ -5175,6 +5203,35 @@ def _set_result(self, host, connection, pool, response): self._paging_state = response.paging_state self._col_names = response.column_names self._col_types = response.column_types + new_result_metadata_id = getattr(response, 'result_metadata_id', None) + if self.prepared_statement and new_result_metadata_id is not None: + if response.column_metadata: + # METADATA_CHANGED: replace metadata and its id as one + # atomic pair so a concurrent reader can never pair the + # new id with the old metadata (the server would then + # skip sending metadata and rows would be decoded + # against stale columns, with no recovery). + self.prepared_statement.update_result_metadata( + response.column_metadata, new_result_metadata_id) + # Metadata recovered — re-arm the anomaly warning so a + # later recurrence is logged again. + self.prepared_statement._warned_missing_column_metadata = False + 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 + # create exactly the stale-metadata/fresh-id state + # described above. Keeping the old pair means the next + # EXECUTE sends the old id, the server detects the + # mismatch, and the driver recovers with full metadata. + # Log once per statement (not per execute) while the + # anomaly persists. + self.prepared_statement._warned_missing_column_metadata = True + log.warning( + "Server sent a new result_metadata_id but no column metadata " + "for prepared statement %r. Ignoring both; the cached metadata " + "and id are left unchanged.", + getattr(self.prepared_statement, 'query_id', None) + ) if getattr(self.message, 'continuous_paging_options', None): self._handle_continuous_paging_first_response(connection, response) else: @@ -5325,10 +5382,17 @@ def _execute_after_prepare(self, host, connection, pool, response): expected=hexlify(self.prepared_statement.query_id), got=hexlify(response.query_id) ) )) - self.prepared_statement.result_metadata = response.column_metadata - new_metadata_id = response.result_metadata_id - if new_metadata_id is not None: - self.prepared_statement.result_metadata_id = new_metadata_id + # 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 + # with metadata from a different schema version than the one the + # old id was computed for (e.g. schema changed and reverted + # between the two PREPAREs) - a stale-but-plausible id a later + # 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) # use self._query to re-use the same host and # at the same time properly borrow the connection diff --git a/cassandra/protocol.py b/cassandra/protocol.py index 4360647fb3..e1a67f6a11 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -558,6 +558,14 @@ def __init__(self, query_params, consistency_level, self.skip_meta = skip_meta self.keyspace = keyspace + def _should_skip_metadata(self, protocol_version, protocol_features): + """Whether to set ``_SKIP_METADATA_FLAG`` on this message. + + The base is unconditional (the message's own ``skip_meta``); subclasses + narrow it based on the connection's negotiated features. + """ + return self.skip_meta + def _write_query_params(self, f, protocol_version, protocol_features=None): write_consistency_level(f, self.consistency_level) flags = 0x00 @@ -576,6 +584,9 @@ def _write_query_params(self, f, protocol_version, protocol_features=None): if self.timestamp is not None: flags |= _PROTOCOL_TIMESTAMP_FLAG + if self._should_skip_metadata(protocol_version, protocol_features): + flags |= _SKIP_METADATA_FLAG + if self.keyspace is not None: if ProtocolVersion.uses_keyspace_flag(protocol_version): flags |= _WITH_KEYSPACE_FLAG @@ -638,13 +649,45 @@ def __init__(self, query_id, query_params, consistency_level, super(ExecuteMessage, self).__init__(query_params, consistency_level, serial_consistency_level, fetch_size, paging_state, timestamp, skip_meta, continuous_paging_options) + @staticmethod + def _metadata_id_negotiated(protocol_version, protocol_features): + """Whether the result-metadata-id field is part of this EXECUTE frame. + + It is part of the frame layout whenever the connection speaks CQL v5+ + natively or negotiated SCYLLA_USE_METADATA_ID, so on such connections it + must always be written. + """ + return (ProtocolVersion.uses_prepared_metadata(protocol_version) + or (protocol_features is not None and protocol_features.use_metadata_id)) + + def _should_skip_metadata(self, protocol_version, protocol_features): + """Whether to ask the server to skip sending result metadata. + + Only when the SCYLLA_USE_METADATA_ID extension is negotiated on this + connection. Without the metadata-id mechanism a schema change after + PREPARE would leave the driver decoding rows with stale cached metadata. + + This is deliberately narrower than :meth:`_metadata_id_negotiated`: on + native CQL v5 the metadata-id field is part of the frame layout, but we + do NOT emit ``_SKIP_METADATA_FLAG`` there. Upstream never emitted it on + any version, and turning the skip optimization on for native v5 is a + separate behavior change out of scope for this Scylla extension. + """ + return (self.skip_meta + and protocol_features is not None + and protocol_features.use_metadata_id) + def _write_query_params(self, f, protocol_version, protocol_features=None): super(ExecuteMessage, self)._write_query_params(f, protocol_version, protocol_features) def send_body(self, f, protocol_version, protocol_features=None): write_string(f, self.query_id) - if ProtocolVersion.uses_prepared_metadata(protocol_version): - write_string(f, self.result_metadata_id) + if self._metadata_id_negotiated(protocol_version, protocol_features): + # An empty id is written when the statement has no cached metadata id + # (prepared before the extension was negotiated, e.g. in a mixed + # cluster): the server treats the mismatch as METADATA_CHANGED and + # responds with full metadata plus the current id. + write_string(f, self.result_metadata_id if self.result_metadata_id is not None else b'') self._write_query_params(f, protocol_version, protocol_features) @@ -748,7 +791,7 @@ def decode_row(row): def recv_results_prepared(self, f, protocol_version, protocol_features, user_type_map): self.query_id = read_binary_string(f) - if ProtocolVersion.uses_prepared_metadata(protocol_version): + if ProtocolVersion.uses_prepared_metadata(protocol_version) or protocol_features.use_metadata_id: self.result_metadata_id = read_binary_string(f) else: self.result_metadata_id = None diff --git a/cassandra/protocol_features.py b/cassandra/protocol_features.py index 1bad379208..7165117e80 100644 --- a/cassandra/protocol_features.py +++ b/cassandra/protocol_features.py @@ -10,6 +10,7 @@ LWT_OPTIMIZATION_META_BIT_MASK = "LWT_OPTIMIZATION_META_BIT_MASK" RATE_LIMIT_ERROR_EXTENSION = "SCYLLA_RATE_LIMIT_ERROR" TABLETS_ROUTING_V1 = "TABLETS_ROUTING_V1" +USE_METADATA_ID = "SCYLLA_USE_METADATA_ID" class ProtocolFeatures(object): rate_limit_error = None @@ -17,15 +18,18 @@ class ProtocolFeatures(object): sharding_info = None tablets_routing_v1 = False lwt_info = None + use_metadata_id = False # Keyword-only so that independently developed protocol extensions can add # new fields without conflicting over positional-argument order. - def __init__(self, *, rate_limit_error=None, shard_id=0, sharding_info=None, tablets_routing_v1=False, lwt_info=None): + def __init__(self, *, rate_limit_error=None, shard_id=0, sharding_info=None, tablets_routing_v1=False, lwt_info=None, + use_metadata_id=False): self.rate_limit_error = rate_limit_error self.shard_id = shard_id self.sharding_info = sharding_info self.tablets_routing_v1 = tablets_routing_v1 self.lwt_info = lwt_info + self.use_metadata_id = use_metadata_id @staticmethod def parse_from_supported(supported): @@ -33,8 +37,10 @@ def parse_from_supported(supported): shard_id, sharding_info = ProtocolFeatures.parse_sharding_info(supported) tablets_routing_v1 = ProtocolFeatures.parse_tablets_info(supported) lwt_info = ProtocolFeatures.parse_lwt_info(supported) + use_metadata_id = ProtocolFeatures.parse_use_metadata_id(supported) return ProtocolFeatures(rate_limit_error=rate_limit_error, shard_id=shard_id, sharding_info=sharding_info, - tablets_routing_v1=tablets_routing_v1, lwt_info=lwt_info) + tablets_routing_v1=tablets_routing_v1, lwt_info=lwt_info, + use_metadata_id=use_metadata_id) @staticmethod def maybe_parse_rate_limit_error(supported): @@ -60,6 +66,8 @@ def add_startup_options(self, options): options[TABLETS_ROUTING_V1] = "" if self.lwt_info is not None: options[LWT_ADD_METADATA_MARK] = str(self.lwt_info.lwt_meta_bit_mask) + if self.use_metadata_id: + options[USE_METADATA_ID] = "" @staticmethod def parse_sharding_info(options): @@ -84,6 +92,11 @@ def parse_sharding_info(options): def parse_tablets_info(options): return TABLETS_ROUTING_V1 in options + @staticmethod + def parse_use_metadata_id(options): + """Return True if the ``SCYLLA_USE_METADATA_ID`` extension is advertised in ``options``.""" + return USE_METADATA_ID in options + @staticmethod def parse_lwt_info(options): value_list = options.get(LWT_ADD_METADATA_MARK, [None]) diff --git a/cassandra/query.py b/cassandra/query.py index 6c6878fdb4..97b1959b96 100644 --- a/cassandra/query.py +++ b/cassandra/query.py @@ -451,13 +451,16 @@ class PreparedStatement(object): protocol_version = None query_id = None query_string = None - result_metadata = None - result_metadata_id = None + _result_metadata_and_id = (None, None) column_encryption_policy = None routing_key_indexes = None _routing_key_index_set = None serial_consistency_level = None # TODO never used? _is_lwt = False + # Set once we've logged the "new metadata id without column metadata" anomaly + # for this statement, to avoid logging it on every execute while a misbehaving + # server keeps returning it. Re-armed whenever the metadata is updated. + _warned_missing_column_metadata = False def __init__(self, column_metadata, query_id, routing_key_indexes, query, keyspace, protocol_version, result_metadata, result_metadata_id, @@ -468,12 +471,57 @@ def __init__(self, column_metadata, query_id, routing_key_indexes, query, self.query_string = query self.keyspace = keyspace self.protocol_version = protocol_version - self.result_metadata = result_metadata - self.result_metadata_id = result_metadata_id + self._result_metadata_and_id = (result_metadata, result_metadata_id) self.column_encryption_policy = column_encryption_policy self.is_idempotent = False self._is_lwt = is_lwt + @property + def result_metadata_and_id(self): + """ + The cached result metadata and its metadata id as one immutable + ``(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. + """ + return self._result_metadata_and_id + + @property + def result_metadata(self): + """Cached result metadata (column definitions) from PREPARE; see :attr:`result_metadata_and_id`.""" + return self._result_metadata_and_id[0] + + @result_metadata.setter + def result_metadata(self, metadata): + """Compatibility setter. Not atomic relative to a separate ``result_metadata_id`` + assignment — prefer :meth:`update_result_metadata` to replace both together.""" + self.update_result_metadata(metadata, self._result_metadata_and_id[1]) + + @property + def result_metadata_id(self): + """Cached result metadata id (hash) from PREPARE; see :attr:`result_metadata_and_id`.""" + return self._result_metadata_and_id[1] + + @result_metadata_id.setter + def result_metadata_id(self, metadata_id): + """Compatibility setter. Not atomic relative to a separate ``result_metadata`` + assignment — prefer :meth:`update_result_metadata` to replace both together.""" + self.update_result_metadata(self._result_metadata_and_id[0], metadata_id) + + def update_result_metadata(self, result_metadata, result_metadata_id): + """ + Replace the cached result metadata and metadata id together, in a single + atomic attribute store. Response callbacks may update a statement while + request threads read it; updating the pair in one step (rather than the + two fields separately) prevents a reader from pairing a fresh metadata id + with stale metadata — a state in which the server would skip sending + metadata and rows would be decoded against the wrong columns. + """ + self._result_metadata_and_id = (result_metadata, result_metadata_id) + @classmethod def from_message(cls, query_id, column_metadata, pk_indexes, cluster_metadata, query, prepared_keyspace, protocol_version, result_metadata, diff --git a/docs/scylla-specific.rst b/docs/scylla-specific.rst index 4b28781f1c..f6185a2a13 100644 --- a/docs/scylla-specific.rst +++ b/docs/scylla-specific.rst @@ -156,3 +156,52 @@ https://github.com/scylladb/scylladb/blob/master/docs/dev/protocol-extensions.md Details on the sending tablet information to the drivers https://github.com/scylladb/scylladb/blob/master/docs/dev/protocol-extensions.md#sending-tablet-info-to-the-drivers + + +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. + +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 +negotiated, the server includes a hash of the result metadata in the ``PREPARE`` +response. The driver sends this hash back with every ``EXECUTE`` request. If the +schema has changed, the server sets the ``METADATA_CHANGED`` flag and returns the +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. + +**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`` (or speak CQL v5) — + the latter is decided per connection when the request is serialized. +- 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. +- 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 + hash, the server detects the mismatch and responds with the current hash and + full metadata, and the driver caches both. Subsequent executions get the + ``skip_meta`` optimization — no re-prepare or client restart is needed. + +**Current scope:** the optimization applies to any prepared statement whose +``PREPARE`` response includes non-empty 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. + +For full protocol details see the ScyllaDB CQL protocol extensions documentation: +https://github.com/scylladb/scylladb/blob/master/docs/dev/protocol-extensions.md From 8a25e775602c96d77ba01267f3060da32f3bb47a Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Tue, 21 Jul 2026 15:26:00 +0200 Subject: [PATCH 2/2] DRIVER-153: tests for SCYLLA_USE_METADATA_ID extension Unit tests for the extension across its layers: test_protocol_features.py: SCYLLA_USE_METADATA_ID parsed from SUPPORTED and echoed in STARTUP options; absent by default. test_protocol.py (wire format): - metadata-id field written on v4 iff the connection negotiated the extension, with the exact bytes asserted; empty sentinel (b'') when the statement has no id, on both the extension path (v4) and the v5 native path (previously a TypeError); - _SKIP_METADATA_FLAG written when skip_meta is requested and the SCYLLA_USE_METADATA_ID extension is negotiated (v4 or v5), and NOT set on a native v5 connection without the extension (the id field is still written there, but the driver does not request skip); also suppressed - together with the id field - on a v4 connection without the extension, even when the statement carries an id; - PREPARED response decoding reads result_metadata_id iff the extension was negotiated (or v5); METADATA_CHANGED/NO_METADATA flag handling. test_query.py: PreparedStatement stores the (result_metadata, result_metadata_id) pair atomically - constructor, update_result_metadata, and the backwards-compatible single-attribute setters all replace the pair as one unit, and previously-taken snapshots stay internally consistent. test_response_future.py: - _create_response_future builds ExecuteMessage from a single pair snapshot: skip_meta only with both an id and usable cached metadata; disabled for id-less statements, NO_METADATA/LWT statements (result_metadata None) and zero-column statements (result_metadata []), while the id still rides on the message; - _query sends the message exactly as constructed (no per-connection mutation - regression test for the speculative-execution race) and decodes a skip_meta response against the metadata snapshotted when the message was built, not a later read of the statement cache (regression for a concurrent METADATA_CHANGED racing the send); - _set_result METADATA_CHANGED path replaces the cached pair atomically; a response with a new id but no column metadata (empty or absent) is ignored with a warning, leaving the cached pair unchanged - adopting the id alone would poison the cache with a stale-metadata/current-id pair the server would never refresh; - _execute_after_prepare refreshes the pair from exactly what the reprepare response carries, including the id, and no longer keeps the previous id when the response has none (@dkropachev: doing so risked pairing a stale id with metadata from a different schema version - test_execute_after_prepare_no_metadata_id_in_response_clears_id); - a statement with valid cached metadata+id must still get skip_meta=False when continuous_paging_options is set (@dkropachev: Connection.process_msg hardcodes result_metadata=None for paging-session pages after the first, so a skip_meta response would crash decoding them - test_create_execute_message_continuous_paging_disables_skip_meta). tests/integration/standard/test_scylla_metadata_id.py: live-server coverage against a real Scylla node via CCM, closing the one gap unit tests can't - whether Scylla actually treats the empty result_metadata_id sentinel as a mismatch rather than a protocol error. Confirms extension negotiation, the normal METADATA_CHANGED-after-ALTER-TABLE path, and the sentinel round trip: a statement forced back to result_metadata_id=None (simulating one prepared before the extension was known, e.g. mid rolling-upgrade) executes without error and comes back with a fresh id. Mirrors the equivalent live test already merged in the Java driver (scylladb/java-driver#758, should_handle_empty_metadata_id_when_executing_statement_when_supported). Run locally against Scylla 2026.1.9 via CCM; see PR description for setup and log excerpt. --- cassandra/cluster.py | 2 +- .../standard/test_scylla_metadata_id.py | 110 +++++ tests/unit/test_protocol.py | 257 +++++++++- tests/unit/test_protocol_features.py | 35 ++ tests/unit/test_query.py | 48 ++ tests/unit/test_response_future.py | 462 +++++++++++++++++- 6 files changed, 906 insertions(+), 8 deletions(-) create mode 100644 tests/integration/standard/test_scylla_metadata_id.py diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 8b8523849d..accbc91c99 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -4737,7 +4737,7 @@ class ResponseFuture(object): _host = None _control_connection_query_attempted = False _TABLET_ROUTING_CTYPE = None - _bound_result_metadata = [] + _bound_result_metadata = None _warned_timeout = False diff --git a/tests/integration/standard/test_scylla_metadata_id.py b/tests/integration/standard/test_scylla_metadata_id.py new file mode 100644 index 0000000000..70a3699d35 --- /dev/null +++ b/tests/integration/standard/test_scylla_metadata_id.py @@ -0,0 +1,110 @@ +# Copyright 2026 ScyllaDB, 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 unittest + +import pytest + +from tests.integration import use_singledc, SCYLLA_VERSION, BasicSharedKeyspaceUnitTestCase + +pytestmark = pytest.mark.skipif(SCYLLA_VERSION is None, reason="SCYLLA_USE_METADATA_ID is a Scylla-only protocol extension") + + +def setup_module(): + use_singledc() + + +class ScyllaMetadataIdTests(BasicSharedKeyspaceUnitTestCase): + """ + Live-server coverage for the SCYLLA_USE_METADATA_ID protocol extension (DRIVER-153). + """ + + @classmethod + def setUpClass(cls): + cls.common_setup(1) + # Skip the whole class if this Scylla build does not advertise the + # extension (e.g. a version predating scylladb#23292). Without this the + # tests below would error out instead of skipping on an unsupporting node. + pool = next(iter(cls.session.get_pools())) + connection, _ = pool.borrow_connection(timeout=2) + try: + if not connection.features.use_metadata_id: + raise unittest.SkipTest( + "Scylla node does not advertise SCYLLA_USE_METADATA_ID") + finally: + pool.return_connection(connection) + + def setUp(self): + self.table_name = "{}.{}".format(self.keyspace_name, self.function_table_name) + self.session.execute("CREATE TABLE {} (a int PRIMARY KEY, b int, c int)".format(self.table_name)) + self.session.execute("INSERT INTO {} (a, b, c) VALUES (1, 1, 1)".format(self.table_name)) + + def tearDown(self): + self.session.execute("DROP TABLE {}".format(self.table_name)) + + def test_extension_is_negotiated(self): + """ + Sanity check that SCYLLA_USE_METADATA_ID was actually negotiated on this + connection. Without this, the tests below could pass vacuously if + negotiation silently failed. + """ + pool = next(iter(self.session.get_pools())) + connection, _ = pool.borrow_connection(timeout=2) + try: + assert connection.features.use_metadata_id is True + finally: + pool.return_connection(connection) + + def test_metadata_changed_recovers_after_schema_change(self): + """ + Normal METADATA_CHANGED path: after ALTER TABLE, the next EXECUTE must + come back with a fresh result_metadata_id and updated column metadata, + picked up automatically without re-preparing. + """ + prepared = self.session.prepare("SELECT * FROM {} WHERE a = ?".format(self.table_name)) + id_before = prepared.result_metadata_id + assert id_before is not None + assert len(prepared.result_metadata) == 3 + + self.session.execute(prepared.bind((1,))) + + self.session.execute("ALTER TABLE {} ADD d int".format(self.table_name)) + self.session.execute(prepared.bind((1,))) + + assert prepared.result_metadata_id is not None + assert prepared.result_metadata_id != id_before + assert len(prepared.result_metadata) == 4 + + def test_empty_sentinel_id_triggers_metadata_changed(self): + """ + Statements prepared before the extension was negotiated (e.g. mid rolling + upgrade) start with result_metadata_id=None and must send the empty b'' + sentinel on their first EXECUTE. This must not be treated as a protocol + error by the server: it must be treated as a mismatch, causing Scylla to + respond with METADATA_CHANGED (fresh id + full metadata), which the + driver then caches. + """ + prepared = self.session.prepare("SELECT * FROM {} WHERE a = ?".format(self.table_name)) + assert prepared.result_metadata_id is not None + + # Simulate "prepared before the extension was known" by dropping the + # cached id while keeping the cached metadata (mirrors the java-driver's + # should_handle_empty_metadata_id_when_executing_statement_when_supported). + prepared.update_result_metadata(prepared.result_metadata, None) + assert prepared.result_metadata_id is None + + result = self.session.execute(prepared.bind((1,))) + + assert list(result) == [(1, 1, 1)] + assert prepared.result_metadata_id is not None diff --git a/tests/unit/test_protocol.py b/tests/unit/test_protocol.py index db6c37abda..75dc69bca5 100644 --- a/tests/unit/test_protocol.py +++ b/tests/unit/test_protocol.py @@ -12,15 +12,19 @@ # See the License for the specific language governing permissions and # limitations under the License. +import io +import struct import unittest +from typing import ClassVar from unittest.mock import Mock from cassandra import ConsistencyLevel, ProtocolVersion, UnsupportedOperation from cassandra.protocol import ( - PrepareMessage, QueryMessage, ExecuteMessage, UnsupportedOperation, + PrepareMessage, QueryMessage, ExecuteMessage, BatchMessage, StartupMessage, OptionsMessage, RegisterMessage, - AuthResponseMessage, ProtocolHandler, _MessageType + AuthResponseMessage, ProtocolHandler, _MessageType, + ResultMessage, RESULT_KIND_ROWS ) from cassandra.protocol_features import ProtocolFeatures from cassandra.query import BatchType @@ -66,6 +70,253 @@ def test_execute_message(self): (b'\x00\x04',), (b'\x00\x00\x00\x01',), (b'\x00\x00',)]) + def test_execute_message_skip_meta_flag_with_extension(self): + """ + skip_meta=True must set _SKIP_METADATA_FLAG (0x02) in the flags byte when + the connection negotiated SCYLLA_USE_METADATA_ID, and the metadata id + field must be written on the wire. + """ + message = ExecuteMessage('1', [], 4, skip_meta=True, result_metadata_id=b'foo') + mock_io = Mock() + + message.send_body(mock_io, 4, ProtocolFeatures(use_metadata_id=True)) + # flags byte should be VALUES_FLAG | SKIP_METADATA_FLAG = 0x01 | 0x02 = 0x03 + self._check_calls(mock_io, [(b'\x00\x01',), (b'1',), + (b'\x00\x03',), (b'foo',), + (b'\x00\x04',), (b'\x03',), (b'\x00\x00',)]) + + def test_execute_message_skip_meta_suppressed_without_extension(self): + """ + skip_meta=True must NOT reach the wire on a pre-v5 connection that did not + negotiate SCYLLA_USE_METADATA_ID: without the metadata-id mechanism, a + schema change after PREPARE would leave the driver decoding rows with + stale cached metadata. The metadata id field must not be written either. + """ + message = ExecuteMessage('1', [], 4, skip_meta=True, result_metadata_id=b'foo') + mock_io = Mock() + + message.send_body(mock_io, 4) + # flags byte contains only VALUES_FLAG; no metadata id field + self._check_calls(mock_io, [(b'\x00\x01',), (b'1',), (b'\x00\x04',), (b'\x01',), (b'\x00\x00',)]) + + def test_execute_message_v5_native_skip_meta_not_set(self): + """ + On a native protocol v5 connection (no Scylla extension), skip_meta=True must + NOT set _SKIP_METADATA_FLAG. Upstream never emitted the flag on any version, and + this PR keeps native v5 byte-identical to upstream — enabling skip on native v5 is + a separate, out-of-scope behavior change. The metadata id field is still written + (it is part of the v5 EXECUTE frame layout), so only VALUES_FLAG is set. + """ + message = ExecuteMessage('1', [], 4, skip_meta=True) + mock_io = Mock() + + message.send_body(mock_io, 5) + # v5 wire layout: + # query_id: short(1) + b'1' + # result_metadata_id: short(0) + b'' (sentinel — None on init) + # consistency: short(4) = ONE + # flags (4-byte int): VALUES_FLAG(0x01) only — skip is NOT set on native v5 + # param count: short(0) + self._check_calls(mock_io, [ + (b'\x00\x01',), (b'1',), + (b'\x00\x00',), (b'',), + (b'\x00\x04',), + (b'\x00\x00\x00\x01',), (b'\x00\x00',), + ]) + + def test_execute_message_v5_with_extension_sets_skip_flag(self): + """ + skip is extension-driven, not version-driven: on a v5 connection that ALSO + negotiated SCYLLA_USE_METADATA_ID, skip_meta=True does set _SKIP_METADATA_FLAG. + This also confirms _SKIP_METADATA_FLAG actually reaches the wire (it was dead code + upstream) whenever the extension gates it on. + """ + message = ExecuteMessage('1', [], 4, skip_meta=True) + mock_io = Mock() + + message.send_body(mock_io, 5, ProtocolFeatures(use_metadata_id=True)) + # flags (4-byte int): VALUES_FLAG(0x01) | SKIP_METADATA_FLAG(0x02) = 0x03 + self._check_calls(mock_io, [ + (b'\x00\x01',), (b'1',), + (b'\x00\x00',), (b'',), + (b'\x00\x04',), + (b'\x00\x00\x00\x03',), (b'\x00\x00',), + ]) + + def test_execute_message_scylla_metadata_id_v4(self): + """result_metadata_id should be written on protocol v4 when the connection negotiated the Scylla extension.""" + message = ExecuteMessage('1', [], 4, result_metadata_id=b'foo') + mock_io = Mock() + + message.send_body(mock_io, 4, ProtocolFeatures(use_metadata_id=True)) + # metadata_id written before query params (same position as v5) + self._check_calls(mock_io, [(b'\x00\x01',), (b'1',), + (b'\x00\x03',), (b'foo',), + (b'\x00\x04',), (b'\x01',), (b'\x00\x00',)]) + + def test_execute_message_scylla_metadata_id_none_writes_sentinel(self): + """ + When the connection negotiated the extension but result_metadata_id is None + (e.g. LWT statement or mixed cluster), send_body must still write the field + as an empty string sentinel (\\x00\\x00) so the frame layout matches what + the server expects. + """ + message = ExecuteMessage('1', [], 4) + # result_metadata_id intentionally left as None + mock_io = Mock() + + message.send_body(mock_io, 4, ProtocolFeatures(use_metadata_id=True)) + # empty sentinel: \x00\x00 (zero-length short) + b'' (zero bytes), then normal query params + self._check_calls(mock_io, [(b'\x00\x01',), (b'1',), + (b'\x00\x00',), (b'',), + (b'\x00\x04',), (b'\x01',), (b'\x00\x00',)]) + + def test_execute_message_v5_metadata_id_none_writes_sentinel(self): + """ + On protocol v5, result_metadata_id is always written (uses_prepared_metadata). + When result_metadata_id is None (e.g. LWT statement or mixed cluster where the + statement was prepared before the extension was active), send_body must write an + empty sentinel instead of crashing with TypeError. + """ + message = ExecuteMessage('1', [], 4) + # result_metadata_id intentionally left as None; use_metadata_id stays False (v5 native path) + mock_io = Mock() + + message.send_body(mock_io, 5) + # v5 always writes metadata_id: None → empty sentinel \x00\x00 + b'', then query params + # v5 uses 4-byte flags: VALUES_FLAG = \x00\x00\x00\x01 + self._check_calls(mock_io, [(b'\x00\x01',), (b'1',), + (b'\x00\x00',), (b'',), + (b'\x00\x04',), + (b'\x00\x00\x00\x01',), (b'\x00\x00',)]) + + def test_recv_results_prepared_scylla_extension_reads_metadata_id(self): + """ + When use_metadata_id is True (Scylla extension), result_metadata_id must be + read from the PREPARE response even for protocol v4. + """ + # Build a minimal valid PREPARE response binary (no bind/result columns): + # query_id: short(2) + b'ab' + # result_metadata_id: short(3) + b'xyz' <-- only present when extension active + # prepared flags: int(1) = global_tables_spec + # colcount: int(0) + # num_pk_indexes: int(0) + # ksname: short(2) + b'ks' + # cfname: short(2) + b'tb' + # result flags: int(4) = no_metadata + # result colcount: int(0) + buf = io.BytesIO( + struct.pack('>H', 2) + b'ab' # query_id + + struct.pack('>H', 3) + b'xyz' # result_metadata_id + + struct.pack('>i', 1) # prepared flags: global_tables_spec + + struct.pack('>i', 0) # colcount = 0 + + struct.pack('>i', 0) # num_pk_indexes = 0 + + struct.pack('>H', 2) + b'ks' # ksname + + struct.pack('>H', 2) + b'tb' # cfname + + struct.pack('>i', 4) # result flags: no_metadata + + struct.pack('>i', 0) # result colcount = 0 + ) + + features_with_extension = ProtocolFeatures(use_metadata_id=True) + msg = ResultMessage(kind=4) # RESULT_KIND_PREPARED = 4 + msg.recv_results_prepared(buf, protocol_version=4, + protocol_features=features_with_extension, + user_type_map={}) + assert msg.query_id == b'ab' + assert msg.result_metadata_id == b'xyz' + + def test_recv_results_prepared_no_extension_skips_metadata_id(self): + """ + Without use_metadata_id, result_metadata_id must NOT be read on protocol v4. + The buffer must NOT contain a metadata_id field. + """ + buf = io.BytesIO( + struct.pack('>H', 2) + b'ab' # query_id + # no result_metadata_id + + struct.pack('>i', 1) # prepared flags: global_tables_spec + + struct.pack('>i', 0) # colcount = 0 + + struct.pack('>i', 0) # num_pk_indexes = 0 + + struct.pack('>H', 2) + b'ks' # ksname + + struct.pack('>H', 2) + b'tb' # cfname + + struct.pack('>i', 4) # result flags: no_metadata + + struct.pack('>i', 0) # result colcount = 0 + ) + + features_without_extension = ProtocolFeatures(use_metadata_id=False) + msg = ResultMessage(kind=4) + msg.recv_results_prepared(buf, protocol_version=4, + protocol_features=features_without_extension, + user_type_map={}) + assert msg.query_id == b'ab' + assert msg.result_metadata_id is None + + def test_recv_results_prepared_v5_reads_metadata_id(self): + """ + On protocol v5, ProtocolVersion.uses_prepared_metadata() is True, so + result_metadata_id must be read from the PREPARE response even when + use_metadata_id is False (native v5 path, not the Scylla extension). + """ + buf = io.BytesIO( + struct.pack('>H', 2) + b'ab' # query_id + + struct.pack('>H', 3) + b'xyz' # result_metadata_id (always present on v5) + + struct.pack('>i', 1) # prepared flags: global_tables_spec + + struct.pack('>i', 0) # colcount = 0 + + struct.pack('>i', 0) # num_pk_indexes = 0 + + struct.pack('>H', 2) + b'ks' # ksname + + struct.pack('>H', 2) + b'tb' # cfname + + struct.pack('>i', 4) # result flags: no_metadata + + struct.pack('>i', 0) # result colcount = 0 + ) + + features_no_extension = ProtocolFeatures(use_metadata_id=False) + msg = ResultMessage(kind=4) # RESULT_KIND_PREPARED = 4 + msg.recv_results_prepared(buf, protocol_version=5, + protocol_features=features_no_extension, + user_type_map={}) + assert msg.query_id == b'ab' + assert msg.result_metadata_id == b'xyz' + + def test_recv_results_metadata_reads_metadata_id_on_change(self): + """ + When _METADATA_ID_FLAG (0x0008) is set in a ROWS result, + recv_results_metadata must read and store the new result_metadata_id + sent by the server (METADATA_CHANGED signal), and still populate + column_metadata normally. + """ + # Wire layout for a ROWS result with METADATA_CHANGED: + # flags: int(0x0008) = _METADATA_ID_FLAG + # colcount: int(0) + # result_metadata_id: short(4) + b'new1' + # (no columns — colcount=0 — to keep the buffer minimal) + buf = io.BytesIO( + struct.pack('>i', 0x0008) # flags: METADATA_ID_FLAG + + struct.pack('>i', 0) # colcount = 0 + + struct.pack('>H', 4) + b'new1' # result_metadata_id = b'new1' + ) + msg = ResultMessage(kind=RESULT_KIND_ROWS) + msg.recv_results_metadata(buf, user_type_map={}) + assert msg.result_metadata_id == b'new1' + assert msg.column_metadata == [] + + def test_recv_results_metadata_no_metadata_flag_skips_metadata_id(self): + """ + When _NO_METADATA_FLAG (0x0004) is set, recv_results_metadata returns + early and must NOT read or set result_metadata_id, even if the caller + mistakenly sets _METADATA_ID_FLAG alongside it. + """ + # flags = _NO_METADATA_FLAG (0x0004), colcount = 0 + buf = io.BytesIO( + struct.pack('>i', 0x0004) # flags: NO_METADATA + + struct.pack('>i', 0) # colcount = 0 + ) + msg = ResultMessage(kind=RESULT_KIND_ROWS) + msg.recv_results_metadata(buf, user_type_map={}) + # 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.column_metadata is None + def test_query_message(self): """ Test to check the appropriate calls are made @@ -237,7 +488,7 @@ class FrameByteIdentityTest(unittest.TestCase): The expected frames below were captured from the pre-change encoder. """ - EXPECTED_FRAMES = { + EXPECTED_FRAMES: ClassVar[dict] = { 'startup_v4': '0400000701000000160001000b43514c5f56455253494f4e0005332e342e35', 'options_v4': '040000070500000000', 'register_v4': '040000070b000000220002000f544f504f4c4f47595f4348414e4745000d5354415455535f4348414e4745', diff --git a/tests/unit/test_protocol_features.py b/tests/unit/test_protocol_features.py index 895c384f7e..387583680b 100644 --- a/tests/unit/test_protocol_features.py +++ b/tests/unit/test_protocol_features.py @@ -22,3 +22,38 @@ class OptionsHolder(object): assert protocol_features.rate_limit_error == 123 assert protocol_features.shard_id == 0 assert protocol_features.sharding_info is None + + def test_use_metadata_id_parsing(self): + """ + Test that SCYLLA_USE_METADATA_ID is parsed from SUPPORTED options. + """ + options = {'SCYLLA_USE_METADATA_ID': ['']} + protocol_features = ProtocolFeatures.parse_from_supported(options) + assert protocol_features.use_metadata_id is True + + def test_use_metadata_id_missing(self): + """ + Test that use_metadata_id is False when SCYLLA_USE_METADATA_ID is absent. + """ + options = {'SCYLLA_RATE_LIMIT_ERROR': ['ERROR_CODE=1']} + protocol_features = ProtocolFeatures.parse_from_supported(options) + assert protocol_features.use_metadata_id is False + + def test_use_metadata_id_startup_options(self): + """ + Test that SCYLLA_USE_METADATA_ID is included in STARTUP options when negotiated. + """ + options = {'SCYLLA_USE_METADATA_ID': ['']} + protocol_features = ProtocolFeatures.parse_from_supported(options) + startup = {} + protocol_features.add_startup_options(startup) + assert 'SCYLLA_USE_METADATA_ID' in startup + + def test_use_metadata_id_not_in_startup_when_not_negotiated(self): + """ + Test that SCYLLA_USE_METADATA_ID is NOT included in STARTUP when not negotiated. + """ + protocol_features = ProtocolFeatures.parse_from_supported({}) + startup = {} + protocol_features.add_startup_options(startup) + assert 'SCYLLA_USE_METADATA_ID' not in startup diff --git a/tests/unit/test_query.py b/tests/unit/test_query.py index 6b0ebe690e..fcecfcd056 100644 --- a/tests/unit/test_query.py +++ b/tests/unit/test_query.py @@ -115,3 +115,51 @@ def is_lwt(self): batch_with_simple = BatchStatement() batch_with_simple.add(LwtSimpleStatement()) assert batch_with_simple.is_lwt() is True + + +class PreparedStatementMetadataPairTest(unittest.TestCase): + """ + result_metadata and result_metadata_id are stored as one tuple replaced in a + single attribute assignment: response callbacks update a statement while + request threads read it, and a torn pair (fresh id + stale metadata) would + make the server skip sending metadata while rows are decoded against the + wrong columns. + """ + + @staticmethod + def _make_statement(result_metadata, result_metadata_id): + return PreparedStatement( + column_metadata=[], query_id=b'qid', routing_key_indexes=None, + query="SELECT * FROM foo", keyspace='ks', protocol_version=4, + result_metadata=result_metadata, result_metadata_id=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_id == b'hash' + assert ps.result_metadata_and_id == (meta, b'hash') + + def test_update_replaces_pair_atomically(self): + ps = self._make_statement([('ks', 'tb', 'old', None)], b'old') + snapshot_before = ps.result_metadata_and_id + + new_meta = [('ks', 'tb', 'new', None)] + ps.update_result_metadata(new_meta, b'new') + + # a snapshot taken before the update stays internally consistent + assert snapshot_before == ([('ks', 'tb', 'old', None)], b'old') + assert ps.result_metadata_and_id == (new_meta, b'new') + + def test_individual_setters_keep_pair_consistent(self): + # backwards-compatible attribute assignment still works and replaces + # the whole pair underneath + meta = [('ks', 'tb', 'col', None)] + ps = self._make_statement(meta, b'hash') + + ps.result_metadata_id = b'other' + assert ps.result_metadata_and_id == (meta, b'other') + + new_meta = [] + ps.result_metadata = new_meta + assert ps.result_metadata_and_id == (new_meta, b'other') diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index 9673b0d634..48473930a7 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -23,6 +23,7 @@ from cassandra.connection import Connection, ConnectionException from cassandra.protocol import (ReadTimeoutErrorMessage, WriteTimeoutErrorMessage, UnavailableErrorMessage, ResultMessage, QueryMessage, + ExecuteMessage, OverloadedErrorMessage, IsBootstrappingErrorMessage, PreparedQueryNotFound, PrepareMessage, ServerError, RESULT_KIND_ROWS, RESULT_KIND_SET_KEYSPACE, @@ -30,7 +31,7 @@ ProtocolHandler) from cassandra.policies import RetryPolicy, ExponentialBackoffRetryPolicy from cassandra.pool import NoConnectionsAvailable -from cassandra.query import SimpleStatement +from cassandra.query import SimpleStatement, PreparedStatement, BoundStatement from tests.util import assertEqual, assertIsInstance import pytest @@ -911,7 +912,7 @@ def test_repeat_orig_query_after_succesful_reprepare(self): response = Mock(spec=ResultMessage, kind=RESULT_KIND_PREPARED, - result_metadata_id='foo') + result_metadata_id=b'foo') response.results = (None, None, None, None, None) response.query_id = query_id @@ -919,11 +920,79 @@ def test_repeat_orig_query_after_succesful_reprepare(self): rf._execute_after_prepare('host', None, None, response) rf._query.assert_called_once_with('host') - rf.prepared_statement = Mock() - rf.prepared_statement.query_id = query_id + rf.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_id=None) rf._query = Mock(return_value=True) rf._execute_after_prepare('host', None, None, response) rf._query.assert_called_once_with('host') + assert rf.prepared_statement.result_metadata_id == b'foo' + + def test_execute_after_prepare_updates_result_metadata_id(self): + """ + After a PreparedQueryNotFound triggers a reprepare, _execute_after_prepare + must update both prepared_statement.result_metadata and + prepared_statement.result_metadata_id when the PREPARE response carries a + new metadata id. Deleting those update lines must break this test. + """ + query_id = b'reprepare_qid' + session = self.make_session() + rf = self.make_response_future(session) + + new_meta = [('ks', 'tb', 'new_col', Mock())] + response = Mock(spec=ResultMessage, + kind=RESULT_KIND_PREPARED, + result_metadata_id=b'new_hash', + column_metadata=new_meta) + response.query_id = query_id + + rf.prepared_statement = self._make_prepared_statement( + [('ks', 'tb', 'old_col', Mock())], b'old_hash', query_id=query_id) + + rf._query = Mock(return_value=True) + 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_id == b'new_hash' + assert rf.prepared_statement.result_metadata_and_id == (new_meta, b'new_hash') + rf._query.assert_called_once_with('host') + + def test_execute_after_prepare_no_metadata_id_in_response_clears_id(self): + """ + When the PREPARE response does not carry a result_metadata_id (e.g. the + extension is not active on the reprepare connection), _execute_after_prepare + must clear the cached result_metadata_id rather than keep the previous one: + carrying it forward could pair a stale id with the freshly reprepared column + metadata (e.g. if the schema changed and reverted between the two PREPAREs, + the old id could become valid again for the current schema while paired + locally with an intermediate schema's metadata, with no server-side mismatch + to catch it). Clearing it instead lets the next id-aware execute re-acquire a + correctly paired id via the same b'' sentinel / METADATA_CHANGED self-healing + path a never-prepared statement uses. + """ + query_id = b'reprepare_qid2' + session = self.make_session() + rf = self.make_response_future(session) + + new_meta = [('ks', 'tb', 'col', Mock())] + response = Mock(spec=ResultMessage, + kind=RESULT_KIND_PREPARED, + result_metadata_id=None, + column_metadata=new_meta) + response.query_id = query_id + + rf.prepared_statement = self._make_prepared_statement( + [('ks', 'tb', 'old_col', Mock())], b'old_hash', query_id=query_id) + + rf._query = Mock(return_value=True) + rf._execute_after_prepare('host', None, None, response) + + # 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_id is None def test_timeout_does_not_release_stream_id(self): """ @@ -1008,3 +1077,388 @@ def test_single_host_query_plan_exhausted_after_one_retry(self): # Instead, it should set a NoHostAvailable exception assert rf._final_exception is not None assert isinstance(rf._final_exception, NoHostAvailable) + + # ------------------------------------------------------------------------- + # Helpers for SCYLLA_USE_METADATA_ID tests + # ------------------------------------------------------------------------- + + def _make_rows_response(self, result_metadata_id=None, column_metadata=None): + """ + Return a real ResultMessage(kind=RESULT_KIND_ROWS) with all attributes + that _set_result accesses pre-set, so it passes isinstance checks and + doesn't trigger unexpected code paths. + """ + response = ResultMessage(kind=RESULT_KIND_ROWS) + response.paging_state = None + response.column_names = ['col'] + response.parsed_rows = [] + response.column_types = [] + response.column_metadata = column_metadata + response.result_metadata_id = result_metadata_id + response.trace_id = None + response.warnings = None + response.custom_payload = None + return response + + def _make_prepared_statement(self, result_metadata, result_metadata_id, query_id=b'qid'): + return 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) + + 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) + 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, + ) + 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): + """ + 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() + profile = session._maybe_get_execution_profile.return_value + profile.consistency_level = ConsistencyLevel.ONE + profile.serial_consistency_level = None + profile.continuous_paging_options = continuous_paging_options + profile.speculative_execution_policy = None + profile.load_balancing_policy.make_query_plan.return_value = ['ip1'] + session.default_fetch_size = 5000 + session.use_client_timestamp = False + bound = BoundStatement(prepared_statement).bind(()) + return Session._create_response_future( + session, bound, parameters=None, trace=False, custom_payload=None, timeout=1) + + # ------------------------------------------------------------------------- + # _set_result: METADATA_CHANGED update path + # ------------------------------------------------------------------------- + + def test_set_result_updates_metadata_when_metadata_changed(self): + """ + When the EXECUTE response carries a new result_metadata_id (server + detected a schema change), _set_result must update both + prepared_statement.result_metadata and prepared_statement.result_metadata_id. + """ + session = self.make_session() + pool = session._pools.get.return_value + connection = Mock(spec=Connection) + connection.protocol_version = 4 + connection.features = Mock() + connection.features.use_metadata_id = False + pool.borrow_connection.return_value = (connection, 1) + + old_meta = [('ks', 'tb', 'old_col', Mock())] + new_meta = [('ks', 'tb', 'new_col', Mock())] + ps = self._make_prepared_statement(old_meta, b'old_id') + + rf = self.make_response_future(session) + rf.prepared_statement = ps + rf.send_request() + + response = self._make_rows_response( + result_metadata_id=b'new_id', + column_metadata=new_meta, + ) + rf._set_result(None, None, None, response) + + assert ps.result_metadata is 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') + + def test_set_result_does_not_update_metadata_when_metadata_id_absent(self): + """ + When the EXECUTE response has no result_metadata_id (normal skip-meta + path — server metadata unchanged), _set_result must leave the + prepared_statement's cached metadata untouched. + """ + session = self.make_session() + pool = session._pools.get.return_value + connection = Mock(spec=Connection) + connection.protocol_version = 4 + connection.features = Mock() + connection.features.use_metadata_id = False + pool.borrow_connection.return_value = (connection, 1) + + old_meta = [('ks', 'tb', 'col', Mock())] + ps = self._make_prepared_statement(old_meta, b'old_id') + + rf = self.make_response_future(session) + rf.prepared_statement = ps + rf.send_request() + + # result_metadata_id is None → server sent full metadata, no hash update + response = self._make_rows_response( + result_metadata_id=None, + column_metadata=old_meta, + ) + rf._set_result(None, None, None, response) + + assert ps.result_metadata is old_meta + assert ps.result_metadata_id == b'old_id' + + def test_set_result_warns_when_metadata_id_but_no_column_metadata(self): + """ + If the server sends a new result_metadata_id but no column metadata + (protocol violation), _set_result must emit a WARNING and cache + NEITHER value: adopting the new id while keeping the old metadata would + make the server skip sending metadata on subsequent executes (the ids + match) while the driver decodes with stale metadata — with no recovery. + Keeping the old pair means the next EXECUTE sends the old id, the server + detects the mismatch, and the driver recovers with full metadata. + """ + session = self.make_session() + pool = session._pools.get.return_value + connection = Mock(spec=Connection) + connection.protocol_version = 4 + connection.features = Mock() + connection.features.use_metadata_id = False + pool.borrow_connection.return_value = (connection, 1) + + old_meta = [('ks', 'tb', 'col', Mock())] + ps = self._make_prepared_statement(old_meta, b'old_id') + + rf = self.make_response_future(session) + rf.prepared_statement = ps + rf.send_request() + + # column_metadata is falsy (empty list) but result_metadata_id is set + response = self._make_rows_response( + result_metadata_id=b'new_id', + column_metadata=[], + ) + + with self.assertLogs('cassandra.cluster', level='WARNING') as log_ctx: + rf._set_result(None, None, None, response) + + assert any('result_metadata_id' in msg for msg in log_ctx.output) + # nothing is cached from the anomalous response + assert ps.result_metadata_and_id == (old_meta, b'old_id') + + def test_set_result_warns_when_metadata_id_but_column_metadata_is_none(self): + """ + Like the empty-list variant above, but column_metadata=None (attribute + absent rather than explicitly empty). Both None and [] are falsy, so + the warning branch is taken and the cached pair is left unchanged. + """ + session = self.make_session() + pool = session._pools.get.return_value + connection = Mock(spec=Connection) + connection.protocol_version = 4 + connection.features = Mock() + connection.features.use_metadata_id = False + pool.borrow_connection.return_value = (connection, 1) + + old_meta = [('ks', 'tb', 'col', Mock())] + ps = self._make_prepared_statement(old_meta, b'old_id') + + rf = self.make_response_future(session) + rf.prepared_statement = ps + rf.send_request() + + response = self._make_rows_response( + result_metadata_id=b'new_id', + column_metadata=None, + ) + + with self.assertLogs('cassandra.cluster', level='WARNING') as log_ctx: + rf._set_result(None, None, None, response) + + assert any('result_metadata_id' in msg for msg in log_ctx.output) + assert ps.result_metadata_and_id == (old_meta, b'old_id') + + def test_set_result_anomalous_metadata_id_warns_once_and_rearms(self): + """ + The anomalous-response warning (new id, no column metadata) is logged + once per prepared statement, not once per execute: a persistently + misbehaving server must not spam the log. A successful METADATA_CHANGED + in between re-arms the warning so a later recurrence is logged again. + """ + session = self.make_session() + pool = session._pools.get.return_value + connection = Mock(spec=Connection) + connection.protocol_version = 4 + connection.features = Mock() + connection.features.use_metadata_id = False + pool.borrow_connection.return_value = (connection, 1) + + old_meta = [('ks', 'tb', 'col', Mock())] + ps = self._make_prepared_statement(old_meta, b'old_id') + + rf = self.make_response_future(session) + rf.prepared_statement = ps + rf.send_request() + + anomalous = self._make_rows_response(result_metadata_id=b'new_id', column_metadata=[]) + + # First anomalous response: warns once. + with self.assertLogs('cassandra.cluster', level='WARNING') as first: + rf._set_result(None, None, None, anomalous) + 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'): + rf._set_result(None, None, None, anomalous) + assert ps.result_metadata_and_id == (old_meta, b'old_id') + + # A genuine METADATA_CHANGED recovers the metadata and re-arms the warning. + new_meta = [('ks', 'tb', 'new_col', Mock())] + rf._set_result(None, None, None, + self._make_rows_response(result_metadata_id=b'new_id', column_metadata=new_meta)) + assert ps.result_metadata_and_id == (new_meta, b'new_id') + + # After recovery, the anomaly warns again. + with self.assertLogs('cassandra.cluster', level='WARNING') as after: + rf._set_result(None, None, None, anomalous) + assert sum('result_metadata_id' in msg for msg in after.output) == 1 + + def test_create_execute_message_with_metadata_and_id(self): + """ + When the prepared statement carries both a result_metadata_id and usable + cached result_metadata, _create_response_future must build the + ExecuteMessage with skip_meta=True and the metadata id attached. Whether + either actually reaches the wire is decided per connection at + serialization time (ExecuteMessage.send_body). + """ + ps = self._make_prepared_statement([('ks', 'tbl', 'col', Mock())], b'meta_hash') + + rf = self._create_execute_future(ps) + + assert rf.message.skip_meta is True + assert rf.message.result_metadata_id == b'meta_hash' + + def test_create_execute_message_without_metadata_id(self): + """ + A statement prepared before the extension was active (result_metadata_id + is None) must never request skip_meta — the driver has no hash the server + could validate the cached metadata against. + """ + ps = self._make_prepared_statement([('ks', 'tbl', 'col', Mock())], None) + + rf = self._create_execute_future(ps) + + assert rf.message.skip_meta is False + assert rf.message.result_metadata_id is None + + def test_create_execute_message_result_metadata_none(self): + """ + LWT / conditional statements (INSERT ... IF NOT EXISTS) have a + result_metadata_id but NO_METADATA for the result columns, leaving + result_metadata as None. skip_meta must stay off: the server would omit + column definitions while the driver has nothing cached to decode with. + The id still rides on the message so id-aware connections always send it. + """ + ps = self._make_prepared_statement(None, b'lwt_hash') + + rf = self._create_execute_future(ps) + + assert rf.message.skip_meta is False + assert rf.message.result_metadata_id == b'lwt_hash' + + def test_create_execute_message_result_metadata_empty(self): + """ + Statements returning zero result columns (plain INSERT/UPDATE/DELETE) + have result_metadata == []. Like the None case, skip_meta stays off — + there is no metadata worth skipping. + """ + ps = self._make_prepared_statement([], b'meta_hash') + + rf = self._create_execute_future(ps) + + 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): + """ + 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. + """ + ps = self._make_prepared_statement([('ks', 'tbl', 'col', Mock())], b'meta_hash') + + rf = self._create_execute_future(ps, continuous_paging_options=Mock()) + + assert rf.message.skip_meta is False + assert rf.message.result_metadata_id == b'meta_hash' + + def test_query_does_not_mutate_execute_message(self): + """ + _query() must send the ExecuteMessage exactly as constructed: all + per-connection decisions (whether the id field and the skip_meta flag hit + the wire) happen at serialization time from the connection's negotiated + features. Mutating the shared message per attempt would race with + speculative executions sending the same message on another connection. + """ + session = self.make_basic_session() + session.cluster._default_load_balancing_policy.make_query_plan.return_value = ['ip1'] + session._pools.get.return_value = self.make_pool() + + connection = Mock(spec=Connection) + connection.protocol_version = 4 + connection.features = Mock() + connection.features.use_metadata_id = True + session._pools.get.return_value.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) + original_skip_meta = rf.message.skip_meta + original_id = rf.message.result_metadata_id + + rf.send_request() + + connection.send_msg.assert_called_once() + sent_message = connection.send_msg.call_args[0][0] + assert sent_message is rf.message + assert rf.message.skip_meta is original_skip_meta + assert rf.message.result_metadata_id is original_id + assert not hasattr(rf.message, 'use_metadata_id') + + 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 + was built (paired with the id the immutable message carries), not a fresh read of + the prepared statement's cache. Otherwise a concurrent METADATA_CHANGED landing + between construction and send could pair the message's id with a different schema + version's metadata — the torn read the atomic pair was meant to prevent. + """ + session = self.make_basic_session() + session.cluster._default_load_balancing_policy.make_query_plan.return_value = ['ip1'] + session._pools.get.return_value = self.make_pool() + + connection = Mock(spec=Connection) + connection.protocol_version = 4 + connection.features = Mock() + connection.features.use_metadata_id = True + session._pools.get.return_value.borrow_connection.return_value = (connection, 1) + + meta_v1 = [('ks', 'tbl', 'col_v1', Mock())] + 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 + + # 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 + + 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