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.

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)

Author: mykaul

benchmarks/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/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()

cassandra/cluster.py

@@ -4773,12 +4773,21 @@ 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
+                    # Use pre-cached column names/types from PreparedStatement
+                    # when available to avoid rebuilding lists from metadata.
+                    ps = self.prepared_statement
+                    if ps is not None and ps._result_col_names is not None:
+                        col_names = ps._result_col_names
+                        col_types = ps._result_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:
@@ -4944,6 +4953,7 @@ def _execute_after_prepare(self, host, connection, pool, response):
                             )
                         ))
                     self.prepared_statement.result_metadata = response.column_metadata
+                    self.prepared_statement._cache_result_metadata_columns(response.column_metadata)
                     new_metadata_id = response.result_metadata_id
                     if new_metadata_id is not None:
                         self.prepared_statement.result_metadata_id = new_metadata_id

cassandra/query.py

@@ -459,6 +459,11 @@ class PreparedStatement(object):
     serial_consistency_level = None  # TODO never used?
     _is_lwt = False
 
+    # Cached column names/types derived from result_metadata, to avoid
+    # rebuilding [c[2] for c in result_metadata] on every result set.
+    _result_col_names = None
+    _result_col_types = None
+
     def __init__(self, column_metadata, query_id, routing_key_indexes, query,
                  keyspace, protocol_version, result_metadata, result_metadata_id,
                  is_lwt=False, column_encryption_policy=None):
@@ -469,11 +474,21 @@ def __init__(self, column_metadata, query_id, routing_key_indexes, query,
         self.keyspace = keyspace
         self.protocol_version = protocol_version
         self.result_metadata = result_metadata
+        self._cache_result_metadata_columns(result_metadata)
         self.result_metadata_id = result_metadata_id
         self.column_encryption_policy = column_encryption_policy
         self.is_idempotent = False
         self._is_lwt = is_lwt
 
+    def _cache_result_metadata_columns(self, result_metadata):
+        """Pre-compute column names and types from result_metadata."""
+        if result_metadata:
+            self._result_col_names = [c[2] for c in result_metadata]
+            self._result_col_types = [c[3] for c in result_metadata]
+        else:
+            self._result_col_names = None
+            self._result_col_types = None
+
     @classmethod
     def from_message(cls, query_id, column_metadata, pk_indexes, cluster_metadata,
                      query, prepared_keyspace, protocol_version, result_metadata,