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.

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)

Author: mykaul

benchmarks/bench_session_cluster_cache.py

@@ -0,0 +1,72 @@
+#!/usr/bin/env python
+"""
+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()

cassandra/cluster.py

@@ -4730,26 +4730,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
@@ -4771,7 +4775,6 @@ 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)
                     # 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
@@ -4786,9 +4789,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)