Change prepare_on_all_hosts default to False

CHANGELOG.rst

@@ -58,6 +58,17 @@ Others
   come through unchanged. Previously ``DRIVER_NAME`` and ``DRIVER_VERSION`` could be
   overridden, which misreported the driver to the server for the life of the connection
   and, in the clients table, to the operator reading the row.
+* ``Cluster.prepare_on_all_hosts`` now defaults to unset instead of ``True``. In multi-DC
+  deployments eager preparation previously ran on every pooled host, including remote hosts
+  that are rarely or never queried. Left unset, a ``Session`` now eagerly prepares on all
+  hosts only during a short warm-up window after it connects (``prepare_on_all_hosts_warmup_seconds``,
+  default 15s), when hosts have just been discovered and many different statements are likely
+  to hit many different hosts in quick succession; afterwards it falls back to the lazy
+  behavior (``prepare_on_all_hosts=False``), since steady-state traffic for a given prepared
+  statement usually concentrates on a stable subset of replicas via token-aware routing.
+  Passing ``prepare_on_all_hosts=True`` or ``False`` explicitly disables the warm-up and pins
+  the old, unconditional behavior for the life of the cluster. An ``UNPREPARED`` response
+  still triggers on-demand reprepare and retry, so correctness is unaffected either way.
 * ``PreparedStatement.result_metadata`` and ``PreparedStatement.result_metadata_id`` are
   now read-only. They are replaced together by
   ``PreparedStatement.update_result_metadata()``, so a request can never observe a metadata

cassandra/cluster.py

@@ -985,13 +985,57 @@ def default_retry_policy(self, policy):
     establish connection pools. This can cause a rush of connections and queries if not mitigated with this factor.
     """
 
-    prepare_on_all_hosts = True
+    _prepare_on_all_hosts = False
+    _prepare_on_all_hosts_explicit = False
+
+    @property
+    def prepare_on_all_hosts(self):
+        """
+        Specifies whether statements should be prepared on all hosts, or just one.
+
+        When enabled, statements are eagerly prepared on every host with an open connection pool. In multi-DC
+        deployments this includes remote hosts that are rarely or never queried on the happy path; preparing on them
+        is purely a latency optimization, since an ``UNPREPARED`` response always triggers on-demand reprepare and
+        retry. It can be enabled on long-running applications with numerous clients preparing statements on startup,
+        where a randomized initial condition of the load balancing policy can be expected to distribute prepares from
+        different clients across the cluster.
+
+        If left unset (the default), a :class:`.Session` instead applies :attr:`.prepare_on_all_hosts_warmup_seconds`:
+        it behaves as if this were ``True`` for a short warm-up window right after the session connects, then as if
+        ``False`` afterwards. Explicitly assigning ``True`` or ``False``, whether to the :class:`.Cluster`
+        constructor or to this attribute at any later point, disables the warm-up behavior and pins this to the
+        given value for the lifetime of the cluster.
+        """
+        return self._prepare_on_all_hosts
+
+    @prepare_on_all_hosts.setter
+    def prepare_on_all_hosts(self, value):
+        self._prepare_on_all_hosts = value
+        self._prepare_on_all_hosts_explicit = True
+
+    prepare_on_all_hosts_warmup_seconds = 15
     """
-    Specifies whether statements should be prepared on all hosts, or just one.
+    Length, in seconds, of the warm-up window used to decide whether :meth:`.Session.prepare` eagerly prepares
+    on all pooled hosts, when :attr:`.prepare_on_all_hosts` was not explicitly set by the caller.
+
+    Right after a :class:`.Session` connects, hosts have just been discovered and different callers/tests
+    typically prepare many different statements against many different hosts in quick succession; eagerly
+    broadcasting each prepare avoids a burst of ``UNPREPARED``/reprepare/retry round trips during that period.
+    In steady state, query traffic for a given prepared statement usually concentrates on a stable subset of
+    replicas (via token-aware routing), so broadcasting to every host is normally wasted work, and the driver
+    falls back to lazy on-demand reprepare (the same behavior as ``prepare_on_all_hosts=False``).
 
