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