DRIVER-153: add tests, defensive warning and docs for SCYLLA_USE_METADATA_ID

- Add unit tests for the _METADATA_ID_FLAG path in recv_results_metadata
  (ROWS result with METADATA_CHANGED signal)
- Add unit tests for _set_result metadata cache update on METADATA_CHANGED:
  update both result_metadata and result_metadata_id, no-op when id absent,
  warning when id present but column_metadata empty
- Add unit tests for _query per-connection feature gating: skip_meta and
  result_metadata_id are set only when the connection negotiated
  SCYLLA_USE_METADATA_ID (or protocol v5) and the prepared statement
  carries a result_metadata_id
- Add defensive log.warning in _set_result when server sends a new
  result_metadata_id without column_metadata (protocol violation)
- Add write-order comment explaining thread-safety rationale for the two
  assignments to prepared_statement.result_metadata / result_metadata_id
- Add SCYLLA_USE_METADATA_ID section to docs/scylla-specific.rst

Author: nikagra

cassandra/cluster.py

@@ -4794,7 +4794,20 @@ def _set_result(self, host, connection, pool, response):
                     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:
+                            # Write result_metadata before result_metadata_id intentionally:
+                            # a concurrent reader that still sees the old metadata_id will
+                            # ask the server for full metadata and recover safely; a reader
+                            # that sees the new metadata_id together with the new metadata
+                            # is immediately correct. The opposite write order could expose
+                            # a window where a reader uses a new metadata_id with stale metadata.
                             self.prepared_statement.result_metadata = response.column_metadata
+                        else:
+                            log.warning(
+                                "Server sent a new result_metadata_id but no column metadata "
+                                "for prepared statement %r. The cached column metadata will not "
+                                "be updated; only result_metadata_id is refreshed.",
+                                getattr(self.prepared_statement, 'query_id', None)
+                            )
                         self.prepared_statement.result_metadata_id = new_result_metadata_id
                     if getattr(self.message, 'continuous_paging_options', None):
                         self._handle_continuous_paging_first_response(connection, response)

docs/scylla-specific.rst

@@ -156,3 +156,43 @@ 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 executing prepared SELECT statements, the driver normally requests the server
+to skip sending full result metadata with each response (``skip_meta`` optimization),
+relying on the metadata cached from the initial ``PREPARE`` call. However, if the
+table schema changes after a statement is prepared (e.g., a column is added, removed,
+or its type is altered), this 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 connection must have negotiated ``SCYLLA_USE_METADATA_ID`` (or use
+  CQL v5), *and* the prepared statement must carry a ``result_metadata_id`` obtained
+  from PREPARE.
+- 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) retain ``result_metadata_id=None`` and fall back to always requesting
+  full metadata, which is the safest option.
+
+**Current scope:** schema-change detection is implemented for SELECT statements.
+UPDATE/INSERT coverage is planned in a separate effort.
+
+For full protocol details see the ScyllaDB CQL extensions documentation:
+https://opensource.docs.scylladb.com/stable/cql/cql-extensions.html

tests/unit/test_protocol.py

@@ -24,7 +24,8 @@
     _PAGING_OPTIONS_FLAG, _WITH_SERIAL_CONSISTENCY_FLAG,
     _PAGE_SIZE_FLAG, _WITH_PAGING_STATE_FLAG,
     _SKIP_METADATA_FLAG,
-    BatchMessage, ResultMessage
+    BatchMessage, ResultMessage,
+    RESULT_KIND_ROWS
 )
 from cassandra.protocol_features import ProtocolFeatures
 from cassandra.query import BatchType
@@ -153,6 +154,44 @@ def test_recv_results_prepared_no_extension_skips_metadata_id(self):
         assert msg.query_id == b'ab'
         assert msg.result_metadata_id is None
 
+    def test_recv_results_metadata_changed_flag(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={})
+        assert not hasattr(msg, 'result_metadata_id') or msg.result_metadata_id is None
+        assert not hasattr(msg, 'column_metadata') or msg.column_metadata is None
+
     def test_query_message(self):
         """
         Test to check the appropriate calls are made