From ef0bb90731e46043cbecd5baa42366208fb409d8 Mon Sep 17 00:00:00 2001 From: Yaniv Michael Kaul Date: Fri, 3 Apr 2026 21:15:20 +0300 Subject: [PATCH] (improvement) cache namedtuple class in named_tuple_factory to avoid repeated exec() calls Cache the Row namedtuple class keyed on tuple(colnames) so Python's namedtuple() (which internally calls exec()) is only invoked once per unique column schema. For prepared statements the column names never change, eliminating redundant class creation on every result set. Cache is a plain dict keyed on tuple(colnames) (raw column names before cleaning, exact names, exact order). Since cleaning/sanitizing colnames is a pure function of that key, two schemas only ever share a cached class when their column names match exactly -- differing case, order, or contents all produce distinct keys. Error handling paths (SyntaxError, Exception) preserved unchanged. The cache is naturally bounded by the number of distinct queries for typical usage (a fixed set of prepared statements), but applications that build many ad hoc queries against highly variable or generated schemas could otherwise grow it without bound. It is now capped at 10000 entries with oldest-first (FIFO) eviction once full, relying on dict insertion order, to keep worst-case memory bounded. Address review feedback: - Thread safety: the miss/evict/insert sequence on a cache miss was not synchronized, so one thread's next(iter(_named_tuple_cache)) (picking an eviction victim) could race with another thread mutating the same dict, raising `RuntimeError: dictionary changed size during iteration`; two threads observing the cache under its bound before either inserted could also together push it past that bound. This driver advertises free-threaded Python support, so the whole check-evict-insert sequence is now guarded by a lock (_named_tuple_cache_lock), with the cache-HIT path left lock-free. Verified with a real pre-fix/post-fix repro under CPython's free-threaded (3.14t) build: the pre-fix code reliably raised RuntimeError and exceeded its stated bound under concurrent misses, the post-fix code did neither across repeated runs. Added a regression test (tests/unit/test_row_factories.py, TestNamedTupleFactoryCacheThreadSafety) that hammers the cache from many threads with distinct column-name sets to force evictions, asserting no exception is raised and the bound is respected. - Test placement: the cache correctness tests (cache-hit, cache-key, eviction) were previously only under benchmarks/, which is not run by the project's wheel test commands (only tests/unit is). Moved them into tests/unit/test_row_factories.py so regressions in the production cache are actually caught in CI; benchmarks/ now contains only the timing benchmarks. - Dev dependencies: added pytest-benchmark to the `dev` dependency group in pyproject.toml so `pytest benchmarks/` is runnable in the documented development environment. --- .../test_named_tuple_factory_benchmark.py | 171 +++++++++++++++ cassandra/query.py | 102 ++++++--- pyproject.toml | 1 + tests/unit/test_row_factories.py | 201 +++++++++++++++++- 4 files changed, 449 insertions(+), 26 deletions(-) create mode 100644 benchmarks/test_named_tuple_factory_benchmark.py diff --git a/benchmarks/test_named_tuple_factory_benchmark.py b/benchmarks/test_named_tuple_factory_benchmark.py new file mode 100644 index 0000000000..e0122c995a --- /dev/null +++ b/benchmarks/test_named_tuple_factory_benchmark.py @@ -0,0 +1,171 @@ +# 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. + +""" +Benchmarks for named_tuple_factory with and without namedtuple class caching. + +Run with: pytest benchmarks/test_named_tuple_factory_benchmark.py -v + +Correctness tests for the same cache (cache-hit, cache-key, and eviction +behavior) live in tests/unit/test_row_factories.py instead of here, because +the project's wheel test commands only run `pytest tests/unit` -- code left +only under benchmarks/ is never exercised in CI. Only pure timing/benchmark +code belongs in this file. +""" + +import re +from collections import namedtuple + +import pytest + +from cassandra.query import named_tuple_factory, _named_tuple_cache +from cassandra.util import _sanitize_identifiers + + +# --------------------------------------------------------------------------- +# Reference: original uncached implementation (copied from master) +# --------------------------------------------------------------------------- + +NON_ALPHA_REGEX = re.compile("[^a-zA-Z0-9]") +START_BADCHAR_REGEX = re.compile("^[^a-zA-Z0-9]*") +END_BADCHAR_REGEX = re.compile("[^a-zA-Z0-9_]*$") + +_clean_name_cache_old = {} + + +def _clean_column_name_old(name): + try: + return _clean_name_cache_old[name] + except KeyError: + clean = NON_ALPHA_REGEX.sub( + "_", START_BADCHAR_REGEX.sub("", END_BADCHAR_REGEX.sub("", name)) + ) + _clean_name_cache_old[name] = clean + return clean + + +def named_tuple_factory_uncached(colnames, rows): + """Original implementation without caching (for benchmark comparison).""" + clean_column_names = map(_clean_column_name_old, colnames) + try: + Row = namedtuple("Row", clean_column_names) + except SyntaxError: + raise + except Exception: + clean_column_names = list(map(_clean_column_name_old, colnames)) + Row = namedtuple("Row", _sanitize_identifiers(clean_column_names)) + return [Row(*row) for row in rows] + + +# --------------------------------------------------------------------------- +# Test data generators +# --------------------------------------------------------------------------- + + +def make_colnames(n): + return tuple(f"col_{i}" for i in range(n)) + + +def make_rows(ncols, nrows): + return [tuple(range(ncols)) for _ in range(nrows)] + + +# --------------------------------------------------------------------------- +# Benchmarks +# --------------------------------------------------------------------------- + + +class TestNamedTupleFactoryBenchmark: + """Benchmark cached vs uncached named_tuple_factory.""" + + # --- 5 columns, 100 rows --- + + @pytest.mark.benchmark(group="ntf_5cols_100rows") + def test_uncached_5cols_100rows(self, benchmark): + colnames = make_colnames(5) + rows = make_rows(5, 100) + benchmark(named_tuple_factory_uncached, colnames, rows) + + @pytest.mark.benchmark(group="ntf_5cols_100rows") + def test_cached_5cols_100rows(self, benchmark): + colnames = make_colnames(5) + rows = make_rows(5, 100) + _named_tuple_cache.clear() + # Warm the cache with one call + named_tuple_factory(colnames, rows) + benchmark(named_tuple_factory, colnames, rows) + + # --- 10 columns, 100 rows --- + + @pytest.mark.benchmark(group="ntf_10cols_100rows") + def test_uncached_10cols_100rows(self, benchmark): + colnames = make_colnames(10) + rows = make_rows(10, 100) + benchmark(named_tuple_factory_uncached, colnames, rows) + + @pytest.mark.benchmark(group="ntf_10cols_100rows") + def test_cached_10cols_100rows(self, benchmark): + colnames = make_colnames(10) + rows = make_rows(10, 100) + _named_tuple_cache.clear() + named_tuple_factory(colnames, rows) + benchmark(named_tuple_factory, colnames, rows) + + # --- 20 columns, 100 rows --- + + @pytest.mark.benchmark(group="ntf_20cols_100rows") + def test_uncached_20cols_100rows(self, benchmark): + colnames = make_colnames(20) + rows = make_rows(20, 100) + benchmark(named_tuple_factory_uncached, colnames, rows) + + @pytest.mark.benchmark(group="ntf_20cols_100rows") + def test_cached_20cols_100rows(self, benchmark): + colnames = make_colnames(20) + rows = make_rows(20, 100) + _named_tuple_cache.clear() + named_tuple_factory(colnames, rows) + benchmark(named_tuple_factory, colnames, rows) + + # --- 5 columns, 1000 rows --- + + @pytest.mark.benchmark(group="ntf_5cols_1000rows") + def test_uncached_5cols_1000rows(self, benchmark): + colnames = make_colnames(5) + rows = make_rows(5, 1000) + benchmark(named_tuple_factory_uncached, colnames, rows) + + @pytest.mark.benchmark(group="ntf_5cols_1000rows") + def test_cached_5cols_1000rows(self, benchmark): + colnames = make_colnames(5) + rows = make_rows(5, 1000) + _named_tuple_cache.clear() + named_tuple_factory(colnames, rows) + benchmark(named_tuple_factory, colnames, rows) + + # --- 10 columns, 1 row (measures class creation overhead most clearly) --- + + @pytest.mark.benchmark(group="ntf_10cols_1row") + def test_uncached_10cols_1row(self, benchmark): + colnames = make_colnames(10) + rows = make_rows(10, 1) + benchmark(named_tuple_factory_uncached, colnames, rows) + + @pytest.mark.benchmark(group="ntf_10cols_1row") + def test_cached_10cols_1row(self, benchmark): + colnames = make_colnames(10) + rows = make_rows(10, 1) + _named_tuple_cache.clear() + named_tuple_factory(colnames, rows) + benchmark(named_tuple_factory, colnames, rows) diff --git a/cassandra/query.py b/cassandra/query.py index 39b9fdb0ad..6b414be79f 100644 --- a/cassandra/query.py +++ b/cassandra/query.py @@ -22,6 +22,7 @@ from datetime import datetime, timedelta, timezone import re import struct +import threading import time import warnings @@ -117,6 +118,34 @@ def pseudo_namedtuple_factory(colnames, rows): for od in ordered_dict_factory(colnames, rows)] +# Cache namedtuple Row classes to avoid repeated exec() calls in namedtuple() +# for the same column schema. Keyed on the exact, ordered tuple of raw column +# names, so two schemas only share a cached class if their column names match +# exactly (same names, same case, same order); cleaning/sanitizing is a pure +# function of that key, so the derived Row class is always correct for it. +# +# For typical usage (a bounded set of prepared statements) this cache is +# naturally bounded by the number of distinct queries. Applications that +# build many ad hoc queries against highly variable/generated schemas could +# otherwise grow this without bound, so it is capped and evicted FIFO +# (oldest entry first, relying on dict insertion order) once full. +_named_tuple_cache = {} +_NAMED_TUPLE_CACHE_MAX_SIZE = 10000 + +# Guards the check-evict-insert sequence on the cache-miss path in +# named_tuple_factory() below. This driver supports free-threaded Python, so +# without synchronization, one thread iterating _named_tuple_cache to pick an +# eviction victim (`next(iter(...))`) can race with another thread mutating +# the same dict, raising `RuntimeError: dictionary changed size during +# iteration`; separately, two threads that both observe the cache under its +# size bound before either inserts can together push it past that bound. The +# cache-HIT path (the plain `_named_tuple_cache[key]` lookup above) does NOT +# take this lock -- concurrent reads of a dict are safe, and this cache is on +# a hot path where the whole point is to avoid paying synchronization cost on +# every call. +_named_tuple_cache_lock = threading.Lock() + + def named_tuple_factory(colnames, rows): """ Returns each row as a `namedtuple `_. @@ -146,32 +175,55 @@ def named_tuple_factory(colnames, rows): .. versionchanged:: 2.0.0 moved from ``cassandra.decoder`` to ``cassandra.query`` """ - clean_column_names = map(_clean_column_name, colnames) + key = tuple(colnames) try: - Row = namedtuple('Row', clean_column_names) - except SyntaxError: - warnings.warn( - "Failed creating namedtuple for a result because there were too " - "many columns. This is due to a Python limitation that affects " - "namedtuple in Python 3.0-3.6 (see issue18896). The row will be " - "created with {substitute_factory_name}, which lacks some namedtuple " - "features and is slower. To avoid slower performance accessing " - "values on row objects, Upgrade to Python 3.7, or use a different " - "row factory. (column names: {colnames})".format( - substitute_factory_name=pseudo_namedtuple_factory.__name__, - colnames=colnames - ) - ) - return pseudo_namedtuple_factory(colnames, rows) - except Exception: - clean_column_names = list(map(_clean_column_name, colnames)) # create list because py3 map object will be consumed by first attempt - log.warning("Failed creating named tuple for results with column names %s (cleaned: %s) " - "(see Python 'namedtuple' documentation for details on name rules). " - "Results will be returned with positional names. " - "Avoid this by choosing different names, using SELECT \"\" AS aliases, " - "or specifying a different row_factory on your Session" % - (colnames, clean_column_names)) - Row = namedtuple('Row', _sanitize_identifiers(clean_column_names)) + Row = _named_tuple_cache[key] + except KeyError: + # Miss path: synchronize the whole check-evict-insert sequence. + # Re-check the cache once we hold the lock in case another thread + # already populated `key` while we were waiting for it (the + # double-checked-locking pattern), so we don't do redundant work or + # clobber a class other callers may already hold a reference to. + with _named_tuple_cache_lock: + try: + Row = _named_tuple_cache[key] + except KeyError: + clean_column_names = map(_clean_column_name, colnames) + try: + Row = namedtuple('Row', clean_column_names) + except SyntaxError: + warnings.warn( + "Failed creating namedtuple for a result because there were too " + "many columns. This is due to a Python limitation that affects " + "namedtuple in Python 3.0-3.6 (see issue18896). The row will be " + "created with {substitute_factory_name}, which lacks some namedtuple " + "features and is slower. To avoid slower performance accessing " + "values on row objects, Upgrade to Python 3.7, or use a different " + "row factory. (column names: {colnames})".format( + substitute_factory_name=pseudo_namedtuple_factory.__name__, + colnames=colnames + ) + ) + return pseudo_namedtuple_factory(colnames, rows) + except Exception: + clean_column_names = list(map(_clean_column_name, colnames)) # create list because py3 map object will be consumed by first attempt + log.warning("Failed creating named tuple for results with column names %s (cleaned: %s) " + "(see Python 'namedtuple' documentation for details on name rules). " + "Results will be returned with positional names. " + "Avoid this by choosing different names, using SELECT \"\" AS aliases, " + "or specifying a different row_factory on your Session" % + (colnames, clean_column_names)) + Row = namedtuple('Row', _sanitize_identifiers(clean_column_names)) + if len(_named_tuple_cache) >= _NAMED_TUPLE_CACHE_MAX_SIZE: + # Evict the oldest entry (dicts preserve insertion order) to + # keep memory bounded when many distinct column-name schemas + # are seen. Safe under the lock: no other thread can be + # iterating or mutating _named_tuple_cache concurrently. + try: + _named_tuple_cache.pop(next(iter(_named_tuple_cache))) + except (StopIteration, KeyError): + pass + _named_tuple_cache[key] = Row return [Row(*row) for row in rows] diff --git a/pyproject.toml b/pyproject.toml index 698ff4c37b..41bb677d20 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -43,6 +43,7 @@ auth-kerberos = [ [dependency-groups] dev = [ "pytest~=8.0", + "pytest-benchmark", "PyYAML", "pure-sasl", "twisted[tls]", diff --git a/tests/unit/test_row_factories.py b/tests/unit/test_row_factories.py index 7787f1d271..652898cc36 100644 --- a/tests/unit/test_row_factories.py +++ b/tests/unit/test_row_factories.py @@ -13,15 +13,23 @@ # limitations under the License. -from cassandra.query import named_tuple_factory +import cassandra.query as query_module +from cassandra.query import named_tuple_factory, _named_tuple_cache import logging +import re +import threading import warnings +from collections import namedtuple import sys from unittest import TestCase +import pytest + +from cassandra.util import _sanitize_identifiers + log = logging.getLogger(__name__) @@ -85,3 +93,194 @@ def test_creation_no_warning_on_short_column_list(self): # check that this is a real namedtuple assert hasattr(rows[0], '_fields') assert isinstance(rows[0], tuple) + + +# --------------------------------------------------------------------------- +# Correctness tests for the namedtuple-class cache in named_tuple_factory. +# +# These were originally added under benchmarks/test_named_tuple_factory_benchmark.py +# alongside the timing benchmarks for the same cache, but the project's +# wheel test commands only run `pytest tests/unit`, so none of this +# cache-hit/cache-key/eviction correctness coverage was exercised in CI. +# Moved here so regressions in the production cache are actually caught; +# only genuine timing/benchmark code remains under benchmarks/. +# --------------------------------------------------------------------------- + +NON_ALPHA_REGEX = re.compile("[^a-zA-Z0-9]") +START_BADCHAR_REGEX = re.compile("^[^a-zA-Z0-9]*") +END_BADCHAR_REGEX = re.compile("[^a-zA-Z0-9_]*$") + +_clean_name_cache_old = {} + + +def _clean_column_name_old(name): + try: + return _clean_name_cache_old[name] + except KeyError: + clean = NON_ALPHA_REGEX.sub( + "_", START_BADCHAR_REGEX.sub("", END_BADCHAR_REGEX.sub("", name)) + ) + _clean_name_cache_old[name] = clean + return clean + + +def named_tuple_factory_uncached(colnames, rows): + """Reference implementation without caching, used to verify the cached + implementation in cassandra.query.named_tuple_factory returns + equivalent results.""" + clean_column_names = map(_clean_column_name_old, colnames) + try: + Row = namedtuple("Row", clean_column_names) + except SyntaxError: + raise + except Exception: + clean_column_names = list(map(_clean_column_name_old, colnames)) + Row = namedtuple("Row", _sanitize_identifiers(clean_column_names)) + return [Row(*row) for row in rows] + + +def make_colnames(n): + return tuple(f"col_{i}" for i in range(n)) + + +def make_rows(ncols, nrows): + return [tuple(range(ncols)) for _ in range(nrows)] + + +class TestNamedTupleFactoryCache: + """Verify the cached implementation matches the uncached one, and that + the cache's keying and eviction behavior are correct.""" + + @pytest.mark.parametrize("ncols", [1, 5, 10, 20]) + @pytest.mark.parametrize("nrows", [1, 10, 100]) + def test_results_match(self, ncols, nrows): + colnames = make_colnames(ncols) + rows = make_rows(ncols, nrows) + _named_tuple_cache.clear() + cached_result = named_tuple_factory(colnames, rows) + uncached_result = named_tuple_factory_uncached(colnames, rows) + assert len(cached_result) == len(uncached_result) + for cr, ur in zip(cached_result, uncached_result): + assert tuple(cr) == tuple(ur) + assert cr._fields == ur._fields + + def test_cache_hit_returns_same_class(self): + colnames = ("name", "age", "email") + rows1 = [("Alice", 30, "a@b.com")] + rows2 = [("Bob", 25, "b@c.com")] + _named_tuple_cache.clear() + result1 = named_tuple_factory(colnames, rows1) + result2 = named_tuple_factory(colnames, rows2) + # Same Row class should be reused + assert type(result1[0]) is type(result2[0]) + + def test_different_schemas_get_different_classes(self): + _named_tuple_cache.clear() + result1 = named_tuple_factory(("a", "b"), [(1, 2)]) + result2 = named_tuple_factory(("x", "y"), [(3, 4)]) + assert type(result1[0]) is not type(result2[0]) + assert result1[0]._fields == ("a", "b") + assert result2[0]._fields == ("x", "y") + + def test_case_difference_does_not_collide(self): + # Same names modulo case must not share a cached Row class: the raw + # (uncleaned) column names differ, so the cache key differs too. + _named_tuple_cache.clear() + result1 = named_tuple_factory(("Name", "Age"), [("Alice", 30)]) + result2 = named_tuple_factory(("name", "age"), [("bob", 25)]) + assert type(result1[0]) is not type(result2[0]) + assert result1[0]._fields == ("Name", "Age") + assert result2[0]._fields == ("name", "age") + + def test_column_order_does_not_collide(self): + # Same names in a different order must not share a cached Row class. + _named_tuple_cache.clear() + result1 = named_tuple_factory(("a", "b"), [(1, 2)]) + result2 = named_tuple_factory(("b", "a"), [(2, 1)]) + assert type(result1[0]) is not type(result2[0]) + assert result1[0]._fields == ("a", "b") + assert result2[0]._fields == ("b", "a") + + def test_cache_is_bounded_and_evicts_oldest(self): + # Guard against unbounded growth for applications executing many + # distinct ad hoc queries against highly variable/generated schemas. + _named_tuple_cache.clear() + original_max_size = query_module._NAMED_TUPLE_CACHE_MAX_SIZE + query_module._NAMED_TUPLE_CACHE_MAX_SIZE = 3 + try: + for i in range(5): + named_tuple_factory((f"col_{i}",), [(i,)]) + assert len(_named_tuple_cache) == 3 + # Oldest entries (col_0, col_1) should have been evicted first. + assert ("col_0",) not in _named_tuple_cache + assert ("col_1",) not in _named_tuple_cache + assert ("col_4",) in _named_tuple_cache + finally: + query_module._NAMED_TUPLE_CACHE_MAX_SIZE = original_max_size + _named_tuple_cache.clear() + + +class TestNamedTupleFactoryCacheThreadSafety: + """ + Regression test for a race in the namedtuple-class cache's eviction + path: the check-evict-insert sequence on a cache miss was not + synchronized, so one thread's `next(iter(_named_tuple_cache))` (picking + an eviction victim) could race with another thread mutating the same + dict, raising `RuntimeError: dictionary changed size during iteration`; + separately, two threads that both observed the cache under its size + bound before either inserted could together push it past that bound. + This driver advertises free-threaded Python support (see the + ``Free Threading`` classifier in pyproject.toml and the ``3.14t`` CI + jobs), where such races are far more likely to manifest than under a + GIL-enabled interpreter. + + This test hammers the cache from many threads, each using many distinct + column-name sets (so almost every call is a miss, forcing constant + eviction once the bound -- lowered here for a fast, reliable repro -- + is reached), and asserts that no thread ever observes an exception and + that the cache never exceeds its stated bound. + """ + + def test_concurrent_misses_do_not_raise_and_respect_bound(self): + _named_tuple_cache.clear() + original_max_size = query_module._NAMED_TUPLE_CACHE_MAX_SIZE + max_size = 50 + query_module._NAMED_TUPLE_CACHE_MAX_SIZE = max_size + + n_threads = 32 + n_iters_per_thread = 100 + errors = [] + errors_lock = threading.Lock() + start_barrier = threading.Barrier(n_threads) + + def worker(tid): + # Synchronize thread start as tightly as possible to maximize + # the chance of many threads racing into the cache-miss path + # (and the eviction it triggers) together. + start_barrier.wait() + for i in range(n_iters_per_thread): + colnames = (f"t{tid}_col_{i}_a", f"t{tid}_col_{i}_b", f"t{tid}_col_{i}_c") + try: + named_tuple_factory(colnames, [(1, 2, 3)]) + except Exception as e: + with errors_lock: + errors.append((tid, i, e)) + + try: + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [], ( + "named_tuple_factory raised under concurrent cache misses " + "(expected no exceptions): %r" % (errors,) + ) + assert len(_named_tuple_cache) <= max_size, ( + "cache grew past its stated bound (%d) under concurrent " + "misses: size=%d" % (max_size, len(_named_tuple_cache)) + ) + finally: + query_module._NAMED_TUPLE_CACHE_MAX_SIZE = original_max_size + _named_tuple_cache.clear()