Add fetch_size to ControlConnection topology queries

- Add fetch_size parameter to all QueryMessage instances in ControlConnection
- This ensures system.peers and system.local queries use paging
- Uses existing _schema_meta_page_size (default 1000) for consistency
- Add unit test to verify fetch_size is being passed

Co-authored-by: mykaul <4655593+mykaul@users.noreply.github.com>

Author: Copilot

cassandra/cluster.py

@@ -3638,9 +3638,11 @@ def _try_connect(self, host):
             sel_peers = self._get_peers_query(self.PeersQueryType.PEERS, connection)
             sel_local = self._SELECT_LOCAL if self._token_meta_enabled else self._SELECT_LOCAL_NO_TOKENS
             peers_query = QueryMessage(query=maybe_add_timeout_to_query(sel_peers, self._metadata_request_timeout),
-                                       consistency_level=ConsistencyLevel.ONE)
+                                       consistency_level=ConsistencyLevel.ONE,
+                                       fetch_size=self._schema_meta_page_size)
             local_query = QueryMessage(query=maybe_add_timeout_to_query(sel_local, self._metadata_request_timeout),
-                                       consistency_level=ConsistencyLevel.ONE)
+                                       consistency_level=ConsistencyLevel.ONE,
+                                       fetch_size=self._schema_meta_page_size)
             (peers_success, peers_result), (local_success, local_result) = connection.wait_for_responses(
                 peers_query, local_query, timeout=self._timeout, fail_on_error=False)
 
@@ -3652,7 +3654,8 @@ def _try_connect(self, host):
                 self._uses_peers_v2 = False
                 sel_peers = self._get_peers_query(self.PeersQueryType.PEERS, connection)
                 peers_query = QueryMessage(query=maybe_add_timeout_to_query(sel_peers, self._metadata_request_timeout),
-                                           consistency_level=ConsistencyLevel.ONE)
+                                           consistency_level=ConsistencyLevel.ONE,
+                                           fetch_size=self._schema_meta_page_size)
                 peers_result = connection.wait_for_response(
                     peers_query, timeout=self._timeout)
 
@@ -3797,9 +3800,11 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
                 log.debug("[control connection] Refreshing node list and token map")
                 sel_local = self._SELECT_LOCAL
             peers_query = QueryMessage(query=maybe_add_timeout_to_query(sel_peers, self._metadata_request_timeout),
-                                       consistency_level=cl)
+                                       consistency_level=cl,
+                                       fetch_size=self._schema_meta_page_size)
             local_query = QueryMessage(query=maybe_add_timeout_to_query(sel_local, self._metadata_request_timeout),
-                                       consistency_level=cl)
+                                       consistency_level=cl,
+                                       fetch_size=self._schema_meta_page_size)
             peers_result, local_result = connection.wait_for_responses(
                 peers_query, local_query, timeout=self._timeout)
 
@@ -3856,7 +3861,8 @@ def _refresh_node_list_and_token_map(self, connection, preloaded_results=None,
                         # in system.local. See CASSANDRA-9436.
                         local_rpc_address_query = QueryMessage(
                             query=maybe_add_timeout_to_query(self._SELECT_LOCAL_NO_TOKENS_RPC_ADDRESS, self._metadata_request_timeout),
-                            consistency_level=ConsistencyLevel.ONE)
+                            consistency_level=ConsistencyLevel.ONE,
+                            fetch_size=self._schema_meta_page_size)
                         success, local_rpc_address_result = connection.wait_for_response(
                             local_rpc_address_query, timeout=self._timeout, fail_on_error=False)
                         if success:
@@ -4092,9 +4098,11 @@ def wait_for_schema_agreement(self, connection=None, preloaded_results=None, wai
 
             while elapsed < total_timeout:
                 peers_query = QueryMessage(query=maybe_add_timeout_to_query(select_peers_query, self._metadata_request_timeout),
-                                           consistency_level=cl)
+                                           consistency_level=cl,
+                                           fetch_size=self._schema_meta_page_size)
                 local_query = QueryMessage(query=maybe_add_timeout_to_query(self._SELECT_SCHEMA_LOCAL, self._metadata_request_timeout),
-                                           consistency_level=cl)
+                                           consistency_level=cl,
+                                           fetch_size=self._schema_meta_page_size)
                 try:
                     timeout = min(self._timeout, total_timeout - elapsed)
                     peers_result, local_result = connection.wait_for_responses(

tests/unit/test_control_connection.py

@@ -18,7 +18,7 @@
 from unittest.mock import Mock, ANY, call
 
 from cassandra import OperationTimedOut, SchemaTargetType, SchemaChangeType
-from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS
+from cassandra.protocol import ResultMessage, RESULT_KIND_ROWS, QueryMessage
 from cassandra.cluster import ControlConnection, _Scheduler, ProfileManager, EXEC_PROFILE_DEFAULT, ExecutionProfile
 from cassandra.pool import Host
 from cassandra.connection import EndPoint, DefaultEndPoint, DefaultEndPointFactory
@@ -305,6 +305,27 @@ def test_refresh_nodes_and_tokens(self):
 
         assert self.connection.wait_for_responses.call_count == 1
 
+    def test_topology_queries_use_paging(self):
+        """
+        Test that topology queries (system.peers and system.local) use fetch_size parameter
+        """
+        # Test during refresh_node_list_and_token_map
+        self.control_connection.refresh_node_list_and_token_map()
+        
+        # Verify that wait_for_responses was called
+        assert self.connection.wait_for_responses.called
+        
+        # Get the QueryMessage arguments
+        call_args = self.connection.wait_for_responses.call_args[0]
+        peers_query = call_args[0]
+        local_query = call_args[1]
+        
+        # Verify that both queries have fetch_size set
+        assert isinstance(peers_query, QueryMessage)
+        assert isinstance(local_query, QueryMessage)
+        assert peers_query.fetch_size == 1000  # default schema_meta_page_size
+        assert local_query.fetch_size == 1000  # default schema_meta_page_size
+
     def test_refresh_nodes_and_tokens_with_invalid_peers(self):
         def refresh_and_validate_added_hosts():
             self.connection.wait_for_responses = Mock(return_value=_node_meta_results(