From f183fbe2255cc630469faa464cea04826f05bdf6 Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Wed, 19 Aug 2026 10:49:49 +0300 Subject: [PATCH 1/3] Change prepare_on_all_hosts default to False In multi-DC deployments, eager preparation previously ran on every host with an open connection pool, including remote hosts that are rarely or never queried on the happy path. Since an UNPREPARED response always triggers on-demand reprepare and retry, eager preparation is purely a latency optimization and does not affect correctness. Disabling it by default avoids paying for that optimization on rarely-used remote hosts. --- CHANGELOG.rst | 4 ++++ cassandra/cluster.py | 9 ++++++--- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index 068be2e048..d136210713 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -27,6 +27,10 @@ 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 ``False``. In multi-DC deployments eager + preparation previously ran on every pooled host, including remote hosts that are rarely or + never queried. Disabling it by default avoids that waste; an ``UNPREPARED`` response still + triggers on-demand reprepare and retry, so correctness is unaffected. * ``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 diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 7260bd08b6..07edd757e2 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -984,11 +984,14 @@ 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 """ Specifies whether statements should be prepared on all hosts, or just one. - This can reasonably be disabled on long-running applications with numerous clients preparing statements on startup, + 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. """ @@ -1204,7 +1207,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=False, reprepare_on_up=True, execution_profiles=None, allow_beta_protocol_version=False, From b23c791a6cbc41908a250790450a2b72d5ed1c94 Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Sun, 23 Aug 2026 10:16:40 +0300 Subject: [PATCH 2/3] tests: use the last query trace id, not the first prepare_on_all_hosts now defaults to False, so a query against a host that hasn't prepared the statement yet can get UNPREPARED and silently retry. That appends an earlier, incomplete trace to the response future before the one that actually produced the result, so indexing get_query_trace_ids() with [0] picked the wrong (empty) trace. get_query_trace() (singular) already uses the last trace; match that. --- tests/integration/standard/test_shard_aware.py | 5 ++++- tests/integration/standard/test_tablets.py | 8 ++++++-- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/tests/integration/standard/test_shard_aware.py b/tests/integration/standard/test_shard_aware.py index 6daba6e26f..654ddb20df 100644 --- a/tests/integration/standard/test_shard_aware.py +++ b/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: diff --git a/tests/integration/standard/test_tablets.py b/tests/integration/standard/test_tablets.py index 0491b15f3f..4a37b7a84f 100644 --- a/tests/integration/standard/test_tablets.py +++ b/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() From a032aa436e008fb20628b1f4f1ab54041e7a130b Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Sun, 23 Aug 2026 19:05:04 +0300 Subject: [PATCH 3/3] Add a post-connect warm-up window for prepare_on_all_hosts Broadcasting a PREPARE to every pooled host on every prepare() call is mainly useful right after a Session connects, when hosts are freshly discovered and different callers are likely to hit different hosts in quick succession. In steady state, traffic for a given prepared statement usually concentrates on a stable subset of replicas via token-aware routing, so the broadcast is normally wasted work, and an UNPREPARED response already triggers reprepare-and-retry on demand. Leave prepare_on_all_hosts unset (Cluster's new sentinel default) to get eager broadcast only during prepare_on_all_hosts_warmup_seconds (15s) after Session connects, then fall back to the lazy behavior. Explicitly passing True or False still pins that behavior for the life of the cluster, unaffected by the warm-up window. This is independent of Cluster._prepare_all_queries/reprepare_on_up, which already handles hosts that join after a statement was prepared. --- CHANGELOG.rst | 15 +++-- cassandra/cluster.py | 89 ++++++++++++++++++++++++++---- tests/unit/test_cluster.py | 110 +++++++++++++++++++++++++++++++++++++ 3 files changed, 199 insertions(+), 15 deletions(-) diff --git a/CHANGELOG.rst b/CHANGELOG.rst index d136210713..b5dc522e7f 100644 --- a/CHANGELOG.rst +++ b/CHANGELOG.rst @@ -27,10 +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 ``False``. In multi-DC deployments eager - preparation previously ran on every pooled host, including remote hosts that are rarely or - never queried. Disabling it by default avoids that waste; an ``UNPREPARED`` response still - triggers on-demand reprepare and retry, so correctness is unaffected. +* ``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 diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 07edd757e2..d0cc9a07dc 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -984,16 +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 = False + _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``). - 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. + 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 @@ -1207,7 +1248,8 @@ def __init__(self, schema_metadata_page_size=1000, address_translator=None, status_event_refresh_window=2, - prepare_on_all_hosts=False, + 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, @@ -1483,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) @@ -2645,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: @@ -3241,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) @@ -3250,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``. diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 35dc354465..ad5ba47a16 100644 --- a/tests/unit/test_cluster.py +++ b/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