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

@@ -4736,7 +4736,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_prepared_statements.py

@@ -614,13 +614,15 @@ def _test_updated_conditional(self, session, value):
         prepared_statement = session.prepare(
             "INSERT INTO {}(a, b, d) VALUES "
             "(?, ? , ?) IF NOT EXISTS".format(self.table_name))
-        first_id = prepared_statement.result_metadata_id
-        LOG.debug('initial result_metadata_id: {}'.format(first_id))
+        LOG.debug('initial result_metadata_id: {}'.format(prepared_statement.result_metadata_id))
 
+        # The cached (result_metadata, result_metadata_id) pair is not asserted on:
+        # a METADATA_CHANGED response refreshes it for a conditional statement like
+        # for any other, so its contents are the server's business. What must hold
+        # is that each result is decoded against the metadata describing it, whether
+        # the conditional update applied (narrow shape) or not (whole row).
         def check_result_and_metadata(expected):
             assert session.execute(prepared_statement, (value, value, value)).one() == expected
-            assert prepared_statement.result_metadata_id == first_id
-            assert prepared_statement.result_metadata is None
 
         # Successful conditional update
         check_result_and_metadata((True,))

tests/integration/standard/test_scylla_metadata_id.py

@@ -0,0 +1,166 @@
+# 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
+from unittest.mock import patch
+
+import pytest
+
+from cassandra.cluster import ResponseFuture
+from tests.integration import use_singledc, SCYLLA_VERSION, BasicSharedKeyspaceUnitTestCase, \
+    drop_keyspace_shutdown_cluster
+
+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.
+        try:
+            if not cls._negotiated_use_metadata_id():
+                raise unittest.SkipTest(
+                    "Scylla node does not advertise SCYLLA_USE_METADATA_ID")
+        except Exception:
+            # setUpClass raising means unittest never calls tearDownClass, so the
+            # cluster and keyspace created above are torn down here explicitly.
+            drop_keyspace_shutdown_cluster(cls.ks_name, cls.session, cls.cluster)
+            raise
+
+    @classmethod
+    def _negotiated_use_metadata_id(cls):
+        """Whether this class's data-path connections negotiated SCYLLA_USE_METADATA_ID.
+
+        Reads the pool's existing connection rather than borrowing one:
+        borrow_connection() pops a stream id that only Connection.process_msg gives
+        back, so borrowing without sending a message would leak it.
+        """
+        pool = next(iter(cls.session.get_pools()))
+        return next(iter(pool._connections.values())).features.use_metadata_id
+
+    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.
+        """
+        assert self._negotiated_use_metadata_id() is True
+
+    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
+
+        # The table was not altered, so the statement is still valid server-side and
+        # nothing should re-prepare it. Spying on _reprepare keeps this test honest:
+        # if the id came back via an UNPREPARED/reprepare round trip instead, the
+        # METADATA_CHANGED-on-ROWS path in ResponseFuture._set_result would not
+        # actually be under test here.
+        with patch.object(ResponseFuture, '_reprepare', autospec=True,
+                          side_effect=ResponseFuture._reprepare) as reprepare_spy:
+            result = self.session.execute(prepared.bind((1,)))
+
+        assert reprepare_spy.call_count == 0
+        assert list(result) == [(1, 1, 1)]
+        assert prepared.result_metadata_id is not None
+
+    def test_conditional_statement_metadata_is_stable_across_outcomes(self):
+        """
+        Conditional (LWT) statements get no special handling, and this pins the
+        server behaviour that makes that correct.
+
+        Cassandra returns NO_METADATA for a conditional statement at PREPARE and
+        then varies the result shape per execution — ``(True,)`` when applied, the
+        conflicting row when not — which is what PYTHON-847 is about. Scylla
+        instead describes the result up front as ``[applied]`` plus every column of
+        the row, filling nulls when the update applied. The shape does not
+        alternate, so the cached metadata id stays valid across both outcomes and
+        ``skip_meta`` is exactly as safe here as for any other statement. A schema
+        change still changes the id, and the driver must pick that up.
+        """
+        prepared = self.session.prepare(
+            "INSERT INTO {} (a, b, c) VALUES (?, ?, ?) IF NOT EXISTS".format(self.table_name))
+        id_before = prepared.result_metadata_id
+        assert id_before is not None
+        assert len(prepared.result_metadata) == 4  # [applied], a, b, c
+
+        # a=2 is free, so the insert applies; the row columns come back null.
+        assert self.session.execute(prepared.bind((2, 2, 2))).one() == (True, None, None, None)
+
+        # a=1 exists (setUp), so this one does not apply and the conflicting row is
+        # returned — same metadata, so the cached pair is still the right one.
+        assert self.session.execute(prepared.bind((1, 9, 9))).one() == (False, 1, 1, 1)
+        assert prepared.result_metadata_id == id_before
+
+        self.session.execute("ALTER TABLE {} ADD d int".format(self.table_name))
+
+        # The result gained a column, so the server must report a new id and the
+        # driver must adopt it — for a conditional statement like for any other.
+        assert self.session.execute(prepared.bind((1, 9, 9))).one() == (False, 1, 1, 1, None)
+        assert prepared.result_metadata_id != id_before
+        assert len(prepared.result_metadata) == 5
+
+        assert self.session.execute(prepared.bind((3, 3, 3))).one() == (True, None, None, None, None)