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

nikagra · nikagra · commit 6eea397cf99c · 2026-04-10T00:32:00.000+02:00
- 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
diff --git a/cassandra/cluster.py b/cassandra/cluster.py
@@ -4784,7 +4784,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)
diff --git a/docs/scylla-specific.rst b/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
diff --git a/tests/unit/test_protocol.py b/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
diff --git 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,
@@ -723,3 +724,253 @@ 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_execute_response_future(self, session, connection, prepared_statement):
+        """
+        Return a ResponseFuture whose message is an ExecuteMessage and which
+        has a prepared_statement set so that _query()'s feature-gating logic
+        is exercised.
+        """
+        execute_msg = ExecuteMessage(b'qid', [], ConsistencyLevel.ONE)
+        query = SimpleStatement("SELECT * FROM foo")
+        rf = ResponseFuture(
+            session, execute_msg, query, timeout=1,
+            prepared_statement=prepared_statement,
+        )
+        pool = session._pools.get.return_value
+        pool.borrow_connection.return_value = (connection, 1)
+        return rf
+
+    # -------------------------------------------------------------------------
+    # _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 = Mock()
+        ps.result_metadata = old_meta
+        ps.result_metadata_id = 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'
+
+    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 = Mock()
+        ps.result_metadata = old_meta
+        ps.result_metadata_id = 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 still update result_metadata_id
+        and emit a WARNING log so the inconsistency is visible in logs.
+        """
+        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)
+
+        ps = Mock()
+        ps.result_metadata = [('ks', 'tb', 'col', Mock())]
+        ps.result_metadata_id = 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)
+        # metadata_id is still updated even without column metadata
+        assert ps.result_metadata_id == b'new_id'
+
+    # -------------------------------------------------------------------------
+    # _query: per-connection feature gating for skip_meta / result_metadata_id
+    # -------------------------------------------------------------------------
+
+    def test_query_sets_skip_meta_with_scylla_extension(self):
+        """
+        When the borrowed connection has negotiated SCYLLA_USE_METADATA_ID and
+        the prepared statement carries a result_metadata_id, _query() must set
+        skip_meta=True and attach the metadata_id to the ExecuteMessage.
+        """
+        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 = Mock()
+        ps.result_metadata = []
+        ps.result_metadata_id = b'meta_hash'
+
+        rf = self._make_execute_response_future(session, connection, ps)
+        rf.send_request()
+
+        assert rf.message.skip_meta is True
+        assert rf.message.result_metadata_id == b'meta_hash'
+
+    def test_query_no_skip_meta_without_extension(self):
+        """
+        When the connection does NOT have SCYLLA_USE_METADATA_ID (and protocol
+        is v4), _query() must leave skip_meta=False and result_metadata_id=None.
+        """
+        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 = False
+        session._pools.get.return_value.borrow_connection.return_value = (connection, 1)
+
+        ps = Mock()
+        ps.result_metadata = []
+        ps.result_metadata_id = b'meta_hash'
+
+        rf = self._make_execute_response_future(session, connection, ps)
+        rf.send_request()
+
+        assert rf.message.skip_meta is False
+        assert rf.message.result_metadata_id is None
+
+    def test_query_no_skip_meta_when_prepared_statement_has_no_metadata_id(self):
+        """
+        Even if the connection supports SCYLLA_USE_METADATA_ID, if the prepared
+        statement was created before the extension was active (result_metadata_id
+        is None), _query() must NOT set skip_meta — the driver has no hash to send.
+        """
+        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   # extension active on this connection
+        session._pools.get.return_value.borrow_connection.return_value = (connection, 1)
+
+        ps = Mock()
+        ps.result_metadata = []
+        ps.result_metadata_id = None   # statement prepared without extension
+
+        rf = self._make_execute_response_future(session, connection, ps)
+        rf.send_request()
+
+        assert rf.message.skip_meta is False
+        assert rf.message.result_metadata_id is None
+
+    def test_query_sets_skip_meta_for_protocol_v5(self):
+        """
+        On a protocol-v5 connection (ProtocolVersion.uses_prepared_metadata),
+        _query() must set skip_meta=True and send result_metadata_id even if
+        the Scylla-specific use_metadata_id flag is False.
+        """
+        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 = 5  # CQL v5 — uses_prepared_metadata() is True
+        connection.features = Mock()
+        connection.features.use_metadata_id = False  # Scylla extension not needed
+        session._pools.get.return_value.borrow_connection.return_value = (connection, 1)
+
+        ps = Mock()
+        ps.result_metadata = []
+        ps.result_metadata_id = b'v5_hash'
+
+        rf = self._make_execute_response_future(session, connection, ps)
+        rf.send_request()
+
+        assert rf.message.skip_meta is True
+        assert rf.message.result_metadata_id == b'v5_hash'
