fix: gate skip_meta and result_metadata_id on connection features at send-time

Using result_metadata_id != None as a proxy for whether SCYLLA_USE_METADATA_ID
was negotiated is unreliable in mixed-version clusters: a statement prepared
against a Scylla node (which sets result_metadata_id) could then be executed
against a node that doesn't support the extension, shifting the EXECUTE wire
layout and causing protocol errors.

Move the skip_meta and result_metadata_id decisions into _query(), where the
borrowed connection is known. Both are now gated on the connection's actual
negotiated features (connection.features.use_metadata_id for the Scylla
extension, ProtocolVersion.uses_prepared_metadata() for CQL v5+), ensuring
the correct wire format is used for each specific connection.

Author: nikagra

cassandra/cluster.py

@@ -2998,10 +2998,7 @@ def _create_response_future(self, query, parameters, trace, custom_payload,
             message = ExecuteMessage(
                 prepared_statement.query_id, query.values, cl,
                 serial_cl, fetch_size, paging_state, timestamp,
-                skip_meta=(ProtocolVersion.uses_prepared_metadata(self._protocol_version)
-                           or bool(prepared_statement.result_metadata_id)),
-                continuous_paging_options=continuous_paging_options,
-                result_metadata_id=prepared_statement.result_metadata_id)
+                continuous_paging_options=continuous_paging_options)
         elif isinstance(query, BatchStatement):
             if self._protocol_version < 2:
                 raise UnsupportedOperation(
@@ -4619,6 +4616,12 @@ def _query(self, host, message=None, cb=None):
             self._connection = connection
             result_meta = self.prepared_statement.result_metadata if self.prepared_statement else []
 
+            if self.prepared_statement and isinstance(message, ExecuteMessage):
+                use_metadata_id = (ProtocolVersion.uses_prepared_metadata(connection.protocol_version)
+                                   or connection.features.use_metadata_id)
+                message.skip_meta = use_metadata_id
+                message.result_metadata_id = self.prepared_statement.result_metadata_id if use_metadata_id else None
+
             if cb is None:
                 cb = partial(self._set_result, host, connection, pool)