From 0ad82404ac515cd6b9f66fd76d93727f6ad4610d Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Fri, 3 Apr 2026 18:20:03 +0300 Subject: [PATCH 1/9] (improvement) metadata: add lightweight _RowView and _row_factory Introduce _RowView, a __slots__-based read-only row wrapper that stores data as tuples with a shared column-name-to-index map, and _row_factory that creates these views. _RowView inherits from collections.abc.Mapping, providing a complete dict-like read interface. This eliminates per-row dict allocation during schema parsing. All rows from the same result set share a single index map object. --- cassandra/metadata.py | 61 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 0cb17e1337..7c9d1f7733 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -47,6 +47,67 @@ from cassandra.tablets import Tablets from cassandra.util import maybe_add_timeout_to_query + +class _RowView(Mapping): + """ + Lightweight read-only view over a row tuple, supporting dict-like access. + Shares a single index map across all rows from the same result set, + avoiding per-row dict allocation overhead. + + Implements the :class:`collections.abc.Mapping` protocol, providing + ``__getitem__``, ``__iter__``, ``__len__``, ``get``, ``keys``, + ``values``, ``items``, and ``__contains__`` for free. + """ + + __slots__ = ("_row", "_index_map") + + def __init__(self, row, index_map): + if len(row) < max(index_map.values(), default=-1) + 1: + raise ValueError( + "row length (%d) is insufficient for index_map with max index %d" + % (len(row), max(index_map.values(), default=-1)) + ) + self._row = row + self._index_map = index_map + + def __getitem__(self, key): + return self._row[self._index_map[key]] + + def __iter__(self): + return iter(self._index_map) + + def __len__(self): + return len(self._index_map) + + def values(self): + return (self._row[i] for i in self._index_map.values()) + + def items(self): + return ((k, self._row[i]) for k, i in self._index_map.items()) + + def get(self, key, default=None): + idx = self._index_map.get(key) + if idx is not None: + return self._row[idx] + return default + + def __contains__(self, key): + return key in self._index_map + + def __repr__(self): + return repr({k: self._row[i] for k, i in self._index_map.items()}) + + +def _row_factory(colnames, rows): + """ + Lightweight replacement for dict_factory used internally by schema parsers. + Returns a list of _RowView objects that support row["key"] and row.get("key") + but store data as tuples with a shared column-name-to-index map. + """ + index_map = {name: i for i, name in enumerate(colnames)} + return [_RowView(row, index_map) for row in rows] + + log = logging.getLogger(__name__) cql_keywords = set(( From f7e6201776edfcd894930753a266a19357dbcae7 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Fri, 3 Apr 2026 18:20:38 +0300 Subject: [PATCH 2/9] (improvement) metadata: replace OrderedDict with dict Python 3.7+ guarantees dict preserves insertion order, making OrderedDict unnecessary. Replace OrderedDict() with {} in TableMetadata.columns, TableMetadata.triggers, and MaterializedViewMetadata.columns. Remove the now-unused OrderedDict import. --- cassandra/metadata.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 7c9d1f7733..1a5f44ca36 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -41,7 +41,7 @@ from cassandra.marshal import varint_unpack from cassandra.protocol import QueryMessage from cassandra.query import dict_factory, bind_params -from cassandra.util import OrderedDict, Version +from cassandra.util import Version from cassandra.pool import HostDistance from cassandra.connection import EndPoint from cassandra.tablets import Tablets @@ -1454,11 +1454,11 @@ def __init__(self, keyspace_name, name, partition_key=None, clustering_key=None, self.name = name self.partition_key = [] if partition_key is None else partition_key self.clustering_key = [] if clustering_key is None else clustering_key - self.columns = OrderedDict() if columns is None else columns + self.columns = {} if columns is None else columns self.indexes = {} self.options = {} if options is None else options self.comparator = None - self.triggers = OrderedDict() if triggers is None else triggers + self.triggers = {} if triggers is None else triggers self.views = {} self.virtual = virtual @@ -2924,7 +2924,7 @@ def _build_table_columns(self, meta, col_rows, compact_static=False, is_dense=Fa partition_rows = sorted(partition_rows, key=lambda row: row.get('position')) for r in partition_rows: # we have to add meta here (and not in the later loop) because TableMetadata.columns is an - # OrderedDict, and it assumes keys are inserted first, in order, when exporting CQL + # dict (ordered since Python 3.7), and it assumes keys are inserted first, in order, when exporting CQL column_meta = self._build_column_metadata(meta, r) meta.columns[column_meta.name] = column_meta meta.partition_key.append(meta.columns[r.get('column_name')]) @@ -3533,7 +3533,7 @@ def __init__(self, keyspace_name, view_name, base_table_name, include_all_column self.base_table_name = base_table_name self.partition_key = [] self.clustering_key = [] - self.columns = OrderedDict() + self.columns = {} self.include_all_columns = include_all_columns self.where_clause = where_clause self.options = options or {} From 9934c1fec42e8de1e79693cd18852ce6e8bcbdd9 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Fri, 3 Apr 2026 18:21:14 +0300 Subject: [PATCH 3/9] (improvement) metadata: select only needed columns from system_schema.columns Replace SELECT * with an explicit column list for the system_schema.columns query in SchemaParserV3 (inherited by V4). Only the 7 columns actually consumed by the parser are fetched: keyspace_name, table_name, column_name, clustering_order, kind, position, type. This reduces network transfer and deserialization overhead during schema refresh. --- cassandra/metadata.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 1a5f44ca36..912a9adc48 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -2693,7 +2693,10 @@ class SchemaParserV3(SchemaParserV22): """ _SELECT_KEYSPACES = "SELECT * FROM system_schema.keyspaces" _SELECT_TABLES = "SELECT * FROM system_schema.tables" - _SELECT_COLUMNS = "SELECT * FROM system_schema.columns" + # Only fetch the columns used by _build_column_metadata / _build_table_columns. + # If _build_column_metadata or _build_table_columns needs more columns, this query + # should be updated accordingly. + _SELECT_COLUMNS = "SELECT keyspace_name, table_name, column_name, clustering_order, kind, position, type FROM system_schema.columns" _SELECT_INDEXES = "SELECT * FROM system_schema.indexes" _SELECT_TRIGGERS = "SELECT * FROM system_schema.triggers" _SELECT_TYPES = "SELECT * FROM system_schema.types" From 7e150ef6c93a8b5ebce55e2e23c7a93ee071aea8 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Fri, 3 Apr 2026 18:22:15 +0300 Subject: [PATCH 4/9] (improvement) metadata: replace dict_factory with lightweight _RowView Replace dict_factory in _SchemaParser._handle_results and get_column_from_system_local with _row_factory, eliminating per-row dict allocation during schema parsing. Also refactor SchemaParserV4._build_keyspace_metadata_internal to read from the row without mutating it, since _RowView is read-only. Note: V22-only dict_factory call sites are left unchanged as they do not affect the V3/V4 code path (V3 and V4 fully override _query_all). --- cassandra/metadata.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 912a9adc48..2563ccef02 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -2131,7 +2131,7 @@ def get_next_pages(): yield next_result.parsed_rows result.parsed_rows += itertools.chain(*get_next_pages()) - return dict_factory(result.column_names, result.parsed_rows) if result else [] + return _row_factory(result.column_names, result.parsed_rows) if result else [] else: raise result @@ -3264,11 +3264,12 @@ def get_all_keyspaces(self): @staticmethod def _build_keyspace_metadata_internal(row): - # necessary fields that aren't int virtual ks - row["durable_writes"] = row.get("durable_writes", None) - row["replication"] = row.get("replication", {}) - row["replication"]["class"] = row["replication"].get("class", None) - return super(SchemaParserV4, SchemaParserV4)._build_keyspace_metadata_internal(row) + # Read without mutating the row, since _RowView is read-only + name = row["keyspace_name"] + durable_writes = row.get("durable_writes", None) + replication = dict(row.get("replication")) if row.get("replication") else {} + replication_class = replication.pop("class") if 'class' in replication else None + return KeyspaceMetadata(name, durable_writes, replication_class, replication) class SchemaParserDSE67(SchemaParserV4): @@ -3334,7 +3335,7 @@ def get_table(self, keyspaces, keyspace, table): def _build_keyspace_metadata_internal(row): name = row["keyspace_name"] durable_writes = row.get("durable_writes", None) - replication = dict(row.get("replication")) if 'replication' in row else {} + replication = dict(row.get("replication")) if row.get("replication") else {} replication_class = replication.pop("class") if 'class' in replication else None graph_engine = row.get("graph_engine", None) return KeyspaceMetadata(name, durable_writes, replication_class, replication, graph_engine) @@ -3672,7 +3673,7 @@ def get_column_from_system_local(connection, column_name: str, timeout, metadata , timeout=timeout, fail_on_error=False) if not success or not local_result.parsed_rows: return "" - local_rows = dict_factory(local_result.column_names, local_result.parsed_rows) + local_rows = _row_factory(local_result.column_names, local_result.parsed_rows) local_row = local_rows[0] return local_row.get(column_name) From a6453c21c9df0fa7735d6a0e0527d33afb1e2675 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Fri, 3 Apr 2026 18:26:17 +0300 Subject: [PATCH 5/9] (improvement) metadata: single-pass _build_table_columns Replace three list comprehension passes over col_rows with a single classification loop that sorts columns into partition, clustering, and other buckets. Also use in-place sort() instead of sorted() and reuse the already-built column_meta instead of a redundant dict lookup. --- cassandra/metadata.py | 37 +++++++++++++++++++++++-------------- 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 2563ccef02..d40be4ece4 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -2920,31 +2920,40 @@ def _build_table_options(self, row): return dict((o, row.get(o)) for o in self.recognized_table_options if o in row) def _build_table_columns(self, meta, col_rows, compact_static=False, is_dense=False, virtual=False): - # partition key - partition_rows = [r for r in col_rows - if r.get('kind', None) == "partition_key"] + # Single-pass classification of column rows by kind + partition_rows = [] + clustering_rows = [] + other_rows = [] + for r in col_rows: + kind = r.get('kind', None) + if kind == "partition_key": + partition_rows.append(r) + elif kind == "clustering": + if not compact_static: + clustering_rows.append(r) + # else: skip clustering rows entirely for compact_static tables + else: + other_rows.append(r) + + # partition key - must be inserted first into meta.columns for CQL export ordering if len(partition_rows) > 1: - partition_rows = sorted(partition_rows, key=lambda row: row.get('position')) + partition_rows.sort(key=lambda row: row.get('position')) for r in partition_rows: - # we have to add meta here (and not in the later loop) because TableMetadata.columns is an - # dict (ordered since Python 3.7), and it assumes keys are inserted first, in order, when exporting CQL column_meta = self._build_column_metadata(meta, r) meta.columns[column_meta.name] = column_meta - meta.partition_key.append(meta.columns[r.get('column_name')]) + meta.partition_key.append(column_meta) # clustering key - if not compact_static: - clustering_rows = [r for r in col_rows - if r.get('kind', None) == "clustering"] + if clustering_rows: if len(clustering_rows) > 1: - clustering_rows = sorted(clustering_rows, key=lambda row: row.get('position')) + clustering_rows.sort(key=lambda row: row.get('position')) for r in clustering_rows: column_meta = self._build_column_metadata(meta, r) meta.columns[column_meta.name] = column_meta - meta.clustering_key.append(meta.columns[r.get('column_name')]) + meta.clustering_key.append(column_meta) - for col_row in (r for r in col_rows - if r.get('kind', None) not in ('partition_key', 'clustering')): + # remaining columns (static, regular, etc.) + for col_row in other_rows: column_meta = self._build_column_metadata(meta, col_row) if is_dense and column_meta.cql_type == types.cql_empty_type: continue From 1912883eeee5c14e83123399f92aaeb97f647ece Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Fri, 3 Apr 2026 18:27:18 +0300 Subject: [PATCH 6/9] tests: add _RowView and _row_factory unit tests Cover __getitem__, get(), __contains__, __repr__, shared index map, read-only enforcement, empty input, single-column, and multi-row scenarios. --- cassandra/metadata.py | 8 +--- tests/unit/test_metadata.py | 94 ++++++++++++++++++++++++++++++++++++- 2 files changed, 94 insertions(+), 8 deletions(-) diff --git a/cassandra/metadata.py b/cassandra/metadata.py index d40be4ece4..4c15491cc7 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -59,7 +59,7 @@ class _RowView(Mapping): ``values``, ``items``, and ``__contains__`` for free. """ - __slots__ = ("_row", "_index_map") + __slots__ = ("_index_map", "_row") def __init__(self, row, index_map): if len(row) < max(index_map.values(), default=-1) + 1: @@ -79,12 +79,6 @@ def __iter__(self): def __len__(self): return len(self._index_map) - def values(self): - return (self._row[i] for i in self._index_map.values()) - - def items(self): - return ((k, self._row[i]) for k, i in self._index_map.items()) - def get(self, key, default=None): idx = self._index_map.get(key) if idx is not None: diff --git a/tests/unit/test_metadata.py b/tests/unit/test_metadata.py index 2a1fced6cf..00ef508c6a 100644 --- a/tests/unit/test_metadata.py +++ b/tests/unit/test_metadata.py @@ -32,7 +32,8 @@ IndexMetadata, Function, Aggregate, Metadata, TokenMap, ReplicationFactor, SchemaParserDSE68, SchemaParserV3, - _ConsistencyMode, _consistency_mode_from_string) + _ConsistencyMode, _consistency_mode_from_string, + _RowView, _row_factory) from cassandra.policies import SimpleConvictionPolicy from cassandra.pool import Host from cassandra.protocol import QueryMessage @@ -1054,3 +1055,94 @@ def test_strip_frozen(self): for argument, expected_result in argument_to_expected_results: result = strip_frozen(argument) assert result == expected_result, "strip_frozen() arg: {}".format(argument) + +class RowViewTest(unittest.TestCase): + """Tests for the internal _RowView and _row_factory helpers.""" + + def test_getitem(self): + rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1}) + self.assertEqual(rv["a"], "a_val") + self.assertEqual(rv["b"], "b_val") + + def test_getitem_missing_key(self): + rv = _RowView(("a_val",), {"a": 0}) + with self.assertRaises(KeyError): + rv["missing"] + + def test_get_present(self): + rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1}) + self.assertEqual(rv.get("a"), "a_val") + self.assertEqual(rv.get("b"), "b_val") + + def test_get_missing_returns_default(self): + rv = _RowView(("a_val",), {"a": 0}) + self.assertIsNone(rv.get("missing")) + self.assertEqual(rv.get("missing", 42), 42) + + def test_contains(self): + rv = _RowView(("a_val",), {"a": 0}) + self.assertIn("a", rv) + self.assertNotIn("b", rv) + + def test_repr(self): + rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1}) + r = repr(rv) + self.assertIn("'a'", r) + self.assertIn("'a_val'", r) + + def test_shared_index_map(self): + """All _RowView objects from the same _row_factory call share one index map.""" + rows = _row_factory(["x", "y"], [("x1", "y1"), ("x2", "y2")]) + self.assertIs(rows[0]._index_map, rows[1]._index_map) + + def test_read_only(self): + """_RowView must not allow item assignment or deletion.""" + rv = _RowView(("val",), {"col": 0}) + with self.assertRaises(TypeError): + rv["col"] = "new" + with self.assertRaises(TypeError): + del rv["col"] + + def test_row_factory_empty(self): + result = _row_factory(["a", "b"], []) + self.assertEqual(result, []) + + def test_row_factory_single_column(self): + rows = _row_factory(["only"], [("v1",), ("v2",)]) + self.assertEqual(rows[0]["only"], "v1") + self.assertEqual(rows[1]["only"], "v2") + + def test_row_factory_values(self): + rows = _row_factory(["id", "name"], [(1, "alice"), (2, "bob")]) + self.assertEqual(rows[0]["id"], 1) + self.assertEqual(rows[0]["name"], "alice") + self.assertEqual(rows[1]["id"], 2) + self.assertEqual(rows[1]["name"], "bob") + + def test_len(self): + rv = _RowView(("a", "b", "c"), {"x": 0, "y": 1, "z": 2}) + self.assertEqual(len(rv), 3) + rv2 = _RowView((), {}) + self.assertEqual(len(rv2), 0) + + def test_keys(self): + rv = _RowView(("a", "b"), {"x": 0, "y": 1}) + self.assertEqual(set(rv.keys()), {"x", "y"}) + + def test_values(self): + rv = _RowView(("a", "b"), {"x": 0, "y": 1}) + self.assertEqual(list(rv.values()), ["a", "b"]) + + def test_items(self): + rv = _RowView(("a", "b"), {"x": 0, "y": 1}) + self.assertEqual(set(rv.items()), {("x", "a"), ("y", "b")}) + + def test_init_raises_on_short_row(self): + with self.assertRaises(ValueError): + _RowView(("val",), {"a": 0, "b": 1}) + + def test_init_accepts_exact_row(self): + _RowView(("a", "b"), {"a": 0, "b": 1}) + + def test_init_accepts_empty(self): + _RowView((), {}) From 2349244acccd9ac1d68725f818de1985ed513713 Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Mon, 21 Sep 2026 14:34:46 +0300 Subject: [PATCH 7/9] metadata: validate row width once per result set, not per row _RowView.__init__ computed max(index_map.values()) on every row, making construction O(columns) per row instead of O(1). That is the hot path when parsing system_schema results, so the max() defeated the point of the lightweight view. Compute the column count once in _row_factory and validate each row there, leaving _RowView.__init__ as two plain slot assignments. While here, stop materializing every page at once: _handle_results unpacked the entire get_next_pages() generator into itertools.chain, holding all pages and the accumulated rows in memory simultaneously. Extend per page instead. --- cassandra/metadata.py | 23 ++++++++++------ tests/unit/test_metadata.py | 53 ++----------------------------------- 2 files changed, 17 insertions(+), 59 deletions(-) diff --git a/cassandra/metadata.py b/cassandra/metadata.py index 4c15491cc7..c5d8582ebf 100644 --- a/cassandra/metadata.py +++ b/cassandra/metadata.py @@ -26,7 +26,6 @@ from threading import RLock import struct import random -import itertools from typing import Optional murmur3 = None @@ -62,11 +61,6 @@ class _RowView(Mapping): __slots__ = ("_index_map", "_row") def __init__(self, row, index_map): - if len(row) < max(index_map.values(), default=-1) + 1: - raise ValueError( - "row length (%d) is insufficient for index_map with max index %d" - % (len(row), max(index_map.values(), default=-1)) - ) self._row = row self._index_map = index_map @@ -97,9 +91,21 @@ def _row_factory(colnames, rows): Lightweight replacement for dict_factory used internally by schema parsers. Returns a list of _RowView objects that support row["key"] and row.get("key") but store data as tuples with a shared column-name-to-index map. + + Row length is validated once per result set here (not per _RowView), so the + hot path allocates only the two slots of each view. """ index_map = {name: i for i, name in enumerate(colnames)} - return [_RowView(row, index_map) for row in rows] + ncols = len(index_map) + views = [] + for row in rows: + if len(row) < ncols: + raise ValueError( + "row length (%d) is insufficient for result set with %d columns" + % (len(row), ncols) + ) + views.append(_RowView(row, index_map)) + return views log = logging.getLogger(__name__) @@ -2124,7 +2130,8 @@ def get_next_pages(): break yield next_result.parsed_rows - result.parsed_rows += itertools.chain(*get_next_pages()) + for page in get_next_pages(): + result.parsed_rows.extend(page) return _row_factory(result.column_names, result.parsed_rows) if result else [] else: raise result diff --git a/tests/unit/test_metadata.py b/tests/unit/test_metadata.py index 00ef508c6a..f60143007c 100644 --- a/tests/unit/test_metadata.py +++ b/tests/unit/test_metadata.py @@ -1064,16 +1064,6 @@ def test_getitem(self): self.assertEqual(rv["a"], "a_val") self.assertEqual(rv["b"], "b_val") - def test_getitem_missing_key(self): - rv = _RowView(("a_val",), {"a": 0}) - with self.assertRaises(KeyError): - rv["missing"] - - def test_get_present(self): - rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1}) - self.assertEqual(rv.get("a"), "a_val") - self.assertEqual(rv.get("b"), "b_val") - def test_get_missing_returns_default(self): rv = _RowView(("a_val",), {"a": 0}) self.assertIsNone(rv.get("missing")) @@ -1084,12 +1074,6 @@ def test_contains(self): self.assertIn("a", rv) self.assertNotIn("b", rv) - def test_repr(self): - rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1}) - r = repr(rv) - self.assertIn("'a'", r) - self.assertIn("'a_val'", r) - def test_shared_index_map(self): """All _RowView objects from the same _row_factory call share one index map.""" rows = _row_factory(["x", "y"], [("x1", "y1"), ("x2", "y2")]) @@ -1103,15 +1087,6 @@ def test_read_only(self): with self.assertRaises(TypeError): del rv["col"] - def test_row_factory_empty(self): - result = _row_factory(["a", "b"], []) - self.assertEqual(result, []) - - def test_row_factory_single_column(self): - rows = _row_factory(["only"], [("v1",), ("v2",)]) - self.assertEqual(rows[0]["only"], "v1") - self.assertEqual(rows[1]["only"], "v2") - def test_row_factory_values(self): rows = _row_factory(["id", "name"], [(1, "alice"), (2, "bob")]) self.assertEqual(rows[0]["id"], 1) @@ -1119,30 +1094,6 @@ def test_row_factory_values(self): self.assertEqual(rows[1]["id"], 2) self.assertEqual(rows[1]["name"], "bob") - def test_len(self): - rv = _RowView(("a", "b", "c"), {"x": 0, "y": 1, "z": 2}) - self.assertEqual(len(rv), 3) - rv2 = _RowView((), {}) - self.assertEqual(len(rv2), 0) - - def test_keys(self): - rv = _RowView(("a", "b"), {"x": 0, "y": 1}) - self.assertEqual(set(rv.keys()), {"x", "y"}) - - def test_values(self): - rv = _RowView(("a", "b"), {"x": 0, "y": 1}) - self.assertEqual(list(rv.values()), ["a", "b"]) - - def test_items(self): - rv = _RowView(("a", "b"), {"x": 0, "y": 1}) - self.assertEqual(set(rv.items()), {("x", "a"), ("y", "b")}) - - def test_init_raises_on_short_row(self): + def test_row_factory_raises_on_short_row(self): with self.assertRaises(ValueError): - _RowView(("val",), {"a": 0, "b": 1}) - - def test_init_accepts_exact_row(self): - _RowView(("a", "b"), {"a": 0, "b": 1}) - - def test_init_accepts_empty(self): - _RowView((), {}) + _row_factory(["a", "b"], [("val",)]) From ebb004e879f7ed9f2687a0df2432d6be2efd6963 Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Mon, 21 Sep 2026 14:54:22 +0300 Subject: [PATCH 8/9] cluster: back off schema agreement polling instead of a fixed 200ms Schema agreement is typically reached a few ms after a DDL returns, but the peers-view version signal is propagated by gossip (up to a gossip round on both Scylla and Cassandra). The fixed 200ms retry quantizes every wait up to the next interval, adding up to 200ms to a change that was already agreed. Poll at 10ms and double up to the historic 200ms cap on both the session and control-connection loops, so a fast change is noticed promptly while a slow one still converges under the same total timeout. --- cassandra/cluster.py | 24 +++++++++++++++++++-- tests/unit/test_cluster.py | 6 +++--- tests/unit/test_session_schema_agreement.py | 20 ++++++++++++++--- 3 files changed, 42 insertions(+), 8 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 7260bd08b6..043a630d9f 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -173,6 +173,24 @@ class SchemaAgreementScope(str, Enum): CLUSTER = 'cluster' +# Schema agreement is usually reached within a few milliseconds of a DDL +# returning (both system.local and the inter-node schema apply), but the version +# signal the peers view exposes is propagated by gossip, which can lag by up to a +# gossip round. A fixed retry interval therefore either over-waits on a slow +# change or adds up to a full interval of latency on a fast one. Start polling +# quickly and back off to the historic 200ms cap instead. +_SCHEMA_AGREEMENT_INITIAL_DELAY = 0.01 +_SCHEMA_AGREEMENT_MAX_DELAY = 0.2 + + +def _backoff_delays(): + """Yield successive schema agreement poll delays, doubling up to the cap.""" + delay = _SCHEMA_AGREEMENT_INITIAL_DELAY + while True: + yield delay + delay = min(delay * 2, _SCHEMA_AGREEMENT_MAX_DELAY) + + def _future_completed(future): """ Helper for run_in_executor() """ exc = future.exception() @@ -3514,6 +3532,7 @@ def wait_for_schema_agreement(self, wait_time: Optional[float] = None, deadline = time.time() + total_timeout schema_mismatches = None + delays = _backoff_delays() scope_label = 'local rack' if scope is SchemaAgreementScope.RACK else ( 'local datacenter' if scope is SchemaAgreementScope.DC else 'cluster') @@ -3525,7 +3544,7 @@ def wait_for_schema_agreement(self, wait_time: Optional[float] = None, log.debug("[session] Connected hosts in the %s still disagree on schema, trying again", scope_label) remaining = deadline - time.time() if remaining > 0: - time.sleep(min(0.2, remaining)) + time.sleep(min(next(delays), remaining)) log.warning("[session] Connected hosts in the %s are reporting a schema disagreement: %s", scope_label, schema_mismatches) @@ -4404,6 +4423,7 @@ def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wa elapsed = 0 cl = ConsistencyLevel.ONE schema_mismatches = None + delays = _backoff_delays() select_peers_query = self._get_peers_query(self.PeersQueryType.PEERS_SCHEMA, connection) while elapsed < total_timeout: @@ -4433,7 +4453,7 @@ def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wa return True log.debug("[control connection] Schemas mismatched, trying again") - self._time.sleep(0.2) + self._time.sleep(min(next(delays), max(0.0, total_timeout - elapsed))) elapsed = self._time.time() - start log.warning("Node %s is reporting a schema disagreement: %s", diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 35dc354465..07c26d3ed7 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -24,7 +24,7 @@ from cassandra import ConsistencyLevel, DriverException, Timeout, Unavailable, RequestExecutionException, ReadTimeout, WriteTimeout, CoordinationFailure, ReadFailure, WriteFailure, FunctionFailure, AlreadyExists,\ InvalidRequest, Unauthorized, AuthenticationFailed, OperationTimedOut, UnsupportedOperation, RequestValidationException, ConfigurationException, ProtocolVersion from cassandra.cluster import _Scheduler, Session, Cluster, ResultSet, SchemaAgreementScope, ControlConnectionQueryFallback, default_lbp_factory, \ - ExecutionProfile, _ConfigMode, EXEC_PROFILE_DEFAULT + ExecutionProfile, _ConfigMode, EXEC_PROFILE_DEFAULT, _SCHEMA_AGREEMENT_INITIAL_DELAY from cassandra.connection import ConnectionBusy, ConnectionException from cassandra.driver_config import DriverConfigReporter from cassandra.pool import Host @@ -627,7 +627,7 @@ def test_wait_for_schema_agreement_retries_until_local_hosts_match(self, *_): assert session.wait_for_schema_agreement(wait_time=1) for host in hosts: assert self._host_query_count(session, host) == 2 - assert clock.clock == 0.2 + assert clock.clock == _SCHEMA_AGREEMENT_INITIAL_DELAY @mock_session_pools def test_wait_for_schema_agreement_retries_when_local_connection_is_busy(self, *_): @@ -641,7 +641,7 @@ def test_wait_for_schema_agreement_retries_when_local_connection_is_busy(self, * assert session.wait_for_schema_agreement(wait_time=1) for host in hosts: assert self._host_query_count(session, host) == 2 - assert clock.clock == 0.2 + assert clock.clock == _SCHEMA_AGREEMENT_INITIAL_DELAY @mock_session_pools def test_wait_for_schema_agreement_ignores_local_hosts_without_session_pool(self, *_): diff --git a/tests/unit/test_session_schema_agreement.py b/tests/unit/test_session_schema_agreement.py index ffad687fcc..077b893770 100644 --- a/tests/unit/test_session_schema_agreement.py +++ b/tests/unit/test_session_schema_agreement.py @@ -7,7 +7,8 @@ import cassandra.cluster as cluster_module from cassandra.connection import ConnectionBusy -from cassandra.cluster import ControlConnection, Session, ResultSet +from cassandra.cluster import (ControlConnection, Session, ResultSet, + _SCHEMA_AGREEMENT_INITIAL_DELAY) from cassandra.policies import HostDistance, SimpleConvictionPolicy from cassandra.pool import Host from cassandra.util import maybe_add_timeout_to_query @@ -16,12 +17,14 @@ class FakeTime: def __init__(self): self.clock = 0 + self.sleeps = [] def time(self): return self.clock def sleep(self, amount): self.clock += amount + self.sleeps.append(amount) class MockPool: @@ -134,7 +137,7 @@ def test_wait_for_schema_agreement_retries_with_module_time(monkeypatch): connections[hosts[1]].future_outcomes = ["b", "a"] assert session.wait_for_schema_agreement(wait_time=1) - assert clock.clock == pytest.approx(0.2) + assert clock.clock == pytest.approx(_SCHEMA_AGREEMENT_INITIAL_DELAY) for host in hosts: assert _host_query_count(session, host) == 2 @@ -182,11 +185,22 @@ def test_wait_for_schema_agreement_retries_after_host_targeted_query_error(monke connections[hosts[1]].future_outcomes = [ConnectionBusy("connection overloaded"), "a"] assert session.wait_for_schema_agreement(wait_time=1) - assert clock.clock == pytest.approx(0.2) + assert clock.clock == pytest.approx(_SCHEMA_AGREEMENT_INITIAL_DELAY) for host in hosts: assert _host_query_count(session, host) == 2 +def test_wait_for_schema_agreement_backs_off_from_a_small_interval(monkeypatch): + session, hosts, connections = _new_session(["a", "b"]) + clock = FakeTime() + monkeypatch.setattr(cluster_module, "time", clock) + # host[1] disagrees for six rounds, then agrees; every round polls both hosts + connections[hosts[1]].future_outcomes = ["b"] * 6 + ["a"] + + assert session.wait_for_schema_agreement(wait_time=10) + assert clock.sleeps == pytest.approx([0.01, 0.02, 0.04, 0.08, 0.16, 0.2]) + + def test_wait_for_schema_agreement_queries_hosts_in_order_under_one_deadline(monkeypatch): session, hosts, _ = _new_session(["a", "a", "a"]) clock = FakeTime() From f851abb9bb559ab08779ccacf93cacfd85641325 Mon Sep 17 00:00:00 2001 From: Yaniv Kaul Date: Mon, 21 Sep 2026 15:06:35 +0300 Subject: [PATCH 9/9] cluster: use connected hosts' system.local for DDL schema agreement The DDL-triggered refresh waited on the control connection's system.peers view, but the schema version there is only propagated by gossip: measured ~0.4-0.7s behind the nodes actually applying the schema (Scylla 2026.3 and Cassandra 5.0), which dominated DDL latency at ~1s on a 2-node cluster. Thread the response's session into _refresh_schema and, when it has connected hosts, ask each of them for system.local directly (the existing session scope check) instead of polling gossip. Peers remain the fallback for callers without a session, such as control-connection startup. Also simplify the fallback _get_schema_mismatches to compare every reachable peer against a single reference version, only building the per-version endpoint breakdown when there is a mismatch to report. On a 2-node Cassandra 5.0 cluster this cuts a CREATE TABLE from ~1.0s to ~0.13s end to end. --- cassandra/cluster.py | 66 ++++++++++++++++++--- tests/integration/standard/test_cluster.py | 9 ++- tests/unit/test_cluster.py | 1 + tests/unit/test_control_connection.py | 63 ++++++++++++++++++++ tests/unit/test_session_schema_agreement.py | 2 + 5 files changed, 130 insertions(+), 11 deletions(-) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 043a630d9f..01267733e4 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -3622,7 +3622,7 @@ def _get_schema_agreement_hosts(self, scope: SchemaAgreementScope) -> Tuple[Host def _query_local_schema_version(self, host: Host, query: str, deadline: float) -> Future: remaining = max(0.0, deadline - time.time()) try: - response_future = self.execute_async( + response_future = self._send_schema_version_query( query, timeout=self._schema_agreement_query_timeout(remaining), host=host, @@ -3634,7 +3634,7 @@ def _query_local_schema_version(self, host: Host, query: str, deadline: float) - log.debug("[session] Error querying schema version from %s: %s", host, exc) raise - # execute_async returns cassandra.cluster.ResponseFuture, which does not have bulk waiting logic for it. + # _send_schema_version_query returns cassandra.cluster.ResponseFuture, which does not have bulk waiting logic for it. # That is why _query_local_schema_version returns concurrent.futures.Future # so that schema agreement logic could use concurrent.futures.wait_futures to wait on them. # schema_version_future is an adapter between cassandra.cluster.ResponseFuture and concurrent.futures.Future @@ -3662,6 +3662,23 @@ def _set_exception(exc, result_future=schema_version_future): return schema_version_future + def _send_schema_version_query(self, query: str, timeout: float, host: Host) -> "ResponseFuture": + """Send an internal schema version query with a pinned row factory. + + The session's configured row factory is deliberately not used: cqlengine, + for example, installs ``dict_factory``, which does not expose columns as + attributes and would make the version column unreadable. + """ + response_future = self._create_response_future( + query, parameters=None, trace=False, custom_payload=None, + timeout=timeout, host=host, + ) + response_future.row_factory = named_tuple_factory + response_future._protocol_handler = self.client_protocol_handler + self._on_request(response_future) + response_future.send_request() + return response_future + def _schema_agreement_query_timeout(self, remaining: float) -> float: control_timeout = self.cluster.control_connection._timeout if control_timeout is None: @@ -4065,6 +4082,8 @@ def shutdown(self): def refresh_schema(self, force=False, **kwargs): try: if self._connection: + if kwargs.get('session') is None: + kwargs['session'] = next(iter(self._cluster.sessions), None) return self._refresh_schema(self._connection, force=force, **kwargs) except ReferenceError: pass # our weak reference to the Cluster is no good @@ -4073,13 +4092,15 @@ def refresh_schema(self, force=False, **kwargs): self._signal_error() return False - def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_wait=None, force=False, **kwargs): + def _refresh_schema(self, connection, preloaded_results=None, schema_agreement_wait=None, force=False, + session=None, **kwargs): if self._cluster.is_shutdown: return False agreed = self._wait_for_schema_agreement(connection=connection, preloaded_results=preloaded_results, - wait_time=schema_agreement_wait) + wait_time=schema_agreement_wait, + session=session) if not self._schema_meta_enabled and not force: log.debug("[control connection] Skipping schema refresh because schema metadata is disabled") @@ -4393,7 +4414,8 @@ def wait_for_schema_agreement(self, connection=None, preloaded_results=None, wai preloaded_results=preloaded_results, wait_time=wait_time) - def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wait_time=None): + def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wait_time=None, + session=None): total_timeout = wait_time if wait_time is not None else self._cluster.max_schema_agreement_wait if total_timeout <= 0: return True @@ -4418,6 +4440,14 @@ def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wa if schema_mismatches is None: return True + if session is not None and self._connected_host_agreement_available(session): + # system.local reflects an applied schema change immediately, while + # system.peers only learns the version on the next gossip round. When + # a session is available, ask the connected hosts directly so a DDL + # does not block on gossip propagation. + log.debug("[control connection] Waiting for schema agreement via connected hosts") + return session.wait_for_schema_agreement(wait_time=total_timeout) + log.debug("[control connection] Waiting for schema agreement") start = self._time.time() elapsed = 0 @@ -4460,14 +4490,31 @@ def _wait_for_schema_agreement(self, connection=None, preloaded_results=None, wa connection.endpoint, schema_mismatches) return False + @staticmethod + def _connected_host_agreement_available(session): + """Whether the session has at least one connected host to check directly. + + Returns False (so the peers-based loop is used) when the session has no + connected pools, e.g. during control-connection startup. + """ + try: + return bool(session._get_schema_agreement_hosts(SchemaAgreementScope.CLUSTER)) + except Exception: + log.debug("[control connection] Cannot use connected hosts for schema agreement, " + "falling back to peers", exc_info=True) + return False + def _get_schema_mismatches(self, peers_result, local_result, local_address): peers_result = dict_factory(peers_result.column_names, peers_result.parsed_rows) - versions = defaultdict(set) + local_row = None if local_result.parsed_rows: local_row = dict_factory(local_result.column_names, local_result.parsed_rows)[0] - if local_row.get("schema_version"): - versions[local_row.get("schema_version")].add(local_address) + local_version = local_row.get("schema_version") if local_row else None + + versions = defaultdict(set) + if local_version: + versions[local_version].add(local_address) for row in peers_result: schema_ver = row.get('schema_version') @@ -4677,7 +4724,8 @@ def refresh_schema_and_set_result(control_conn, response_future, connection, **k try: log.debug("Refreshing schema in response to schema change. " "%s", kwargs) - response_future.is_schema_agreed = control_conn._refresh_schema(connection, **kwargs) + response_future.is_schema_agreed = control_conn._refresh_schema( + connection, session=response_future.session, **kwargs) except Exception: log.exception("Exception refreshing schema in response to schema change:") response_future.session.submit(control_conn.refresh_schema, **kwargs) diff --git a/tests/integration/standard/test_cluster.py b/tests/integration/standard/test_cluster.py index e4e3a8f5e1..03251c4436 100644 --- a/tests/integration/standard/test_cluster.py +++ b/tests/integration/standard/test_cluster.py @@ -27,7 +27,7 @@ import pytest import cassandra -from cassandra.cluster import NoHostAvailable, ExecutionProfile, EXEC_PROFILE_DEFAULT, Cluster +from cassandra.cluster import NoHostAvailable, ExecutionProfile, EXEC_PROFILE_DEFAULT, Cluster, Session from cassandra.concurrent import execute_concurrent from cassandra.policies import (RoundRobinPolicy, ExponentialReconnectionPolicy, SimpleConvictionPolicy, HostDistance, @@ -513,7 +513,12 @@ def patched_wait_for_responses(*args, **kwargs): response[1].parsed_rows[0] = (new_uuid,) return response - with patch.object(connection.Connection, "wait_for_responses", patched_wait_for_responses): + def patched_schema_mismatches(self, deadline, scope): + # The connected hosts never agree on a schema version + return {"disagreement": ["fake"]} + + with patch.object(connection.Connection, "wait_for_responses", patched_wait_for_responses), \ + patch.object(Session, "_get_schema_mismatches_for_scope", patched_schema_mismatches): agreement_timeout = 1 # cluster agreement wait exceeded diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 07c26d3ed7..3aecefcd1a 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -513,6 +513,7 @@ def execute_async(query, parameters=None, trace=False, return self._mock_schema_future(outcome) session.execute_async = Mock(side_effect=execute_async) + session._send_schema_version_query = session.execute_async return session, hosts, connections diff --git a/tests/unit/test_control_connection.py b/tests/unit/test_control_connection.py index fd62323f33..d8acc8cf9b 100644 --- a/tests/unit/test_control_connection.py +++ b/tests/unit/test_control_connection.py @@ -111,6 +111,7 @@ def __init__(self): self.profile_manager.profiles[EXEC_PROFILE_DEFAULT] = ExecutionProfile(RoundRobinPolicy()) self.endpoint_factory = DefaultEndPointFactory().configure(self) self.ssl_options = None + self.sessions = set() def add_host(self, endpoint, datacenter, rack, signal=False, refresh_nodes=True, host_id=None): host = Host(endpoint, SimpleConvictionPolicy, datacenter, rack, host_id=host_id) @@ -214,6 +215,35 @@ def test_wait_for_schema_agreement(self): # the control connection should not have slept at all assert self.time.clock == 0 + def test_wait_for_schema_agreement_prefers_connected_hosts_when_session_available(self): + """ + With a session that has connected hosts, ask system.local directly instead + of reading the gossip-based system.peers view. + """ + session = Mock() + session._get_schema_agreement_hosts = Mock(return_value=("host",)) + session.wait_for_schema_agreement = Mock(return_value=True) + + assert self.control_connection._wait_for_schema_agreement(session=session) + + session.wait_for_schema_agreement.assert_called_once_with( + wait_time=self.cluster.max_schema_agreement_wait) + assert self.connection.wait_for_responses.call_count == 0 + assert self.time.clock == 0 + + def test_wait_for_schema_agreement_falls_back_to_peers_without_connected_hosts(self): + """ + A session with no connected hosts falls back to the peers-based loop. + """ + session = Mock() + session._get_schema_agreement_hosts = Mock(return_value=()) + + assert self.control_connection._wait_for_schema_agreement(session=session) + + session.wait_for_schema_agreement.assert_not_called() + assert self.connection.wait_for_responses.call_count == 1 + assert self.time.clock == 0 + @patch('cassandra.cluster.warn') def test_wait_for_schema_agreement_warns_about_deprecation(self, mocked_warn): assert self.control_connection.wait_for_schema_agreement() @@ -465,6 +495,20 @@ def bad_wait_for_responses(*args, **kwargs): assert self.connection.wait_for_responses.call_args[1]['timeout'] == self.control_connection._timeout mocked_warn.assert_not_called() + def test_refresh_schema_uses_connected_session(self): + """ + A refresh without an explicit session should forward a connected + session from the cluster so agreement can use system.local directly. + """ + session = Mock() + self.cluster.sessions.add(session) + self.control_connection._refresh_schema = Mock(return_value=True) + + assert self.control_connection.refresh_schema() + + self.control_connection._refresh_schema.assert_called_once_with( + self.connection, force=False, session=session) + def test_handle_topology_change(self): event = { 'change_type': 'NEW_NODE', @@ -673,3 +717,22 @@ def test_event_delay_timing(self): current_delay = self.cluster.scheduler.mock_calls[0][1][0] assert prior_delay < current_delay prior_delay = current_delay + + +def test_refresh_schema_and_set_result_passes_response_session(): + """ + The DDL-triggered refresh must hand the response's session to the control + connection so schema agreement can use connected hosts instead of gossip. + """ + from cassandra.cluster import refresh_schema_and_set_result + + control_conn = Mock() + response_future = Mock() + connection = Mock() + + refresh_schema_and_set_result(control_conn, response_future, connection, + target_type='KEYSPACE', keyspace='ks') + + control_conn._refresh_schema.assert_called_once_with( + connection, session=response_future.session, target_type='KEYSPACE', keyspace='ks') + response_future._set_final_result.assert_called_once_with(None) diff --git a/tests/unit/test_session_schema_agreement.py b/tests/unit/test_session_schema_agreement.py index 077b893770..fdd39ee4a6 100644 --- a/tests/unit/test_session_schema_agreement.py +++ b/tests/unit/test_session_schema_agreement.py @@ -126,6 +126,7 @@ def execute_async(query, parameters=None, trace=False, return MockSchemaVersionFuture(outcome) session.execute_async = Mock(side_effect=execute_async) + session._send_schema_version_query = session.execute_async return session, hosts, connections @@ -213,6 +214,7 @@ def execute_async(query, parameters=None, trace=False, return MockSchemaVersionFuture("a") session.execute_async = Mock(side_effect=execute_async) + session._send_schema_version_query = session.execute_async assert session.wait_for_schema_agreement(wait_time=1) assert [call.kwargs["host"] for call in session.execute_async.call_args_list] == hosts