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/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/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 88c8d2707a..801fcc2130 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 @@ -5150,58 +5169,38 @@ 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 and 'tablets-routing-v1' in self._custom_payload: - protocol = self.session.cluster.protocol_version - 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 isinstance(response, ResultMessage): - 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: + # 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 + session = self.session + if trace_id: + if not self._query_traces: + self._query_traces = [] + self._query_traces.append(QueryTrace(trace_id, session)) + + self._warnings = response.warnings + custom_payload = response.custom_payload + self._custom_payload = custom_payload + + # 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, cluster.protocol_version) + tablet = Tablet.from_row(first_token, last_token, tablet_replicas) + if tablet is not None: + cluster.metadata._tablets.add_tablet(self.query.keyspace, self.query.table, tablet) + + if 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 +5209,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,15 +5230,89 @@ def _set_result(self, host, connection, pool, response): "and id are left unchanged.", getattr(self.prepared_statement, 'query_id', None) ) - if getattr(self.message, 'continuous_paging_options', 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 self.message.continuous_paging_options: 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) + elif response.kind == RESULT_KIND_SET_KEYSPACE: + # 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 + # 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 + session.submit( + refresh_schema_and_set_result, + cluster.control_connection, + self, connection, **response.schema_change_event) 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): @@ -5316,6 +5391,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): @@ -5325,6 +5404,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: @@ -5387,7 +5468,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) @@ -5555,9 +5638,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 +5914,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/cassandra/protocol.py b/cassandra/protocol.py index 9dfdbf3022..8e3edc989d 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) ) @@ -545,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, @@ -704,9 +711,12 @@ class ResultMessage(_MessageType): opcode = 0x08 name = 'RESULT' - kind = None - results = None - paging_state = None + __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('_')) @@ -718,25 +728,44 @@ 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' - 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 __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: @@ -947,6 +976,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): @@ -1234,11 +1268,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) @@ -1270,6 +1307,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/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 cf1194a91f..80bf0d0a09 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 @@ -429,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({}) @@ -912,8 +923,8 @@ 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) + result_metadata_id=b'foo', + column_metadata=[('ks', 'tb', 'col', Mock())]) response.query_id = query_id rf._query = Mock(return_value=True) @@ -1504,3 +1515,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]] 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