From 69e1c482ef1afcc2c96bb32674d5a78910ab6e17 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Tue, 7 Apr 2026 09:09:18 +0300 Subject: [PATCH 01/11] perf: remove dead and duplicate class attributes in ResultMessage Remove 'results = None' (never assigned or read in production code), duplicate 'kind = None' (declared twice), and duplicate 'paging_state = None' (declared at lines 663 and 681). The canonical declarations at lines 672-685 are kept. --- cassandra/protocol.py | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/cassandra/protocol.py b/cassandra/protocol.py index 9dfdbf3022..42007ac494 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -704,10 +704,6 @@ class ResultMessage(_MessageType): opcode = 0x08 name = 'RESULT' - kind = None - results = None - paging_state = None - # Names match type name in module scope. Most are imported from cassandra.cqltypes (except CUSTOM_TYPE) type_codes = _cqltypes_by_code = dict((v, globals()[k]) for k, v in type_codes.__dict__.items() if not k.startswith('_')) @@ -718,9 +714,8 @@ class ResultMessage(_MessageType): _CONTINUOUS_PAGING_LAST_FLAG = 0x80000000 _METADATA_ID_FLAG = 0x0008 - kind = None - # These are all the things a result message might contain. They are populated according to 'kind' + kind = None column_names = None column_types = None parsed_rows = None From 8fbf5e876a334f953b587e73d85a5bd0fcc073d4 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Tue, 7 Apr 2026 09:09:58 +0300 Subject: [PATCH 02/11] perf: skip redundant None attribute assignments in decode_message When trace_id, custom_payload, and warnings are None (the common case for non-traced, non-warned messages), skip the instance attribute assignment. The class-level defaults on _MessageType already provide None, so reading these attributes returns the correct value without the per-instance __dict__ write. --- cassandra/protocol.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/cassandra/protocol.py b/cassandra/protocol.py index 42007ac494..f17e6a26ed 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -1229,11 +1229,14 @@ def decode_message(cls, protocol_version, protocol_features, user_type_map, stre msg_class = cls.message_types_by_opcode[opcode] msg = msg_class.recv_body(body, protocol_version, protocol_features, user_type_map, result_metadata, cls.column_encryption_policy) msg.stream_id = stream_id - msg.trace_id = trace_id - msg.custom_payload = custom_payload - msg.warnings = warnings - - if msg.warnings: + if trace_id is not None: + msg.trace_id = trace_id + if custom_payload is not None: + msg.custom_payload = custom_payload + if warnings is not None: + msg.warnings = warnings + + if warnings: for w in msg.warnings: log.warning("Server warning: %s", w) From 9b6c8f3d6a584a68305ba474067006f92bd1082e Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Tue, 7 Apr 2026 09:13:38 +0300 Subject: [PATCH 03/11] perf: avoid throwaway ResultSet creation in fetch_next_page Extract _wait_for_result() from result() to return the raw result without wrapping in a ResultSet. Use it in fetch_next_page() to avoid creating a full ResultSet object just to extract _current_rows. For paged queries with N pages, this eliminates N-1 throwaway ResultSet allocations (each with 6 attributes). --- cassandra/cluster.py | 8 +++++--- tests/unit/test_resultset.py | 20 ++++++++++---------- 2 files changed, 15 insertions(+), 13 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 88c8d2707a..d111f4d63c 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -5555,9 +5555,12 @@ def result(self): ... log.exception("Operation failed:") """ + return ResultSet(self, self._wait_for_result()) + + def _wait_for_result(self): self._event.wait() if self._final_result is not _NOT_SET: - return ResultSet(self, self._final_result) + return self._final_result else: raise self._final_exception @@ -5828,8 +5831,7 @@ def fetch_next_page(self): """ if self.response_future.has_more_pages: self.response_future.start_fetching_next_page() - result = self.response_future.result() - self._current_rows = result._current_rows # ResultSet has already _set_current_rows to the appropriate form + self._set_current_rows(self.response_future._wait_for_result()) else: self._current_rows = [] diff --git a/tests/unit/test_resultset.py b/tests/unit/test_resultset.py index 80e9c21ff9..6d87fa229d 100644 --- a/tests/unit/test_resultset.py +++ b/tests/unit/test_resultset.py @@ -33,7 +33,7 @@ def test_iter_non_paged(self): def test_iter_paged(self): expected = list(range(10)) response_future = Mock(has_more_pages=True, _continuous_paging_session=None) - response_future.result.side_effect = (ResultSet(Mock(), expected[-5:]), ) # ResultSet is iterable, so it must be protected in order to be returned whole by the Mock + response_future._wait_for_result.side_effect = (expected[-5:], ) rs = ResultSet(response_future, expected[:5]) itr = iter(rs) # this is brittle, depends on internal impl details. Would like to find a better way @@ -43,11 +43,11 @@ def test_iter_paged(self): def test_iter_paged_with_empty_pages(self): expected = list(range(10)) response_future = Mock(has_more_pages=True, _continuous_paging_session=None) - response_future.result.side_effect = [ - ResultSet(Mock(), []), - ResultSet(Mock(), [0, 1, 2, 3, 4]), - ResultSet(Mock(), []), - ResultSet(Mock(), [5, 6, 7, 8, 9]), + response_future._wait_for_result.side_effect = [ + [], + [0, 1, 2, 3, 4], + [], + [5, 6, 7, 8, 9], ] rs = ResultSet(response_future, []) itr = iter(rs) @@ -65,7 +65,7 @@ def test_list_paged(self): # list access on RS for backwards-compatibility expected = list(range(10)) response_future = Mock(has_more_pages=True, _continuous_paging_session=None) - response_future.result.side_effect = (ResultSet(Mock(), expected[-5:]), ) # ResultSet is iterable, so it must be protected in order to be returned whole by the Mock + response_future._wait_for_result.side_effect = (expected[-5:], ) rs = ResultSet(response_future, expected[:5]) # this is brittle, depends on internal impl details. Would like to find a better way type(response_future).has_more_pages = PropertyMock(side_effect=(True, True, True, False)) # First two True are consumed on check entering list mode @@ -98,7 +98,7 @@ def test_iterate_then_index(self): # RuntimeError if indexing during or after pages response_future = Mock(has_more_pages=True, _continuous_paging_session=None) - response_future.result.side_effect = (ResultSet(Mock(), expected[-5:]), ) # ResultSet is iterable, so it must be protected in order to be returned whole by the Mock + response_future._wait_for_result.side_effect = (expected[-5:], ) rs = ResultSet(response_future, expected[:5]) type(response_future).has_more_pages = PropertyMock(side_effect=(True, False)) itr = iter(rs) @@ -131,7 +131,7 @@ def test_index_list_mode(self): # pages response_future = Mock(has_more_pages=True, _continuous_paging_session=None) - response_future.result.side_effect = (ResultSet(Mock(), expected[-5:]), ) # ResultSet is iterable, so it must be protected in order to be returned whole by the Mock + response_future._wait_for_result.side_effect = (expected[-5:], ) rs = ResultSet(response_future, expected[:5]) # this is brittle, depends on internal impl details. Would like to find a better way type(response_future).has_more_pages = PropertyMock(side_effect=(True, True, True, False)) # First two True are consumed on check entering list mode @@ -159,7 +159,7 @@ def test_eq(self): # pages response_future = Mock(has_more_pages=True, _continuous_paging_session=None) - response_future.result.side_effect = (ResultSet(Mock(), expected[-5:]), ) # ResultSet is iterable, so it must be protected in order to be returned whole by the Mock + response_future._wait_for_result.side_effect = (expected[-5:], ) rs = ResultSet(response_future, expected[:5]) type(response_future).has_more_pages = PropertyMock(side_effect=(True, True, True, False)) # eq before iteration causes list to be materialized From 5a700093dbdb8806181deaab2b729440dbd92cc0 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Tue, 7 Apr 2026 09:16:15 +0300 Subject: [PATCH 04/11] perf: add __slots__ to _MessageType, ResultMessage, and FastResultMessage Add __slots__ to eliminate per-instance __dict__ for ResultMessage, the most common response type. ResultMessage has 19 instance attributes; eliminating the __dict__ saves ~200-400 bytes per decoded result message. Changes: - _MessageType: __slots__ = () to signal slots-awareness to subclasses - ResultMessage: full __slots__ with all 19 instance attributes initialized in __init__ (replaces class-level defaults that are incompatible with slots) - FastResultMessage: __slots__ = () to inherit parent slots without __dict__ - _get_params: handle slotted objects gracefully for __repr__ - Remove dead 'response.results' assignment from test mock --- cassandra/protocol.py | 51 ++++++++++++++++++++---------- tests/unit/test_response_future.py | 1 - 2 files changed, 34 insertions(+), 18 deletions(-) diff --git a/cassandra/protocol.py b/cassandra/protocol.py index f17e6a26ed..59e17fefc8 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -89,6 +89,8 @@ def __init__(cls, name, bases, dct): class _MessageType(object, metaclass=_RegisterMessageType): + __slots__ = () + tracing = False custom_payload = None warnings = None @@ -108,7 +110,7 @@ def __repr__(self): def _get_params(message_obj): base_attrs = dir(_MessageType) return ( - (n, a) for n, a in message_obj.__dict__.items() + (n, a) for n, a in getattr(message_obj, '__dict__', {}).items() if n not in base_attrs and not n.startswith('_') and not callable(a) ) @@ -704,6 +706,13 @@ class ResultMessage(_MessageType): opcode = 0x08 name = 'RESULT' + __slots__ = ('kind', 'column_names', 'column_types', 'parsed_rows', + 'paging_state', 'continuous_paging_seq', 'continuous_paging_last', + 'new_keyspace', 'column_metadata', 'query_id', 'bind_metadata', + 'pk_indexes', 'schema_change_event', 'is_lwt', + 'result_metadata_id', 'stream_id', 'trace_id', + 'custom_payload', 'warnings') + # Names match type name in module scope. Most are imported from cassandra.cqltypes (except CUSTOM_TYPE) type_codes = _cqltypes_by_code = dict((v, globals()[k]) for k, v in type_codes.__dict__.items() if not k.startswith('_')) @@ -714,24 +723,31 @@ class ResultMessage(_MessageType): _CONTINUOUS_PAGING_LAST_FLAG = 0x80000000 _METADATA_ID_FLAG = 0x0008 - # These are all the things a result message might contain. They are populated according to 'kind' - kind = None - column_names = None - column_types = None - parsed_rows = None - paging_state = None - continuous_paging_seq = None - continuous_paging_last = None - new_keyspace = None - column_metadata = None - query_id = None - bind_metadata = None - pk_indexes = None - schema_change_event = None - is_lwt = False - def __init__(self, kind): self.kind = kind + self.column_names = None + self.column_types = None + self.parsed_rows = None + self.paging_state = None + self.continuous_paging_seq = None + self.continuous_paging_last = None + self.new_keyspace = None + self.column_metadata = None + self.query_id = None + self.bind_metadata = None + self.pk_indexes = None + self.schema_change_event = None + self.is_lwt = False + # result_metadata_id is intentionally NOT initialized here: it is only + # ever meaningful when the server actually sent one (PREPARED + # responses always set it explicitly; ROWS responses only set it when + # _METADATA_ID_FLAG is present). Leaving it unset lets callers detect + # "the server didn't send an id" via getattr(..., None)/hasattr rather + # than confusing that with an id that happens to be None. + self.stream_id = None + self.trace_id = None + self.custom_payload = None + self.warnings = None def recv(self, f, protocol_version, protocol_features, user_type_map, result_metadata, column_encryption_policy): if self.kind == RESULT_KIND_VOID: @@ -1268,6 +1284,7 @@ class FastResultMessage(ResultMessage): Cython version of Result Message that has a faster implementation of recv_results_row. """ + __slots__ = () # type_codes = ResultMessage.type_codes.copy() code_to_type = dict((v, k) for k, v in ResultMessage.type_codes.items()) recv_results_rows = make_recv_results_rows(colparser) diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index cf1194a91f..62ef6a822b 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -913,7 +913,6 @@ def test_repeat_orig_query_after_succesful_reprepare(self): response = Mock(spec=ResultMessage, kind=RESULT_KIND_PREPARED, result_metadata_id=b'foo') - response.results = (None, None, None, None, None) response.query_id = query_id rf._query = Mock(return_value=True) From c67594381516294604c6742f09a1c0b0dd316770 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Thu, 9 Apr 2026 15:49:25 +0300 Subject: [PATCH 05/11] perf: streamline tablet payload parsing in _set_result MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Replace double dict lookup ('in' + .get()) with single .get() + None check - Use tuple unpacking instead of three separate indexing operations - Guard add_tablet with 'if tablet is not None' (from_row can return None for empty replicas — previously this was silently passed through) - Inline single-use locals (protocol, keyspace, table) Saves ~13 ns on the tablet-hit path (noise-level, but cleaner code). --- cassandra/cluster.py | 24 ++++++++++-------------- 1 file changed, 10 insertions(+), 14 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index d111f4d63c..31f33bf04a 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -5159,21 +5159,17 @@ def _set_result(self, host, connection, pool, response): self._warnings = getattr(response, 'warnings', None) self._custom_payload = getattr(response, 'custom_payload', None) - if self._custom_payload and self.session.cluster.control_connection._tablets_routing_v1 and 'tablets-routing-v1' in self._custom_payload: - protocol = self.session.cluster.protocol_version + if self._custom_payload and self.session.cluster.control_connection._tablets_routing_v1: info = self._custom_payload.get('tablets-routing-v1') - ctype = ResponseFuture._TABLET_ROUTING_CTYPE - if ctype is None: - ctype = types.lookup_casstype('TupleType(LongType, LongType, ListType(TupleType(UUIDType, Int32Type)))') - ResponseFuture._TABLET_ROUTING_CTYPE = ctype - tablet_routing_info = ctype.from_binary(info, protocol) - first_token = tablet_routing_info[0] - last_token = tablet_routing_info[1] - tablet_replicas = tablet_routing_info[2] - tablet = Tablet.from_row(first_token, last_token, tablet_replicas) - keyspace = self.query.keyspace - table = self.query.table - self.session.cluster.metadata._tablets.add_tablet(keyspace, table, tablet) + if info is not None: + ctype = ResponseFuture._TABLET_ROUTING_CTYPE + if ctype is None: + ctype = types.lookup_casstype('TupleType(LongType, LongType, ListType(TupleType(UUIDType, Int32Type)))') + ResponseFuture._TABLET_ROUTING_CTYPE = ctype + first_token, last_token, tablet_replicas = ctype.from_binary(info, self.session.cluster.protocol_version) + tablet = Tablet.from_row(first_token, last_token, tablet_replicas) + if tablet is not None: + self.session.cluster.metadata._tablets.add_tablet(self.query.keyspace, self.query.table, tablet) if isinstance(response, ResultMessage): if response.kind == RESULT_KIND_SET_KEYSPACE: From a305ad4799cb9742c174c12bf454ccff3c43d46e Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Thu, 9 Apr 2026 16:49:48 +0300 Subject: [PATCH 06/11] perf: check isinstance(ResultMessage) first in _set_result for direct attr access Move the isinstance(response, ResultMessage) check to the top of _set_result so the hot path (successful query results) uses direct attribute access instead of getattr() with defaults. ResultMessage has trace_id, warnings, and custom_payload in __slots__, always initialised in __init__, so direct access is safe and avoids the overhead of 3 getattr() calls + 1 isinstance() that was previously done unconditionally before the type dispatch. For the cold ErrorMessage path, getattr() is still used because ErrorMessage does not declare trace_id in __slots__ or __init__. ConnectionException and generic Exception branches explicitly clear _warnings/_custom_payload to avoid stale values from prior retries. Benchmark (best-of-7, 500k iterations, Python 3.14, pure-Python): _set_result ROWS hot path (no tablets, no tracing): Before: 326 ns After: 271 ns (-55 ns, 1.20x) --- cassandra/cluster.py | 64 ++++++++++++++++++++---------- tests/unit/test_response_future.py | 12 ++++-- 2 files changed, 51 insertions(+), 25 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 31f33bf04a..115718ce2d 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -5150,28 +5150,32 @@ def _set_result(self, host, connection, pool, response): if pool and not pool.is_shutdown: pool.return_connection(connection) - trace_id = getattr(response, 'trace_id', None) - if trace_id: - if not self._query_traces: - self._query_traces = [] - self._query_traces.append(QueryTrace(trace_id, self.session)) - - self._warnings = getattr(response, 'warnings', None) - self._custom_payload = getattr(response, 'custom_payload', None) - - if self._custom_payload and self.session.cluster.control_connection._tablets_routing_v1: - info = self._custom_payload.get('tablets-routing-v1') - if info is not None: - ctype = ResponseFuture._TABLET_ROUTING_CTYPE - if ctype is None: - ctype = types.lookup_casstype('TupleType(LongType, LongType, ListType(TupleType(UUIDType, Int32Type)))') - ResponseFuture._TABLET_ROUTING_CTYPE = ctype - first_token, last_token, tablet_replicas = ctype.from_binary(info, self.session.cluster.protocol_version) - tablet = Tablet.from_row(first_token, last_token, tablet_replicas) - if tablet is not None: - self.session.cluster.metadata._tablets.add_tablet(self.query.keyspace, self.query.table, tablet) - if isinstance(response, ResultMessage): + # Hot path: ResultMessage has trace_id, warnings, and + # custom_payload in __slots__, always initialised in __init__, + # so direct attribute access is safe and faster than getattr(). + trace_id = response.trace_id + if trace_id: + if not self._query_traces: + self._query_traces = [] + self._query_traces.append(QueryTrace(trace_id, self.session)) + + self._warnings = response.warnings + custom_payload = response.custom_payload + self._custom_payload = custom_payload + + if custom_payload and self.session.cluster.control_connection._tablets_routing_v1: + info = custom_payload.get('tablets-routing-v1') + if info is not None: + ctype = ResponseFuture._TABLET_ROUTING_CTYPE + if ctype is None: + ctype = types.lookup_casstype('TupleType(LongType, LongType, ListType(TupleType(UUIDType, Int32Type)))') + ResponseFuture._TABLET_ROUTING_CTYPE = ctype + first_token, last_token, tablet_replicas = ctype.from_binary(info, self.session.cluster.protocol_version) + tablet = Tablet.from_row(first_token, last_token, tablet_replicas) + if tablet is not None: + self.session.cluster.metadata._tablets.add_tablet(self.query.keyspace, self.query.table, tablet) + if response.kind == RESULT_KIND_SET_KEYSPACE: session = getattr(self, 'session', None) if connection is not None: @@ -5234,6 +5238,18 @@ def _set_result(self, host, connection, pool, response): else: self._set_final_result(response) elif isinstance(response, ErrorMessage): + # Cold path: ErrorMessage inherits from _MessageType which + # defines warnings/custom_payload as class-level defaults but + # does NOT have trace_id -- getattr is required here. + trace_id = getattr(response, 'trace_id', None) + if trace_id: + if not self._query_traces: + self._query_traces = [] + self._query_traces.append(QueryTrace(trace_id, self.session)) + + self._warnings = getattr(response, 'warnings', None) + self._custom_payload = getattr(response, 'custom_payload', None) + retry_policy = self._retry_policy if isinstance(response, ReadTimeoutErrorMessage): @@ -5312,6 +5328,10 @@ def _set_result(self, host, connection, pool, response): self._handle_retry_decision(retry, response, host) elif isinstance(response, ConnectionException): + # ConnectionException has no trace_id/warnings/custom_payload; + # clear any stale values from a previous retry attempt. + self._warnings = None + self._custom_payload = None if self._metrics is not None: self._metrics.on_connection_error() if not isinstance(response, ConnectionShutdown): @@ -5321,6 +5341,8 @@ def _set_result(self, host, connection, pool, response): self.query, cl, error=response, retry_num=self._query_retries) self._handle_retry_decision(retry, response, host) elif isinstance(response, Exception): + self._warnings = None + self._custom_payload = None if hasattr(response, 'to_exception'): self._set_final_exception(response.to_exception()) else: diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index 62ef6a822b..3bfb16a1e3 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -79,7 +79,8 @@ def make_response_future(self, session): return ResponseFuture(session, message, query, 1) def make_mock_response(self, col_names, rows): - return Mock(spec=ResultMessage, kind=RESULT_KIND_ROWS, column_names=col_names, parsed_rows=rows, paging_state=None, col_types=None) + return Mock(spec=ResultMessage, kind=RESULT_KIND_ROWS, column_names=col_names, parsed_rows=rows, + paging_state=None, col_types=None, trace_id=None, warnings=None, custom_payload=None) def test_result_message(self): session = self.make_basic_session() @@ -122,7 +123,8 @@ def test_set_keyspace_result(self): result = Mock(spec=ResultMessage, kind=RESULT_KIND_SET_KEYSPACE, - results="keyspace1") + results="keyspace1", + trace_id=None, warnings=None, custom_payload=None) rf._set_result(None, None, None, result) rf._set_keyspace_completed({}) assert not rf.result() @@ -136,7 +138,8 @@ def test_schema_change_result(self): 'keyspace': "keyspace1", "table": "table1"} result = Mock(spec=ResultMessage, kind=RESULT_KIND_SCHEMA_CHANGE, - schema_change_event=event_results) + schema_change_event=event_results, + trace_id=None, warnings=None, custom_payload=None) connection = Mock() rf._set_result(None, connection, None, result) session.submit.assert_called_once_with(ANY, ANY, rf, connection, **event_results) @@ -145,7 +148,8 @@ def test_other_result_message_kind(self): session = self.make_session() rf = self.make_response_future(session) rf.send_request() - result = Mock(spec=ResultMessage, kind=999, results=[1, 2, 3]) + result = Mock(spec=ResultMessage, kind=999, results=[1, 2, 3], + trace_id=None, warnings=None, custom_payload=None) rf._set_result(None, None, None, result) assert rf.result()[0] == result From 667196a33e2b92f628bcfbebed1122c4fcdb4dc3 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Sat, 11 Apr 2026 00:02:18 +0300 Subject: [PATCH 07/11] perf: cache column_names/column_types on PreparedStatement to avoid per-result list comprehensions For prepared statements with skip_meta=True (the common case), column_metadata is the same object every time, yet column_names and column_types lists are rebuilt via list comprehension on every result set. Pre-compute and cache these lists on PreparedStatement at prepare time. In _set_result, use the cached lists directly instead of the per-response lists. The cache is invalidated when result_metadata is updated during re-prepare. The cached column names/types are stored together with result_metadata and result_metadata_id in one atomically-replaced tuple (PreparedStatement.result_metadata_snapshot), and a ResponseFuture reads that whole tuple in one shot at construction time, alongside the existing _bound_result_metadata snapshot. _set_result then uses the ResponseFuture's own snapshotted column names/types rather than a fresh read of prepared_statement's live cache: otherwise, if a concurrent in-flight response for the same prepared statement replaces the cache (e.g. via a METADATA_CHANGED reprepare) between this response being decoded and its callback running, the rows -- decoded against this future's own metadata snapshot -- could be paired with column names/types derived from a different schema version. _set_result must still prefer the bound snapshot only when the response itself is metadata-less. When a response DOES carry its own column_metadata (recv_results_rows only sets it when the server actually sent metadata on the wire, i.e. the METADATA_CHANGED case or the first response for a statement with no cached metadata yet), that response's rows were decoded against that fresh metadata, not against whatever this ResponseFuture had bound at construction time. Preferring the stale bound snapshot there paired freshly-decoded rows with the old column names/types; after a schema change that altered the column count, this made the row factory raise TypeError (wrong number of positional arguments). Now the response's own column_names/column_types -- derived from that same fresh column_metadata -- are used whenever the response provides them, and the bound snapshot is used only for the metadata-less/skip_meta case it was designed for. Benchmark (column_names + column_types extraction): 5 cols: 226 ns -> 30 ns (7.4x) 10 cols: 340 ns -> 28 ns (12.2x) 20 cols: 589 ns -> 31 ns (18.9x) 50 cols: 1160 ns -> 29 ns (39.6x) --- benchmarks/micro/bench_col_names_cache.py | 66 ++++++++++++++++ cassandra/cluster.py | 85 ++++++++++++++++---- cassandra/query.py | 75 ++++++++++++++---- tests/unit/test_response_future.py | 94 ++++++++++++++++++++++- 4 files changed, 290 insertions(+), 30 deletions(-) create mode 100644 benchmarks/micro/bench_col_names_cache.py diff --git a/benchmarks/micro/bench_col_names_cache.py b/benchmarks/micro/bench_col_names_cache.py new file mode 100644 index 0000000000..f7eb7c59a2 --- /dev/null +++ b/benchmarks/micro/bench_col_names_cache.py @@ -0,0 +1,66 @@ +# Copyright ScyllaDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Micro-benchmark: column_names / column_types extraction from metadata. + +Measures the cost of building [c[2] for c in metadata] and [c[3] for c in metadata] +vs using pre-cached lists (as done for prepared statements with result_metadata). + +Run: + python benchmarks/micro/bench_col_names_cache.py +""" + +import sys +import timeit + + +def make_column_metadata(ncols): + """Create fake column_metadata tuples like recv_results_metadata produces.""" + class FakeType: + pass + return [(f"ks_{i}", f"tbl_{i}", f"col_{i}", FakeType) for i in range(ncols)] + + +def bench(): + for ncols in (5, 10, 20, 50): + metadata = make_column_metadata(ncols) + + # Pre-cached (done once at prepare time) + cached_names = [c[2] for c in metadata] + cached_types = [c[3] for c in metadata] + + def extract_uncached(): + names = [c[2] for c in metadata] + types = [c[3] for c in metadata] + return names, types + + def extract_cached(): + return cached_names, cached_types + + n = 500_000 + t_uncached = timeit.timeit(extract_uncached, number=n) + t_cached = timeit.timeit(extract_cached, number=n) + + saving_ns = (t_uncached - t_cached) / n * 1e9 + speedup = t_uncached / t_cached if t_cached > 0 else float('inf') + print(f" {ncols} cols: uncached={t_uncached / n * 1e9:.1f} ns, " + f"cached={t_cached / n * 1e9:.1f} ns, " + f"saving={saving_ns:.1f} ns ({speedup:.1f}x)") + + +if __name__ == "__main__": + print(f"Python {sys.version}") + print("\n=== column_names / column_types extraction ===") + bench() diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 115718ce2d..b18e399c63 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -3047,9 +3047,13 @@ def _create_response_future(self, query, parameters, trace, custom_payload, else: timestamp = None - # Snapshot passed to the ResponseFuture for decoding skip_meta responses; only - # bound statements carry cached result metadata (set in the BoundStatement branch). + # Snapshot passed to the ResponseFuture for decoding skip_meta responses, and + # the column names/types derived from that same metadata (see + # PreparedStatement.result_metadata_snapshot); only bound statements carry + # cached result metadata (set in the BoundStatement branch). bound_result_metadata = _NOT_SET + bound_col_names = None + bound_col_types = None if isinstance(query, SimpleStatement): query_string = query.query_string @@ -3062,11 +3066,13 @@ def _create_response_future(self, query, parameters, trace, custom_payload, continuous_paging_options, statement_keyspace) elif isinstance(query, BoundStatement): prepared_statement = query.prepared_statement - # Snapshot metadata and its id as one atomic pair so the message never - # carries the id of one schema version alongside a skip_meta decision - # made for another. skip_meta is requested only when there is both an - # id to validate it with and cached metadata to decode against: while - # a statement has no cached metadata there is nothing to decode a + # Snapshot metadata, its id, and the column names/types derived from + # that metadata all together in one atomic read (result_metadata_snapshot) + # so the message never carries the id of one schema version alongside a + # skip_meta decision, or a later column-name/type lookup, made for + # another. skip_meta is requested only when there is both an id to + # validate it with and cached metadata to decode against: while a + # statement has no cached metadata there is nothing to decode a # metadata-less response with, so the server must send it. # Whether skip_meta and the id actually reach the wire is decided per # connection at serialization time (see ExecuteMessage.send_body). @@ -3074,7 +3080,8 @@ def _create_response_future(self, query, parameters, trace, custom_payload, # result_metadata=None for every page after the first (it isn't threaded # through the paging session), so a skip_meta response has nothing to # decode page 2+ against. - result_metadata, result_metadata_id = prepared_statement.result_metadata_and_id + result_metadata, result_metadata_id, bound_col_names, bound_col_types = \ + prepared_statement.result_metadata_snapshot bound_result_metadata = result_metadata message = ExecuteMessage( prepared_statement.query_id, query.values, cl, @@ -3109,7 +3116,8 @@ def _create_response_future(self, query, parameters, trace, custom_payload, self, message, query, timeout, metrics=self._metrics, prepared_statement=prepared_statement, retry_policy=retry_policy, row_factory=row_factory, load_balancer=load_balancing_policy, start_time=start_time, speculative_execution_plan=spec_exec_plan, - continuous_paging_state=None, host=host, bound_result_metadata=bound_result_metadata) + continuous_paging_state=None, host=host, bound_result_metadata=bound_result_metadata, + bound_col_names=bound_col_names, bound_col_types=bound_col_types) def get_execution_profile(self, name): """ @@ -4737,13 +4745,15 @@ class ResponseFuture(object): _control_connection_query_attempted = False _TABLET_ROUTING_CTYPE = None _bound_result_metadata = None + _bound_col_names = None + _bound_col_types = None _warned_timeout = False def __init__(self, session, message, query, timeout, metrics=None, prepared_statement=None, retry_policy=RetryPolicy(), row_factory=None, load_balancer=None, start_time=None, speculative_execution_plan=None, continuous_paging_state=None, host=None, - bound_result_metadata=_NOT_SET): + bound_result_metadata=_NOT_SET, bound_col_names=None, bound_col_types=None): self.session = session # TODO: normalize handling of retry policy and row factory self.row_factory = row_factory or session.row_factory @@ -4760,6 +4770,15 @@ def __init__(self, session, message, query, timeout, metrics=None, prepared_stat # even if a concurrent METADATA_CHANGED replaces the prepared statement's cache in # between. Defaults to [] for unprepared statements (no cached metadata). self._bound_result_metadata = [] if bound_result_metadata is _NOT_SET else bound_result_metadata + # Column names/types derived from that same result_metadata snapshot (see + # PreparedStatement.result_metadata_snapshot), captured together with it in + # one atomic read. _set_result() uses these -- rather than a fresh read of + # prepared_statement's live cache -- so a response decoded against this + # snapshot can never be paired with column names/types cached for a + # different schema version by a concurrent in-flight response for the same + # prepared statement. + self._bound_col_names = bound_col_names + self._bound_col_types = bound_col_types self._callback_lock = Lock() self._start_time = start_time or time.time() self._host = host @@ -5200,8 +5219,6 @@ def _set_result(self, host, connection, pool, response): self, connection, **response.schema_change_event) elif response.kind == RESULT_KIND_ROWS: self._paging_state = response.paging_state - self._col_names = response.column_names - self._col_types = response.column_types new_result_metadata_id = getattr(response, 'result_metadata_id', None) if self.prepared_statement and new_result_metadata_id is not None: if response.column_metadata: @@ -5210,7 +5227,9 @@ def _set_result(self, host, connection, pool, response): # new id with the old metadata (the server would then # skip sending metadata and rows would be decoded # against stale columns, with no recovery). - # (this also re-arms the anomaly warning below) + # (this also re-arms the anomaly warning below, and + # refreshes the cached column names/types together + # with the metadata they were derived from) self.prepared_statement.update_result_metadata( response.column_metadata, new_result_metadata_id) elif not self.prepared_statement._warned_missing_column_metadata: @@ -5229,10 +5248,44 @@ def _set_result(self, host, connection, pool, response): "and id are left unchanged.", getattr(self.prepared_statement, 'query_id', None) ) + # A response that carries its own column_metadata (set only + # when the server actually sent metadata on the wire -- see + # ResultMessage.recv_results_metadata's _NO_METADATA_FLAG + # check) was decoded against THAT metadata, not against + # whatever this ResponseFuture had bound at construction + # time: this is either the METADATA_CHANGED case (the + # schema changed since the statement was prepared, and the + # response's rows may have a different column count/types + # than the old snapshot) or the very first response for a + # statement with no cached metadata yet. response.column_names/ + # column_types are derived from that same response metadata + # (see recv_results_rows), so they always match + # response.parsed_rows and must be preferred here. + # + # Only when the response is metadata-less (the common, + # schema-unchanged skip_meta case) do we use the column + # names/types snapshotted on this ResponseFuture at + # construction time (see Session._create_response_future), + # to avoid rebuilding lists from metadata. Those come from + # the same atomic read as _bound_result_metadata, which is + # what a metadata-less response was actually decoded + # against; a fresh read of prepared_statement's live cache + # here could instead observe a different schema version if + # a concurrent in-flight response for the same prepared + # statement updated it (via update_result_metadata) between + # decode and this callback. + if not response.column_metadata and self._bound_col_names is not None: + col_names = self._bound_col_names + col_types = self._bound_col_types + else: + col_names = response.column_names + col_types = response.column_types + self._col_names = col_names + self._col_types = col_types if getattr(self.message, 'continuous_paging_options', None): self._handle_continuous_paging_first_response(connection, response) else: - self._set_final_result(self.row_factory(response.column_names, response.parsed_rows)) + self._set_final_result(self.row_factory(col_names, response.parsed_rows)) elif response.kind == RESULT_KIND_VOID: self._set_final_result(None) else: @@ -5405,7 +5458,9 @@ def _execute_after_prepare(self, host, connection, pool, response): # between the two PREPAREs) - a stale-but-plausible id a later # id-aware execute could send without the server detecting the # mismatch. Dropping it instead triggers the same self-healing - # b'' sentinel path a never-prepared id would. + # b'' sentinel path a never-prepared id would. This also + # refreshes the cached column names/types together with the + # metadata they were derived from. self.prepared_statement.update_result_metadata( response.column_metadata, response.result_metadata_id) diff --git a/cassandra/query.py b/cassandra/query.py index 39b9fdb0ad..2a003d91b0 100644 --- a/cassandra/query.py +++ b/cassandra/query.py @@ -451,7 +451,19 @@ class PreparedStatement(object): protocol_version = None query_id = None query_string = None - _result_metadata_and_id = (None, None) + # Cached (result_metadata, result_metadata_id, result_col_names, result_col_types) + # stored as ONE tuple, replaced atomically (single attribute assignment) by + # update_result_metadata(). Response callbacks may update a statement while + # request threads read it; keeping all four values together means a single + # attribute read always observes a self-consistent snapshot: result_col_names + # and result_col_types can never be a torn pair (one from an old assignment, + # one from a newer one), and neither can ever be paired with a + # result_metadata/result_metadata_id from a different schema version. + # Callers that need that full guarantee (e.g. a ResponseFuture snapshotting + # state for one specific in-flight response, in cassandra.cluster) should + # read result_metadata_snapshot in one shot rather than combining + # result_metadata_and_id with a separate column names/types read. + _result_metadata_and_id = (None, None, None, None) column_encryption_policy = None routing_key_indexes = None _routing_key_index_set = None @@ -471,7 +483,7 @@ def __init__(self, column_metadata, query_id, routing_key_indexes, query, self.query_string = query self.keyspace = keyspace self.protocol_version = protocol_version - self._result_metadata_and_id = (result_metadata, result_metadata_id) + self._set_result_metadata(result_metadata, result_metadata_id) self.column_encryption_policy = column_encryption_policy self.is_idempotent = False self._is_lwt = is_lwt @@ -482,12 +494,13 @@ def result_metadata_and_id(self): The cached result metadata and its metadata id as one immutable ``(result_metadata, result_metadata_id)`` pair. - Read this property when both values are needed together: the tuple is - replaced atomically by :meth:`update_result_metadata`, so a single read - can never observe the metadata of one schema version paired with the - metadata id of another. + Read this property when both values are needed together: the + underlying tuple is replaced atomically by + :meth:`update_result_metadata`, so a single read can never observe + the metadata of one schema version paired with the metadata id of + another. """ - return self._result_metadata_and_id + return self._result_metadata_and_id[:2] @property def result_metadata(self): @@ -507,21 +520,55 @@ def result_metadata_id(self): """ return self._result_metadata_and_id[1] + @property + def result_metadata_snapshot(self): + """ + Full atomic snapshot: ``(result_metadata, result_metadata_id, + result_col_names, result_col_types)``, with all four values replaced + together in a single attribute assignment by + :meth:`update_result_metadata`. + + Unlike reading :attr:`result_metadata_and_id` and the column + names/types as two separate reads, a single read of this property is + guaranteed self-consistent even if another thread's + :meth:`update_result_metadata` call (e.g. from a concurrent + in-flight response's callback) races with it: there is no window in + which the two reads could observe different schema versions. + """ + return self._result_metadata_and_id + def update_result_metadata(self, result_metadata, result_metadata_id): """ - Replace the cached result metadata and metadata id together, in a single - atomic attribute store. Response callbacks may update a statement while - request threads read it; updating the pair in one step (rather than the - two fields separately) prevents a reader from pairing a fresh metadata id - with stale metadata — a state in which the server would skip sending - metadata and rows would be decoded against the wrong columns. + Replace the cached result metadata, metadata id, and the column + names/types derived from that metadata, together in a single atomic + attribute store. Response callbacks may update a statement while + request threads read it; updating everything in one step (rather + than as separate fields) prevents a reader from pairing a fresh + metadata id with stale metadata, or column names/types cached for + one schema version with a metadata/id pair from another — a state in + which the server would skip sending metadata and rows would be + decoded against the wrong (or wrongly-named/typed) columns. Also re-arms :attr:`_warned_missing_column_metadata`, so an anomaly that recurs after the metadata was recovered is logged again. """ - self._result_metadata_and_id = (result_metadata, result_metadata_id) + self._set_result_metadata(result_metadata, result_metadata_id) self._warned_missing_column_metadata = False + def _set_result_metadata(self, result_metadata, result_metadata_id): + """ + Store result_metadata/result_metadata_id together with the column + names/types derived from that exact metadata, as one atomic tuple + (see :attr:`_result_metadata_and_id`). + """ + if result_metadata: + col_names = [c[2] for c in result_metadata] + col_types = [c[3] for c in result_metadata] + else: + col_names = None + col_types = None + self._result_metadata_and_id = (result_metadata, result_metadata_id, col_names, col_types) + @classmethod def from_message(cls, query_id, column_metadata, pk_indexes, cluster_metadata, query, prepared_keyspace, protocol_version, result_metadata, diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index 3bfb16a1e3..fb72eaadf4 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -916,7 +916,8 @@ def test_repeat_orig_query_after_succesful_reprepare(self): response = Mock(spec=ResultMessage, kind=RESULT_KIND_PREPARED, - result_metadata_id=b'foo') + result_metadata_id=b'foo', + column_metadata=[('ks', 'tb', 'col', Mock())]) response.query_id = query_id rf._query = Mock(return_value=True) @@ -1507,3 +1508,94 @@ def test_query_decodes_with_construction_snapshot_not_live_cache(self): connection.send_msg.assert_called_once() # _query decodes with the construction snapshot, not the mutated cache assert connection.send_msg.call_args.kwargs['result_metadata'] is meta_v1 + + def test_set_result_uses_construction_snapshot_for_columns_not_live_cache(self): + """ + The column names/types used to build the result set must come from the + snapshot taken when this ResponseFuture's message was constructed -- + the same snapshot the response was decoded against (see + test_query_decodes_with_construction_snapshot_not_live_cache) -- not + from a fresh read of the prepared statement's live cache. + + Simulates two in-flight responses sharing one prepared statement: this + future's message is built while the statement still has old_meta + cached, then another in-flight response's METADATA_CHANGED replaces + the statement's cache with new_meta *before* this future's own + response callback runs. Without a per-future snapshot, the rows + (decoded against old_meta, per the previous test) would be paired + with column names/types derived from new_meta instead. + """ + old_meta = [('ks', 'tb', 'old_col', Mock())] + new_meta = [('ks', 'tb', 'new_col', Mock())] + ps = self._make_prepared_statement(old_meta, b'id1') + + # Message built (and metadata + column names/types snapshotted + # together, atomically) while ps still has old_meta cached. + rf = self._create_execute_future(ps) + assert rf._bound_result_metadata is old_meta + assert rf._bound_col_names == ['old_col'] + assert rf._bound_col_types == [old_meta[0][3]] + + # Another in-flight response for the same prepared statement completes + # a METADATA_CHANGED update before this future's callback runs. + ps.update_result_metadata(new_meta, b'id2') + assert ps.result_metadata is new_meta + + # This response was decoded against old_meta (the skip_meta path: no + # column_metadata and no new result_metadata_id on the response + # itself -- the server didn't resend metadata). + response = self._make_rows_response(result_metadata_id=None, column_metadata=None) + response.column_names = ['SHOULD_NOT_BE_USED'] + response.column_types = ['SHOULD_NOT_BE_USED'] + rf._set_result(None, None, None, response) + + # Must reflect old_meta -- what the rows were actually decoded + # against -- never new_meta, the prepared statement's current, + # since-mutated cache. + assert rf._col_names == ['old_col'] + assert rf._col_types == [old_meta[0][3]] + + def test_set_result_uses_response_metadata_when_metadata_changed(self): + """ + The inverse of test_set_result_uses_construction_snapshot_for_columns_not_live_cache: + when THIS response is itself the METADATA_CHANGED response -- it carries + fresh column_metadata -- its rows were decoded (by + ResultMessage.recv_results_rows) against that fresh metadata, not + against the bound snapshot taken when this ResponseFuture's message was + constructed. response.column_names/column_types are derived from that + same fresh column_metadata, so _set_result must prefer them over the + now-stale bound snapshot. + + Regression test: previously _set_result preferred the bound snapshot + whenever one existed, regardless of whether the response carried its + own metadata. After a schema change that altered the column count, + this paired a changed row shape with the old column-name list, which + made the row factory raise TypeError (wrong number of positional + arguments). + """ + old_meta = [('ks', 'tb', 'old_col', Mock())] + new_meta = [('ks', 'tb', 'a', Mock()), ('ks', 'tb', 'b', Mock())] + ps = self._make_prepared_statement(old_meta, b'id1') + + # Message built (and metadata + column names/types snapshotted + # together, atomically) while ps still has old_meta cached -- one + # column. + rf = self._create_execute_future(ps) + assert rf._bound_col_names == ['old_col'] + + # This response IS the METADATA_CHANGED response: the schema changed + # server-side since prepare, so it carries fresh column_metadata (two + # columns) and a new result_metadata_id, and its rows were decoded + # against that fresh metadata -- not the one-column bound snapshot. + response = self._make_rows_response(result_metadata_id=b'id2', column_metadata=new_meta) + response.column_names = ['a', 'b'] + response.column_types = [new_meta[0][3], new_meta[1][3]] + response.parsed_rows = [(1, 2)] + + rf._set_result(None, None, None, response) + + # Must reflect the response's own fresh metadata -- what its rows were + # actually decoded against -- never the stale one-column bound + # snapshot, which would mismatch the two-value rows. + assert rf._col_names == ['a', 'b'] + assert rf._col_types == [new_meta[0][3], new_meta[1][3]] From 61c6abea2b2765da9e899d5d00111397407786e8 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Sat, 11 Apr 2026 00:28:22 +0300 Subject: [PATCH 08/11] perf: reorder RESULT_KIND dispatch and replace getattr with direct access Two micro-optimizations in _set_result() hot path: 1. Reorder the RESULT_KIND if/elif chain to check ROWS first (was third), since it is by far the most common result type. VOID is second. SET_KEYSPACE and SCHEMA_CHANGE (rare) are now last. 2. Add continuous_paging_options = None class attribute to _QueryMessage, allowing direct attribute access instead of getattr(self.message, 'continuous_paging_options', None). Benchmark (2M iters, Python 3.14): RESULT_KIND reorder: 35.5 -> 24.3 ns (1.46x, -11.2 ns/dispatch) getattr -> direct: 32.0 -> 18.3 ns (1.75x, -13.7 ns/access) Combined: ~25 ns saved per query --- .../micro/bench_result_kind_dispatch.py | 115 ++++++++++++++++++ cassandra/cluster.py | 48 ++++---- cassandra/protocol.py | 10 ++ 3 files changed, 149 insertions(+), 24 deletions(-) create mode 100644 benchmarks/micro/bench_result_kind_dispatch.py diff --git a/benchmarks/micro/bench_result_kind_dispatch.py b/benchmarks/micro/bench_result_kind_dispatch.py new file mode 100644 index 0000000000..8595c57ce6 --- /dev/null +++ b/benchmarks/micro/bench_result_kind_dispatch.py @@ -0,0 +1,115 @@ +# Copyright ScyllaDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Micro-benchmark: RESULT_KIND dispatch ordering and getattr vs direct access. + +Measures the cost difference between: +1. Checking RESULT_KIND_ROWS first vs third in the if/elif chain +2. getattr(msg, 'continuous_paging_options', None) vs msg.continuous_paging_options + +Run: + python benchmarks/micro/bench_result_kind_dispatch.py +""" + +import sys +import timeit + + +def bench(): + n = 2_000_000 + + # Simulate the result kind values + RESULT_KIND_SET_KEYSPACE = 0x0003 + RESULT_KIND_SCHEMA_CHANGE = 0x0005 + RESULT_KIND_ROWS = 0x0002 + RESULT_KIND_VOID = 0x0001 + + kind = RESULT_KIND_ROWS # the common case + + # Old order: SET_KEYSPACE, SCHEMA_CHANGE, ROWS, VOID + def old_dispatch(): + if kind == RESULT_KIND_SET_KEYSPACE: + return 'set_keyspace' + elif kind == RESULT_KIND_SCHEMA_CHANGE: + return 'schema_change' + elif kind == RESULT_KIND_ROWS: + return 'rows' + elif kind == RESULT_KIND_VOID: + return 'void' + + # New order: ROWS, VOID, SET_KEYSPACE, SCHEMA_CHANGE + def new_dispatch(): + if kind == RESULT_KIND_ROWS: + return 'rows' + elif kind == RESULT_KIND_VOID: + return 'void' + elif kind == RESULT_KIND_SET_KEYSPACE: + return 'set_keyspace' + elif kind == RESULT_KIND_SCHEMA_CHANGE: + return 'schema_change' + + print(f"=== RESULT_KIND dispatch order ({n:,} iters) ===\n") + + # Warmup + for _ in range(10000): + old_dispatch() + new_dispatch() + + t_old = timeit.timeit(old_dispatch, number=n) + t_new = timeit.timeit(new_dispatch, number=n) + ns_old = t_old / n * 1e9 + ns_new = t_new / n * 1e9 + saving = ns_old - ns_new + speedup = ns_old / ns_new if ns_new > 0 else float('inf') + print(f" Old (ROWS=3rd): {ns_old:.1f} ns") + print(f" New (ROWS=1st): {ns_new:.1f} ns") + print(f" Saving: {saving:.1f} ns ({speedup:.2f}x)") + + # getattr vs direct attribute access + print(f"\n=== getattr vs direct attribute access ({n:,} iters) ===\n") + + class OldMsg: + pass + + class NewMsg: + continuous_paging_options = None + + old_msg = OldMsg() + new_msg = NewMsg() + + def old_getattr(): + return getattr(old_msg, 'continuous_paging_options', None) + + def new_direct(): + return new_msg.continuous_paging_options + + for _ in range(10000): + old_getattr() + new_direct() + + t_old = timeit.timeit(old_getattr, number=n) + t_new = timeit.timeit(new_direct, number=n) + ns_old = t_old / n * 1e9 + ns_new = t_new / n * 1e9 + saving = ns_old - ns_new + speedup = ns_old / ns_new if ns_new > 0 else float('inf') + print(f" getattr(msg, 'continuous_paging_options', None): {ns_old:.1f} ns") + print(f" msg.continuous_paging_options: {ns_new:.1f} ns") + print(f" Saving: {saving:.1f} ns ({speedup:.2f}x)") + + +if __name__ == "__main__": + print(f"Python {sys.version}\n") + bench() diff --git a/cassandra/cluster.py b/cassandra/cluster.py index b18e399c63..6b58822ead 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -5195,29 +5195,7 @@ def _set_result(self, host, connection, pool, response): if tablet is not None: self.session.cluster.metadata._tablets.add_tablet(self.query.keyspace, self.query.table, tablet) - if response.kind == RESULT_KIND_SET_KEYSPACE: - session = getattr(self, 'session', None) - if connection is not None: - connection.keyspace = response.new_keyspace - # since we're running on the event loop thread, we need to - # use a non-blocking method for setting the keyspace on - # all connections in this session, otherwise the event - # loop thread will deadlock waiting for keyspaces to be - # set. This uses a callback chain which ends with - # self._set_keyspace_completed() being called in the - # event loop thread. - if session: - session._set_keyspace_for_all_pools( - response.new_keyspace, self._set_keyspace_completed) - elif response.kind == RESULT_KIND_SCHEMA_CHANGE: - # refresh the schema before responding, but do it in another - # thread instead of the event loop thread - self.is_schema_agreed = False - self.session.submit( - refresh_schema_and_set_result, - self.session.cluster.control_connection, - self, connection, **response.schema_change_event) - elif response.kind == RESULT_KIND_ROWS: + if response.kind == RESULT_KIND_ROWS: self._paging_state = response.paging_state new_result_metadata_id = getattr(response, 'result_metadata_id', None) if self.prepared_statement and new_result_metadata_id is not None: @@ -5282,12 +5260,34 @@ def _set_result(self, host, connection, pool, response): col_types = response.column_types self._col_names = col_names self._col_types = col_types - if getattr(self.message, 'continuous_paging_options', None): + if self.message.continuous_paging_options: self._handle_continuous_paging_first_response(connection, response) else: self._set_final_result(self.row_factory(col_names, response.parsed_rows)) elif response.kind == RESULT_KIND_VOID: self._set_final_result(None) + elif response.kind == RESULT_KIND_SET_KEYSPACE: + session = getattr(self, 'session', None) + if connection is not None: + connection.keyspace = response.new_keyspace + # since we're running on the event loop thread, we need to + # use a non-blocking method for setting the keyspace on + # all connections in this session, otherwise the event + # loop thread will deadlock waiting for keyspaces to be + # set. This uses a callback chain which ends with + # self._set_keyspace_completed() being called in the + # event loop thread. + if session: + session._set_keyspace_for_all_pools( + response.new_keyspace, self._set_keyspace_completed) + elif response.kind == RESULT_KIND_SCHEMA_CHANGE: + # refresh the schema before responding, but do it in another + # thread instead of the event loop thread + self.is_schema_agreed = False + self.session.submit( + refresh_schema_and_set_result, + self.session.cluster.control_connection, + self, connection, **response.schema_change_event) else: self._set_final_result(response) elif isinstance(response, ErrorMessage): diff --git a/cassandra/protocol.py b/cassandra/protocol.py index 59e17fefc8..cc0d8a2f80 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -547,6 +547,11 @@ def recv_body(cls, f, *args): class _QueryMessage(_MessageType): + # DSE continuous paging: stored when the feature is active, otherwise None. + # Declared as a class attribute so that callers can use direct attribute + # access instead of getattr(msg, 'continuous_paging_options', None). + continuous_paging_options = None + def __init__(self, query_params, consistency_level, serial_consistency_level=None, fetch_size=None, paging_state=None, timestamp=None, skip_meta=False, @@ -958,6 +963,11 @@ class BatchMessage(_MessageType): opcode = 0x0D name = 'BATCH' + # Batch messages never use continuous paging, but callers access this + # attribute directly (instead of getattr) for speed. Declare it here so + # that BatchMessage matches the same interface as _QueryMessage. + continuous_paging_options = None + def __init__(self, batch_type, queries, consistency_level, serial_consistency_level=None, timestamp=None, keyspace=None): From 8ed974d5d941d33695d02c48bfc2e296c40c09a4 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Sat, 11 Apr 2026 13:43:54 +0300 Subject: [PATCH 09/11] perf: cache session.cluster as local in _set_result to avoid repeated double-lookup In the ResultMessage hot path, self.session.cluster was accessed 3 times in the tablet routing block plus additional times in SET_KEYSPACE and SCHEMA_CHANGE branches. Cache session = self.session and cluster = session.cluster once at entry to eliminate redundant attribute-chain lookups. Also reuse the cached 'session' local for the SET_KEYSPACE and SCHEMA_CHANGE branches instead of re-reading self.session. Keep updating this response's own connection's keyspace directly in the SET_KEYSPACE branch, in addition to (not instead of) propagating to session pools via _set_keyspace_for_all_pools: that call only updates session.keyspace and pooled connections, so without the direct update, a control-connection fallback (pool is None, no pooled connection available) would leave its connection's keyspace stale and subsequent fallback queries on it would run against the wrong keyspace. Benchmark (5M iters): 3x self.session.cluster (old): 66.2 ns 1x local + 3x local (new): 39.9 ns Saving: 26.3 ns (1.66x) --- .../micro/bench_session_cluster_cache.py | 86 +++++++++++++++++++ cassandra/cluster.py | 24 ++++-- 2 files changed, 103 insertions(+), 7 deletions(-) create mode 100644 benchmarks/micro/bench_session_cluster_cache.py diff --git a/benchmarks/micro/bench_session_cluster_cache.py b/benchmarks/micro/bench_session_cluster_cache.py new file mode 100644 index 0000000000..c0ea7e166f --- /dev/null +++ b/benchmarks/micro/bench_session_cluster_cache.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python +# Copyright ScyllaDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Benchmark: caching self.session.cluster as a local variable. + +Measures the cost of repeated self.session.cluster double-lookups +vs. a single local assignment. +""" + +import sys +import time + +ITERS = 5_000_000 + + +class FakeCluster: + class control_connection: + _tablets_routing_v1 = True + protocol_version = 5 + class metadata: + class _tablets: + @staticmethod + def add_tablet(ks, tbl, tablet): + pass + + +class FakeSession: + cluster = FakeCluster() + + +class FakeResponseFuture: + def __init__(self): + self.session = FakeSession() + + +def bench_double_lookup(rf, n): + """Simulates 3 accesses to self.session.cluster (tablet routing block).""" + t0 = time.perf_counter_ns() + for _ in range(n): + _ = rf.session.cluster.control_connection + _ = rf.session.cluster.protocol_version + _ = rf.session.cluster.metadata + return (time.perf_counter_ns() - t0) / n + + +def bench_cached_local(rf, n): + """Simulates caching session.cluster in a local.""" + t0 = time.perf_counter_ns() + for _ in range(n): + cluster = rf.session.cluster + _ = cluster.control_connection + _ = cluster.protocol_version + _ = cluster.metadata + return (time.perf_counter_ns() - t0) / n + + +def main(): + print(f"Python {sys.version}\n") + rf = FakeResponseFuture() + + ns_old = bench_double_lookup(rf, ITERS) + ns_new = bench_cached_local(rf, ITERS) + saving = ns_old - ns_new + speedup = ns_old / ns_new if ns_new else float('inf') + + print(f"=== self.session.cluster caching ({ITERS:,} iters) ===\n") + print(f" 3x self.session.cluster (old): {ns_old:.1f} ns") + print(f" 1x local + 3x local (new): {ns_new:.1f} ns") + print(f" Saving: {saving:.1f} ns ({speedup:.2f}x)") + + +if __name__ == "__main__": + main() diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 6b58822ead..801fcc2130 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -5174,26 +5174,30 @@ def _set_result(self, host, connection, pool, response): # custom_payload in __slots__, always initialised in __init__, # so direct attribute access is safe and faster than getattr(). trace_id = response.trace_id + session = self.session if trace_id: if not self._query_traces: self._query_traces = [] - self._query_traces.append(QueryTrace(trace_id, self.session)) + self._query_traces.append(QueryTrace(trace_id, session)) self._warnings = response.warnings custom_payload = response.custom_payload self._custom_payload = custom_payload - if custom_payload and self.session.cluster.control_connection._tablets_routing_v1: + # Cache session.cluster to avoid repeated double-lookup in the + # tablet routing block (3 accesses) and schema-change path. + cluster = session.cluster + if custom_payload and cluster.control_connection._tablets_routing_v1: info = custom_payload.get('tablets-routing-v1') if info is not None: ctype = ResponseFuture._TABLET_ROUTING_CTYPE if ctype is None: ctype = types.lookup_casstype('TupleType(LongType, LongType, ListType(TupleType(UUIDType, Int32Type)))') ResponseFuture._TABLET_ROUTING_CTYPE = ctype - first_token, last_token, tablet_replicas = ctype.from_binary(info, self.session.cluster.protocol_version) + first_token, last_token, tablet_replicas = ctype.from_binary(info, cluster.protocol_version) tablet = Tablet.from_row(first_token, last_token, tablet_replicas) if tablet is not None: - self.session.cluster.metadata._tablets.add_tablet(self.query.keyspace, self.query.table, tablet) + cluster.metadata._tablets.add_tablet(self.query.keyspace, self.query.table, tablet) if response.kind == RESULT_KIND_ROWS: self._paging_state = response.paging_state @@ -5267,7 +5271,13 @@ def _set_result(self, host, connection, pool, response): elif response.kind == RESULT_KIND_VOID: self._set_final_result(None) elif response.kind == RESULT_KIND_SET_KEYSPACE: - session = getattr(self, 'session', None) + # Update this response's own connection directly and + # immediately. This matters most when pool is None (the + # control-connection fallback case): _set_keyspace_for_all_pools + # below only updates session.keyspace and pooled connections, so + # without this the control connection's fallback connection would + # keep its old keyspace and subsequent fallback queries would run + # against the wrong keyspace. if connection is not None: connection.keyspace = response.new_keyspace # since we're running on the event loop thread, we need to @@ -5284,9 +5294,9 @@ def _set_result(self, host, connection, pool, response): # refresh the schema before responding, but do it in another # thread instead of the event loop thread self.is_schema_agreed = False - self.session.submit( + session.submit( refresh_schema_and_set_result, - self.session.cluster.control_connection, + cluster.control_connection, self, connection, **response.schema_change_event) else: self._set_final_result(response) From 45ddcb230d5b88450de4c5ead5210b26f575ae1c Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Mon, 29 Jun 2026 22:54:08 +0300 Subject: [PATCH 10/11] perf: add __repr__ to ResultMessage for slotted attribute display --- cassandra/protocol.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/cassandra/protocol.py b/cassandra/protocol.py index cc0d8a2f80..8e3edc989d 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -754,6 +754,19 @@ def __init__(self, kind): self.custom_payload = None self.warnings = None + def __repr__(self): + slots = set() + for cls in type(self).__mro__: + slots.update(getattr(cls, '__slots__', ())) + params = ', '.join( + '%s=%r' % (s, getattr(self, s, None)) + for s in slots + if not s.startswith('_') + and getattr(self, s, None) is not None + and getattr(self, s, None) is not False + ) + return '<%s(%s)>' % (self.__class__.__name__, params) + def recv(self, f, protocol_version, protocol_features, user_type_map, result_metadata, column_encryption_policy): if self.kind == RESULT_KIND_VOID: return From 5044031954ac1bfd65c0e95882cd10faa1582da6 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Mon, 29 Jun 2026 22:54:09 +0300 Subject: [PATCH 11/11] test: keep response-future mocks faithful to the real production behavior The _set_keyspace_for_all_pools error-reporting fix and the wait_for_schema_agreement scope validation this commit originally carried are now already present on master (independently landed as 26c201a74 and 03c4c9601). Rebasing onto master leaves this commit with only test updates. Two mocks needed correcting rather than adding to: - test_control_connection_fallback_reprepares_prepared_statement's prepared_statement Mock no longer needs _result_col_names/ _result_col_types: the column-name/type cache these referred to has since been redesigned as one atomic tuple exposed via PreparedStatement.result_metadata_snapshot, and _set_result now reads a ResponseFuture-level snapshot of it rather than these attributes directly, so setting them on the mock no longer serves any purpose. - test_control_connection_fallback_updates_connection_keyspace's set_keyspace_for_all_pools stub must NOT update `connection.keyspace` itself: the real Session._set_keyspace_for_all_pools only ever updates session.keyspace and pooled connections, never an arbitrary connection passed in by the caller. A stub that updates `connection` directly would mask a regression where _set_result stops updating a fallback control connection's keyspace on its own -- exactly the class of bug this test exists to catch. --- tests/unit/test_response_future.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/unit/test_response_future.py b/tests/unit/test_response_future.py index fb72eaadf4..80bf0d0a09 100644 --- a/tests/unit/test_response_future.py +++ b/tests/unit/test_response_future.py @@ -433,6 +433,13 @@ def test_control_connection_fallback_updates_connection_keyspace(self): session.cluster._default_load_balancing_policy.make_query_plan.return_value = ['ip1'] session._pools = {} + # Faithful to the real Session._set_keyspace_for_all_pools: it only + # updates session.keyspace and pooled connections (none here, since + # _pools is empty) -- it has no way to reach an out-of-pool + # connection such as the control connection's fallback connection. + # If this mock updated `connection` itself, it would mask a + # regression where _set_result stops updating a fallback control + # connection's keyspace directly. def set_keyspace_for_all_pools(keyspace, callback): session.keyspace = keyspace callback({})