-    This can reasonably be disabled on long-running applications with numerous clients preparing statements on startup,
-    where a randomized initial condition of the load balancing policy can be expected to distribute prepares from
-    different clients across the cluster.
+    The window is measured from when the :class:`.Session` finished establishing its initial connection pools,
+    not from the first call to :meth:`.Session.prepare`. An application that waits well past connect before
+    ever calling ``prepare()`` (lazy-first-use) will not benefit from the warm-up window, since by then hosts
+    are no longer "freshly discovered" and the startup thundering-herd risk this is meant to mitigate has
+    already passed.
+
+    Setting this to zero (or a falsy value) disables the warm-up behavior entirely, equivalent to leaving
+    :attr:`.prepare_on_all_hosts` at its unset default with no warm-up: statements are never eagerly broadcast
+    unless the flag is set explicitly.
+
+    Has no effect when :attr:`.prepare_on_all_hosts` was explicitly set by the caller.
     """
 
     reprepare_on_up = True
@@ -1205,7 +1249,7 @@ def __init__(self,
                  schema_metadata_page_size=1000,
                  address_translator=None,
                  status_event_refresh_window=2,
-                 prepare_on_all_hosts=True,
+                 prepare_on_all_hosts=_NOT_SET,
                  reprepare_on_up=True,
                  execution_profiles=None,
                  allow_beta_protocol_version=False,
@@ -1222,7 +1266,8 @@ def __init__(self,
                  application_info:Optional[ApplicationInfoBase]=None,
                  client_routes_config:Optional[ClientRoutesConfig]=None,
                  allow_control_connection_query_fallback:Optional[ControlConnectionQueryFallback]=ControlConnectionQueryFallback.Disabled,
-                 driver_config_reporting_enabled=True
+                 driver_config_reporting_enabled=True,
+                 prepare_on_all_hosts_warmup_seconds=15
                  ):
         """
         ``executor_threads`` defines the number of threads in a pool for handling asynchronous tasks such as
