(fix) row_parser: restrict ParseDesc cache to prepared statements and clear on shutdown

Fix unbounded cache growth for non-prepared queries by splitting the
cache logic into two paths:

- _get_or_build_parse_desc: cached path, used only when column_metadata
  comes from result_metadata (prepared statements with stable id()).
- _build_parse_desc: uncached path, used for inline metadata from
  non-prepared queries that creates a fresh list every execution.

Also add clear_parse_desc_cache() call in Cluster.shutdown() to release
cached ParseDesc entries (and their referenced type objects) when the
cluster is torn down.

Additional minor fixes:
- Move <tuple> cast after None check to avoid casting None.
- Expand comments documenting thread-safety guarantees and cache scope.

Author: mykaul

cassandra/cluster.py

@@ -114,6 +114,11 @@
 except ImportError:
     from cassandra.util import WeakSet  # NOQA
 
+try:
+    from cassandra.row_parser import clear_parse_desc_cache as _clear_parse_desc_cache
+except ImportError:
+    _clear_parse_desc_cache = None
+
 def _is_gevent_monkey_patched():
     if 'gevent.monkey' not in sys.modules:
         return False
@@ -1800,6 +1805,9 @@ def shutdown(self):
 
         _discard_cluster_shutdown(self)
 
+        if _clear_parse_desc_cache is not None:
+            _clear_parse_desc_cache()
+
     def __enter__(self):
         return self
 

cassandra/row_parser.pyx

@@ -21,40 +21,61 @@ include "ioutils.pyx"
 
 # Cache for ParseDesc objects keyed by id(column_metadata).
 # For prepared statements, result_metadata is stored on PreparedStatement
-# and reused across executions, so id() is stable.
+# and reused across executions, so id() is stable.  The cache is only
+# populated on the prepared-statement path (where column_metadata comes from
+# result_metadata); inline metadata from non-prepared queries is always fresh
+# and must not be cached to avoid unbounded growth.
+#
 # Cache value: (column_metadata_ref, column_encryption_policy_ref,
 #               protocol_version, desc, column_names, column_types)
+#
+# Thread safety: individual dict operations are atomic under CPython's GIL
+# and under free-threaded builds (PEP 703).  This cache relies on that
+# guarantee; no additional locking is needed.
 cdef dict _parse_desc_cache = {}
 
 cdef inline object _get_or_build_parse_desc(object column_metadata, object column_encryption_policy, int protocol_version):
-    """Look up or build a ParseDesc for the given column_metadata."""
+    """Look up or build a ParseDesc for the given column_metadata (cached path)."""
     cdef object cache_key = id(column_metadata)
-    cdef tuple cached = <tuple>_parse_desc_cache.get(cache_key)
+    cdef object cached_or_none = _parse_desc_cache.get(cache_key)
 
-    if cached is not None:
-        # Verify identity — the object at this id must be the same list
+    if cached_or_none is not None:
+        # Verify identity -- the object at this id must be the same list
         # and session-level settings must match
+        cached = <tuple>cached_or_none
         if (cached[0] is column_metadata and
             cached[1] is column_encryption_policy and
             cached[2] == protocol_version):
             return cached  # hit
 
-    # Cache miss — build everything
+    # Cache miss -- build everything
     cdef list column_names = [md[2] for md in column_metadata]
     cdef list column_types = [md[3] for md in column_metadata]
     cdef object desc = ParseDesc(
         column_names, column_types, column_encryption_policy,
         [ColDesc(md[0], md[1], md[2]) for md in column_metadata],
         make_deserializers(column_types), protocol_version)
 
-    cached = (column_metadata, column_encryption_policy, protocol_version,
-              desc, column_names, column_types)
-    _parse_desc_cache[cache_key] = cached
-    return cached
+    cdef tuple cached_entry = (column_metadata, column_encryption_policy,
+                               protocol_version, desc, column_names, column_types)
+    _parse_desc_cache[cache_key] = cached_entry
+    return cached_entry
+
+
+cdef inline object _build_parse_desc(object column_metadata, object column_encryption_policy, int protocol_version):
+    """Build a ParseDesc without caching (for non-prepared inline metadata)."""
+    cdef list column_names = [md[2] for md in column_metadata]
+    cdef list column_types = [md[3] for md in column_metadata]
+    cdef object desc = ParseDesc(
+        column_names, column_types, column_encryption_policy,
+        [ColDesc(md[0], md[1], md[2]) for md in column_metadata],
+        make_deserializers(column_types), protocol_version)
+    return (column_metadata, column_encryption_policy, protocol_version,
+            desc, column_names, column_types)
 
 
 def clear_parse_desc_cache():
-    """Clear the ParseDesc cache. Exposed for testing."""
+    """Clear the ParseDesc cache. Exposed for testing and Cluster.shutdown()."""
     _parse_desc_cache.clear()
 
 
@@ -68,7 +89,15 @@ def make_recv_results_rows(ColumnParser colparser):
 
         column_metadata = self.column_metadata or result_metadata
 
-        cached = _get_or_build_parse_desc(column_metadata, column_encryption_policy, protocol_version)
+        # Only use the cache for prepared statements (self.column_metadata is
+        # None, so column_metadata comes from result_metadata which is a
+        # stable list stored on PreparedStatement).  Inline metadata from
+        # non-prepared queries creates a fresh list every time and would
+        # cause unbounded cache growth.
+        if self.column_metadata is None and result_metadata is not None:
+            cached = _get_or_build_parse_desc(column_metadata, column_encryption_policy, protocol_version)
+        else:
+            cached = _build_parse_desc(column_metadata, column_encryption_policy, protocol_version)
         self.column_names = cached[4]
         self.column_types = cached[5]
         desc = cached[3]