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.

Author: nikagra

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
 

tests/integration/standard/test_scylla_metadata_id.py

@@ -0,0 +1,84 @@
+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)
+
+    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.protocol_version == 4
+            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

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',