@@ -1491,7 +1536,12 @@ def __init__(self,
         self.topology_event_refresh_window = topology_event_refresh_window
         self.status_event_refresh_window = status_event_refresh_window
         self.connect_timeout = connect_timeout
-        self.prepare_on_all_hosts = prepare_on_all_hosts
+        if prepare_on_all_hosts is _NOT_SET:
+            self._prepare_on_all_hosts = False
+            self._prepare_on_all_hosts_explicit = False
+        else:
+            self.prepare_on_all_hosts = prepare_on_all_hosts
+        self.prepare_on_all_hosts_warmup_seconds = prepare_on_all_hosts_warmup_seconds
         self.reprepare_on_up = reprepare_on_up
         self.shard_aware_options = ShardAwareOptions(opts=shard_aware_options)
 
@@ -1786,6 +1836,8 @@ def connect(self, keyspace=None, wait_for_all_pools=False):
         session = self._new_session(keyspace)
         if wait_for_all_pools:
             wait_futures(session._initial_connect_futures)
+            # reset so the warm-up window starts after all pools are up, not just the first
+            session._connect_time = time.time()
 
         self._set_default_dbaas_consistency(session)
 
@@ -2652,6 +2704,9 @@ def __init__(self, cluster, hosts, keyspace=None):
                 raise NoHostAvailable(msg, [h.address for h in hosts])
 
         self.session_id = uuid.uuid4()
+        # marks when this session finished its initial pool setup; used to gauge whether we're
+        # still in the post-connect warm-up window for prepare_on_all_hosts (see _should_prepare_on_all_hosts)
+        self._connect_time = time.time()
 
         if self.cluster.column_encryption_policy is not None:
             try:
@@ -3248,7 +3303,7 @@ def prepare(self, query, custom_payload=None, keyspace=None):
 
         self.cluster.add_prepared(response.query_id, prepared_statement)
 
-        if self.cluster.prepare_on_all_hosts:
+        if self._should_prepare_on_all_hosts():
             host = future._current_host
             try:
                 self.prepare_on_all_hosts(prepared_statement.query_string, host, prepared_keyspace)
@@ -3257,6 +3312,23 @@ def prepare(self, query, custom_payload=None, keyspace=None):
 
         return prepared_statement
 
+    def _should_prepare_on_all_hosts(self):
+        """
+        Decide whether this prepare() call should eagerly broadcast to all pooled hosts.
+
+        If the user explicitly set Cluster.prepare_on_all_hosts, that choice always wins. Otherwise, act as
+        if it were True during the post-connect warm-up window (see prepare_on_all_hosts_warmup_seconds) and
+        False afterwards.
+        """
+        cluster = self.cluster
+        if cluster._prepare_on_all_hosts_explicit:
+            return cluster.prepare_on_all_hosts
+
+        warmup_seconds = cluster.prepare_on_all_hosts_warmup_seconds
+        if not warmup_seconds:
+            return False
+        return (time.time() - self._connect_time) <= warmup_seconds
+
     def prepare_on_all_hosts(self, query, excluded_host, keyspace=None):
         """
         Prepare the given query on all hosts, excluding ``excluded_host``.

tests/integration/standard/test_query.py

@@ -527,6 +527,37 @@ def test_prepare_on_all_hosts(self):
                 session.execute(select_statement, (1, ), host=host)
             assert 2 == mock_handler.get_message_count('debug', "Re-preparing")
 
+    def test_prepare_on_all_hosts_default_and_explicit_true(self):
+        """
+        Regression test for the prepare_on_all_hosts default flip to False.
+
+        test_prepare_on_all_hosts above pins prepare_on_all_hosts=False explicitly, so it
+        can't catch a regression in the class attribute or constructor default. Disable the
+        warm-up shim (warmup_seconds=0) so the unset default is exercised deterministically,
+        and also cover the explicit True opt-in to eager preparation.
+        """
+        with MockLoggingHandler().set_module_name(cluster.__name__) as mock_handler:
+            clus = TestCluster(reprepare_on_up=False, prepare_on_all_hosts_warmup_seconds=0)
+            self.addCleanup(clus.shutdown)
+            assert clus.prepare_on_all_hosts is False
+
+            session = clus.connect(wait_for_all_pools=True)
+            select_statement = session.prepare("SELECT k FROM test3rf.test WHERE k = ?")
+            for host in clus.metadata.all_hosts():
+                session.execute(select_statement, (1, ), host=host)
+            assert 2 == mock_handler.get_message_count('debug', "Re-preparing")
+
+        with MockLoggingHandler().set_module_name(cluster.__name__) as mock_handler:
+            clus = TestCluster(prepare_on_all_hosts=True, reprepare_on_up=False)
+            self.addCleanup(clus.shutdown)
+            assert clus.prepare_on_all_hosts is True
+
+            session = clus.connect(wait_for_all_pools=True)
+            select_statement = session.prepare("SELECT k FROM test3rf.test WHERE k = ?")
+            for host in clus.metadata.all_hosts():
+                session.execute(select_statement, (1, ), host=host)
+            assert 0 == mock_handler.get_message_count('debug', "Re-preparing")
+
     def test_prepare_batch_statement(self):
         """
         Test to validate a prepared statement used inside a batch statement is correctly handled

tests/integration/standard/test_shard_aware.py

@@ -71,7 +71,10 @@ def verify_same_shard_in_tracing(self, results, shard_name):
             assert shard_name in event.thread_name
         assert 'querying locally' in "\n".join([event.description for event in events])
 
-        trace_id = results.response_future.get_query_trace_ids()[0]
+        # Use the last trace id: prepare_on_all_hosts defaults to False now, so a query
+        # against a host that hasn't prepared the statement yet can get UNPREPARED and
+        # retry, which appends an earlier, incomplete trace before the one that matters.
+        trace_id = results.response_future.get_query_trace_ids()[-1]
         traces = self.session.execute("SELECT * FROM system_traces.events WHERE session_id = %s", (trace_id,))
         events = [event for event in traces]
         for event in events:

tests/integration/standard/test_tablets.py

@@ -37,7 +37,10 @@ def verify_hosts_in_tracing(self, results, expected):
         assert len(host_set) == expected
         assert 'locally' in "\n".join([event.description for event in events])
 
-        trace_id = results.response_future.get_query_trace_ids()[0]
+        # Use the last trace id: prepare_on_all_hosts defaults to False now, so a query
+        # against a host that hasn't prepared the statement yet can get UNPREPARED and
+        # retry, which appends an earlier, incomplete trace before the one that matters.
+        trace_id = results.response_future.get_query_trace_ids()[-1]
         traces = self.session.execute("SELECT * FROM system_traces.events WHERE session_id = %s", (trace_id,))
         events = [event for event in traces]
         host_set = set()
@@ -63,7 +66,8 @@ def verify_same_shard_in_tracing(self, results):
         assert len(shard_set) == 1
         assert 'locally' in "\n".join([event.description for event in events])
 
-        trace_id = results.response_future.get_query_trace_ids()[0]
+        # See verify_hosts_in_tracing: use the last trace id, not the first.
+        trace_id = results.response_future.get_query_trace_ids()[-1]
         traces = self.session.execute("SELECT * FROM system_traces.events WHERE session_id = %s", (trace_id,))
         events = [event for event in traces]
         shard_set = set()