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.

Author: nikagra

CHANGELOG.rst

@@ -1,8 +1,19 @@
 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
 ------
+* ``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.
 * 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``.

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,27 @@ 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: while
+            # a statement has no cached metadata there is nothing to decode a
+            # 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
             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 +3109,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 +4736,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 +4754,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 +4983,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 +5037,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 +5202,33 @@ 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).
+                            # (this also re-arms the anomaly warning below)
+                            self.prepared_statement.update_result_metadata(
+                                response.column_metadata, new_result_metadata_id)
+                        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 +5379,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

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
@@ -625,6 +636,17 @@ def send_body(self, f, protocol_version, protocol_features=None):
         self._write_query_params(f, protocol_version, protocol_features)
 
 
+def _metadata_id_negotiated(protocol_version, protocol_features):
+    """Whether the result-metadata-id field is part of the frame layout.
+
+    It is part of the layout of EXECUTE requests and PREPARE responses whenever
+    the connection speaks CQL v5+ natively or negotiated SCYLLA_USE_METADATA_ID,
+    so on such connections it must always be written and always be read.
+    """
+    return (ProtocolVersion.uses_prepared_metadata(protocol_version)
+            or (protocol_features is not None and protocol_features.use_metadata_id))
+
+
 class ExecuteMessage(_QueryMessage):
     opcode = 0x0A
     name = 'EXECUTE'
@@ -638,13 +660,34 @@ 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)
 
+    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 :func:`_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 _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 _metadata_id_negotiated(protocol_version, protocol_features):
             self.result_metadata_id = read_binary_string(f)
         else:
             self.result_metadata_id = None

cassandra/protocol_features.py

@@ -10,31 +10,37 @@
 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
     shard_id = 0
     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):
         rate_limit_error = ProtocolFeatures.maybe_parse_rate_limit_error(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])