Change prepare_on_all_hosts default to False

CHANGELOG.rst

@@ -27,6 +27,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

@@ -984,13 +984,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
@@ -1204,7 +1248,8 @@ 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,
+                 prepare_on_all_hosts_warmup_seconds=15,
                  reprepare_on_up=True,
                  execution_profiles=None,
                  allow_beta_protocol_version=False,
@@ -1480,7 +1525,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)
 
@@ -2642,6 +2692,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:
@@ -3238,7 +3291,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)
@@ -3247,6 +3300,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_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()

tests/unit/test_cluster.py

@@ -16,6 +16,7 @@
 from concurrent.futures import Future
 import logging
 import socket
+import time
 from types import SimpleNamespace
 
 from unittest.mock import patch, Mock
@@ -381,6 +382,115 @@ def test_connection_factory_ignores_a_caller_supplied_session_id_and_reporter(se
         assert factory.call_args.kwargs['driver_config_reporter'] is None
 
 
+class PrepareOnAllHostsWarmupTest(unittest.TestCase):
+    """
+    Covers the post-connect warm-up window that decides whether Session.prepare()
+    eagerly broadcasts to all pooled hosts when Cluster.prepare_on_all_hosts was
+    left unset. See Session._should_prepare_on_all_hosts.
+    """
+
+    def _make_session(self, **cluster_kwargs):
+        cluster = Cluster(**cluster_kwargs)
+        self.addCleanup(cluster.shutdown)
+        host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())
+        host.set_up()
+        cluster.metadata.add_or_return_host(host)
+        return Session(cluster, [host])
+
+    @mock_session_pools
+    def test_within_warmup_window_prepares_eagerly_by_default(self, *_):
+        session = self._make_session()
+        session._connect_time = time.time()
+
+        assert session._should_prepare_on_all_hosts() is True
+
+    @mock_session_pools
+    def test_after_warmup_window_falls_back_to_lazy_by_default(self, *_):
+        session = self._make_session()
+        session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1
+
+        assert session._should_prepare_on_all_hosts() is False
+
+    @mock_session_pools
+    def test_explicit_true_is_respected_even_after_warmup_elapses(self, *_):
+        session = self._make_session(prepare_on_all_hosts=True)
+        session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1
+
+        assert session._should_prepare_on_all_hosts() is True
+
+    @mock_session_pools
+    def test_explicit_false_is_respected_even_within_warmup_window(self, *_):
+        session = self._make_session(prepare_on_all_hosts=False)
+        session._connect_time = time.time()
+
+        assert session._should_prepare_on_all_hosts() is False
+
+    @mock_session_pools
+    def test_runtime_assignment_after_construction_is_respected(self, *_):
+        session = self._make_session()
+        session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1
+
+        session.cluster.prepare_on_all_hosts = True
+        assert session._should_prepare_on_all_hosts() is True
+
+        session._connect_time = time.time()
+        session.cluster.prepare_on_all_hosts = False
+        assert session._should_prepare_on_all_hosts() is False
+
+    @mock_session_pools
+    def test_zero_warmup_seconds_disables_eager_behavior(self, *_):
+        session = self._make_session(prepare_on_all_hosts_warmup_seconds=0)
+        session._connect_time = time.time()
+
+        assert session._should_prepare_on_all_hosts() is False
+
+    @mock_session_pools
+    def test_prepare_uses_should_prepare_on_all_hosts_decision(self, *_):
+        session = self._make_session()
+        session._connect_time = time.time()
+
+        message = Mock(query_id=b'qid', bind_metadata=[], pk_indexes=[], column_metadata=[],
+                        result_metadata_id=None, is_lwt=False)
+        future = Mock()
+        future.result.return_value.one.return_value = message
+        future._current_host = Host("127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4())
+
+        with patch('cassandra.cluster.ResponseFuture', return_value=future), \
+             patch.object(session.cluster, 'add_prepared'), \
+             patch.object(Session, 'prepare_on_all_hosts') as prepare_on_all_hosts:
+            session.prepare("SELECT * FROM t")
+
+        assert prepare_on_all_hosts.call_count == 1
+
+        session._connect_time = time.time() - session.cluster.prepare_on_all_hosts_warmup_seconds - 1
+        with patch('cassandra.cluster.ResponseFuture', return_value=future), \
+             patch.object(session.cluster, 'add_prepared'), \
+             patch.object(Session, 'prepare_on_all_hosts') as prepare_on_all_hosts:
+            session.prepare("SELECT * FROM t")
+
+        assert prepare_on_all_hosts.call_count == 0
+
+    @mock_session_pools
+    def test_prepare_all_queries_on_host_up_is_unaffected_by_flag_or_warmup(self, *_):
+        # Cluster._prepare_all_queries (the reprepare_on_up path for late-joining hosts)
+        # is a separate mechanism from prepare_on_all_hosts/warmup and must keep firing
+        # regardless of either.
+        session = self._make_session(prepare_on_all_hosts=False, prepare_on_all_hosts_warmup_seconds=0)
+        session._connect_time = time.time() - 1000
+        cluster = session.cluster
+
+        prepared_statement = Mock(query_string="SELECT * FROM t", keyspace=None)
+        cluster._prepared_statements = {b'qid': prepared_statement}
+
+        new_host = Host("127.0.0.2", SimpleConvictionPolicy, host_id=uuid.uuid4())
+        new_host.set_up()
+
+        with patch.object(cluster, 'connection_factory') as connection_factory:
+            cluster._prepare_all_queries(new_host)
+
+        assert connection_factory.call_count == 1
+
+
 class SchedulerTest(unittest.TestCase):
     # TODO: this suite could be expanded; for now just adding a test covering a ticket