From ece08e0887af34330e82454f394c1b16539173f9 Mon Sep 17 00:00:00 2001
From: Ronny Pfannschmidt <opensource@ronnypfannschmidt.de>
Date: Wed, 22 Jul 2026 18:02:26 +0200
Subject: [PATCH] WIP: reduce collection memory via dense CallSpec and lazy
 node state

Experimental changes aimed at large parametrized suites (#619): store
CallSpec2 as parallel tuples, derive NodeKeywords without eager marker
dicts, and defer TopRequest until first use. Adds
bench/collection_memory.py to measure RSS per collected item.

Co-authored-by: Cursor AI <ai@cursor.sh>
Co-authored-by: Cursor Grok 4.5 <grok@x.ai>
---
 bench/collection_memory.py     | 200 +++++++++++++++++++++++++++++++++
 src/_pytest/mark/structures.py |  95 ++++++++++++++--
 src/_pytest/python.py          | 163 ++++++++++++++++++++-------
 3 files changed, 407 insertions(+), 51 deletions(-)
 create mode 100644 bench/collection_memory.py

diff --git a/bench/collection_memory.py b/bench/collection_memory.py
new file mode 100644
index 00000000000..db9ecb8d267
--- /dev/null
+++ b/bench/collection_memory.py
@@ -0,0 +1,200 @@
+#!/usr/bin/env python3
+"""Measure RSS growth attributable to collecting parametrized items.
+
+Used to track collection memory density (related to issue #619).
+
+Examples::
+
+    python bench/collection_memory.py
+    python bench/collection_memory.py --counts 1000,10000,50000,100000
+    python bench/collection_memory.py --counts 50000 --json
+
+Reports bytes/item as (RSS after collection - RSS after importing pytest)
+divided by the number of collected items. Each count runs in a fresh
+subprocess so import/setup costs do not accumulate.
+"""
+
+from __future__ import annotations
+
+import argparse
+from dataclasses import asdict
+from dataclasses import dataclass
+import json
+import subprocess
+import sys
+import tempfile
+import textwrap
+
+
+@dataclass(frozen=True, slots=True)
+class CollectionMemoryResult:
+    count: int
+    items: int
+    baseline_rss: int
+    after_rss: int
+    delta_rss: int
+    bytes_per_item: float
+    request_deferred: bool
+    keywords_markers_is_none: bool
+    callspec_marks_type: str
+    callspec_argnames: tuple[str, ...]
+
+
+WORKER = textwrap.dedent(
+    r"""
+    from __future__ import annotations
+
+    import contextlib
+    import gc
+    import io
+    import sys
+    from pathlib import Path
+
+    import pytest
+
+    def rss() -> int:
+        with open("/proc/self/status") as f:
+            for line in f:
+                if line.startswith("VmRSS:"):
+                    return int(line.split()[1]) * 1024
+        return 0
+
+    n = int(sys.argv[1])
+    path = Path("bench_param.py")
+    path.write_text(
+        "import pytest\n\n"
+        f'@pytest.mark.parametrize("i", range({n}))\n'
+        "def test_it(i):\n"
+        "    assert i >= 0\n"
+    )
+
+    gc.collect()
+    baseline = rss()
+
+    class Plugin:
+        def pytest_collection_finish(self, session) -> None:
+            self.items = session.items
+            item = session.items[0]
+            self.request_deferred = item.__dict__.get("_request") is None
+            self.keywords_markers_is_none = item.keywords._markers is None
+            callspec = item.callspec
+            self.callspec_marks_type = type(callspec.marks).__name__
+            self.callspec_argnames = callspec._argnames
+            gc.collect()
+            self.after = rss()
+
+    plugin = Plugin()
+    with contextlib.redirect_stdout(io.StringIO()):
+        ret = pytest.main(
+            ["-q", "--collect-only", "-p", "no:cacheprovider", "--no-header", str(path)],
+            plugins=[plugin],
+        )
+    if ret != 0:
+        raise SystemExit(ret)
+
+    items = len(plugin.items)
+    delta = plugin.after - baseline
+    # Machine-readable one-liner for the parent process.
+    print(
+        "RESULT",
+        items,
+        baseline,
+        plugin.after,
+        delta,
+        delta / items,
+        int(plugin.request_deferred),
+        int(plugin.keywords_markers_is_none),
+        plugin.callspec_marks_type,
+        ",".join(plugin.callspec_argnames),
+        sep="\t",
+    )
+    """
+)
+
+
+def _run_one(count: int) -> CollectionMemoryResult:
+    with tempfile.TemporaryDirectory(prefix="pytest-collection-mem-") as tmp:
+        proc = subprocess.run(
+            [sys.executable, "-c", WORKER, str(count)],
+            capture_output=True,
+            text=True,
+            cwd=tmp,
+            check=False,
+        )
+    if proc.returncode != 0:
+        raise RuntimeError(
+            f"worker failed for count={count} (exit {proc.returncode})\n"
+            f"stdout:\n{proc.stdout}\nstderr:\n{proc.stderr}"
+        )
+    line = next(ln for ln in proc.stdout.splitlines() if ln.startswith("RESULT\t"))
+    (
+        _tag,
+        items_s,
+        baseline_s,
+        after_s,
+        delta_s,
+        per_s,
+        req_s,
+        kw_s,
+        marks_type,
+        argnames_s,
+    ) = line.split("\t", 9)
+    argnames = tuple(a for a in argnames_s.split(",") if a)
+    return CollectionMemoryResult(
+        count=count,
+        items=int(items_s),
+        baseline_rss=int(baseline_s),
+        after_rss=int(after_s),
+        delta_rss=int(delta_s),
+        bytes_per_item=float(per_s),
+        request_deferred=bool(int(req_s)),
+        keywords_markers_is_none=bool(int(kw_s)),
+        callspec_marks_type=marks_type,
+        callspec_argnames=argnames,
+    )
+
+
+def _parse_counts(value: str) -> list[int]:
+    counts = [int(part.strip()) for part in value.split(",") if part.strip()]
+    if not counts or any(c <= 0 for c in counts):
+        raise argparse.ArgumentTypeError("counts must be positive integers")
+    return counts
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--counts",
+        type=_parse_counts,
+        default=[1_000, 10_000, 50_000, 100_000],
+        help="Comma-separated item counts to measure (default: 1000,10000,50000,100000)",
+    )
+    parser.add_argument(
+        "--json",
+        action="store_true",
+        help="Emit JSON instead of a human-readable table",
+    )
+    args = parser.parse_args(argv)
+
+    results = [_run_one(count) for count in args.counts]
+
+    if args.json:
+        print(json.dumps([asdict(r) for r in results], indent=2))
+        return 0
+
+    print(f"python={sys.version.split()[0]} pytest={__import__('pytest').__version__}")
+    print(
+        f"{'count':>10} {'items':>10} {'Δ MiB':>10} {'B/item':>10} "
+        f"{'req deferred':>12} {'kw lazy':>8} {'marks':>8}"
+    )
+    for r in results:
+        print(
+            f"{r.count:10d} {r.items:10d} {r.delta_rss / 1024 / 1024:10.1f} "
+            f"{r.bytes_per_item:10.0f} {r.request_deferred!s:>12} "
+            f"{r.keywords_markers_is_none!s:>8} {r.callspec_marks_type:>8}"
+        )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/src/_pytest/mark/structures.py b/src/_pytest/mark/structures.py
index 5449b17a1c6..bcedb7c736c 100644
--- a/src/_pytest/mark/structures.py
+++ b/src/_pytest/mark/structures.py
@@ -641,52 +641,125 @@ def __getattr__(self, name: str) -> MarkDecorator:
 
 @final
 class NodeKeywords(MutableMapping[str, Any]):
+    """Keyword mapping for a node.
+
+    Explicit writes go into ``_markers``, allocated lazily on first write.
+
+    Until then (and in addition), keywords are derived on read from:
+
+    * ``node.name``
+    * ``node.own_markers``
+    * ``node._obj.__dict__`` when present (test function attributes)
+    * ``node.callspec.id`` when the node is parametrized
+    * parent node keywords
+    """
+
     __slots__ = ("_markers", "node", "parent")
 
     def __init__(self, node: Node) -> None:
         self.node = node
         self.parent = node.parent
-        self._markers = {node.name: True}
+        # Lazily allocated on first write.
+        self._markers: dict[str, Any] | None = None
+
+    def _ensure_markers(self) -> dict[str, Any]:
+        markers = self._markers
+        if markers is None:
+            markers = {}
+            self._markers = markers
+        return markers
+
+    def _lookup_derived(self, key: str) -> Any:
+        """Return a derived keyword value or raise ``KeyError``."""
+        if key == self.node.name:
+            return True
+        callspec = getattr(self.node, "callspec", None)
+        if callspec is not None and key == callspec.id:
+            return True
+        obj = getattr(self.node, "_obj", None)
+        if obj is not None:
+            obj_dict = getattr(obj, "__dict__", None)
+            if obj_dict is not None and key in obj_dict:
+                return obj_dict[key]
+        for mark in getattr(self.node, "own_markers", ()):
+            if mark.name == key:
+                return mark
+        raise KeyError(key)
 
     def __getitem__(self, key: str) -> Any:
+        markers = self._markers
+        if markers is not None:
+            try:
+                return markers[key]
+            except KeyError:
+                pass
         try:
-            return self._markers[key]
+            return self._lookup_derived(key)
         except KeyError:
             if self.parent is None:
                 raise
             return self.parent.keywords[key]
 
     def __setitem__(self, key: str, value: Any) -> None:
-        self._markers[key] = value
+        self._ensure_markers()[key] = value
 
     # Note: we could've avoided explicitly implementing some of the methods
     # below and use the collections.abc fallback, but that would be slow.
 
     def __contains__(self, key: object) -> bool:
-        return key in self._markers or (
-            self.parent is not None and key in self.parent.keywords
-        )
+        if not isinstance(key, str):
+            return False
+        markers = self._markers
+        if markers is not None and key in markers:
+            return True
+        try:
+            self._lookup_derived(key)
+            return True
+        except KeyError:
+            return self.parent is not None and key in self.parent.keywords
 
     def update(  # type: ignore[override]
         self,
         other: Mapping[str, Any] | Iterable[tuple[str, Any]] = (),
         **kwds: Any,
     ) -> None:
-        self._markers.update(other)
-        self._markers.update(kwds)
+        markers = self._ensure_markers()
+        markers.update(other)
+        markers.update(kwds)
 
     def __delitem__(self, key: str) -> None:
         raise ValueError("cannot delete key in keywords dict")
 
     def __iter__(self) -> Iterator[str]:
         # Doesn't need to be fast.
-        yield from self._markers
+        seen: set[str] = set()
+        markers = self._markers
+        if markers is not None:
+            for key in markers:
+                seen.add(key)
+                yield key
+        for key, _value in self._iter_derived():
+            if key not in seen:
+                seen.add(key)
+                yield key
         if self.parent is not None:
             for keyword in self.parent.keywords:
-                # self._marks and self.parent.keywords can have duplicates.
-                if keyword not in self._markers:
+                if keyword not in seen:
                     yield keyword
 
+    def _iter_derived(self) -> Iterator[tuple[str, Any]]:
+        yield self.node.name, True
+        callspec = getattr(self.node, "callspec", None)
+        if callspec is not None and callspec._idlist:
+            yield callspec.id, True
+        obj = getattr(self.node, "_obj", None)
+        if obj is not None:
+            obj_dict = getattr(obj, "__dict__", None)
+            if obj_dict is not None:
+                yield from obj_dict.items()
+        for mark in getattr(self.node, "own_markers", ()):
+            yield mark.name, mark
+
     def __len__(self) -> int:
         # Doesn't need to be fast.
         return sum(1 for keyword in self)
diff --git a/src/_pytest/python.py b/src/_pytest/python.py
index 2f3c908c666..55efe84de89 100644
--- a/src/_pytest/python.py
+++ b/src/_pytest/python.py
@@ -310,9 +310,8 @@ def obj(self):
             # used to avoid Function marker duplication
             if self._ALLOW_MARKERS:
                 self.own_markers.extend(get_unpacked_marks(self.obj))
-                # This assumes that `obj` is called before there is a chance
-                # to add custom keys to `self.keywords`, so no fear of overriding.
-                self.keywords.update((mark.name, mark) for mark in self.own_markers)
+                # Marker names and obj.__dict__ keys are derived by NodeKeywords
+                # on read; no need to copy them into keywords here.
         return obj
 
     @obj.setter
@@ -511,7 +510,7 @@ def _genfunctions(self, name: str, funcobj) -> Iterator[Function]:
                     name=subname,
                     callspec=callspec,
                     fixtureinfo=fixtureinfo,
-                    keywords={callspec.id: True},
+                    # callspec.id is derived by NodeKeywords; no need to copy it in.
                     originalname=name,
                 )
 
@@ -1150,27 +1149,70 @@ def _make_error_prefix(self) -> str:
             return ""
 
 
+class _CallSpecMap(Mapping[str, Any]):
+    """Dense zip-map over parallel argname/value tuples for CallSpec2."""
+
+    __slots__ = ("_keys", "_values")
+
+    def __init__(self, keys: tuple[str, ...], values: tuple[Any, ...]) -> None:
+        self._keys = keys
+        self._values = values
+
+    def __getitem__(self, key: str) -> Any:
+        try:
+            return self._values[self._keys.index(key)]
+        except ValueError:
+            raise KeyError(key) from None
+
+    def __iter__(self) -> Iterator[str]:
+        return iter(self._keys)
+
+    def __len__(self) -> int:
+        return len(self._keys)
+
+    def __contains__(self, key: object) -> bool:
+        return isinstance(key, str) and key in self._keys
+
+    def __repr__(self) -> str:
+        return f"{{{', '.join(f'{k!r}: {v!r}' for k, v in self.items())}}}"
+
+
 @final
-@dataclasses.dataclass(frozen=True)
+@dataclasses.dataclass(frozen=True, slots=True)
 class CallSpec2:
     """A planned parameterized invocation of a test function.
 
     Calculated during collection for a given test function's Metafunc.
     Once collection is over, each callspec is turned into a single Item
     and stored in item.callspec.
+
+    Stored densely as parallel tuples (argnames/values/indices/scopes) instead
+    of per-item dicts. Dict-like views are exposed via :class:`_CallSpecMap`.
     """
 
-    # arg name -> arg value which will be passed to a fixture of the same name.
-    params: dict[str, object] = dataclasses.field(default_factory=dict)
-    # arg name -> arg index.
-    indices: dict[str, int] = dataclasses.field(default_factory=dict)
-    # arg name -> parameter scope.
-    # Used for sorting parametrized resources.
-    _arg2scope: Mapping[str, Scope] = dataclasses.field(default_factory=dict)
+    _argnames: tuple[str, ...] = ()
+    _values: tuple[object, ...] = ()
+    _indices: tuple[int, ...] = ()
+    _scopes: tuple[Scope, ...] = ()
     # Parts which will be added to the item's name in `[..]` separated by "-".
-    _idlist: Sequence[str] = dataclasses.field(default_factory=tuple)
+    _idlist: tuple[str, ...] = ()
     # Marks which will be applied to the item.
-    marks: list[Mark] = dataclasses.field(default_factory=list)
+    marks: tuple[Mark, ...] = ()
+
+    @property
+    def params(self) -> Mapping[str, object]:
+        """Arg name -> arg value which will be passed to a fixture of same name."""
+        return _CallSpecMap(self._argnames, self._values)
+
+    @property
+    def indices(self) -> Mapping[str, int]:
+        """Arg name -> arg index."""
+        return _CallSpecMap(self._argnames, self._indices)
+
+    @property
+    def _arg2scope(self) -> Mapping[str, Scope]:
+        """Arg name -> parameter scope (for sorting parametrized resources)."""
+        return _CallSpecMap(self._argnames, self._scopes)
 
     def setmulti(
         self,
@@ -1183,23 +1225,33 @@ def setmulti(
         param_index: int,
         nodeid: str,
     ) -> CallSpec2:
-        params = self.params.copy()
-        indices = self.indices.copy()
-        arg2scope = dict(self._arg2scope)
-        for arg, val in zip(argnames, valset, strict=True):
-            if arg in params:
+        argnames_t = tuple(argnames)
+        values_t = tuple(valset)
+        if len(argnames_t) != len(values_t):
+            raise ValueError("argnames and valset must have the same length")
+        for arg in argnames_t:
+            if arg in self._argnames:
                 raise nodes.Collector.CollectError(
                     f"{nodeid}: duplicate parametrization of {arg!r}"
                 )
-            params[arg] = val
-            indices[arg] = param_index
-            arg2scope[arg] = scope
         return CallSpec2(
-            params=params,
-            indices=indices,
-            _arg2scope=arg2scope,
-            _idlist=self._idlist if id is HIDDEN_PARAM else [*self._idlist, id],
-            marks=[*self.marks, *normalize_mark_list(marks)],
+            _argnames=self._argnames + argnames_t,
+            _values=self._values + values_t,
+            _indices=self._indices + (param_index,) * len(argnames_t),
+            _scopes=self._scopes + (scope,) * len(argnames_t),
+            _idlist=self._idlist if id is HIDDEN_PARAM else (*self._idlist, id),
+            marks=(*self.marks, *normalize_mark_list(marks)),
+        )
+
+    def with_indices(self, indices: Mapping[str, int]) -> CallSpec2:
+        """Return a copy with replaced per-arg indices (used after parametrize)."""
+        return CallSpec2(
+            _argnames=self._argnames,
+            _values=self._values,
+            _indices=tuple(indices[name] for name in self._argnames),
+            _scopes=self._scopes,
+            _idlist=self._idlist,
+            marks=self.marks,
         )
 
     def getparam(self, name: str) -> object:
@@ -1586,10 +1638,20 @@ def _validate_if_using_arg_names(
                     )
 
     def _recompute_direct_params_indices(self) -> None:
-        for argname, param_type in self._params_directness.items():
-            if param_type == "direct":
-                for i, callspec in enumerate(self._calls):
-                    callspec.indices[argname] = i
+        direct_args = [
+            argname
+            for argname, param_type in self._params_directness.items()
+            if param_type == "direct"
+        ]
+        if not direct_args:
+            return
+        new_calls: list[CallSpec2] = []
+        for i, callspec in enumerate(self._calls):
+            indices = dict(callspec.indices)
+            for argname in direct_args:
+                indices[argname] = i
+            new_calls.append(callspec.with_indices(indices))
+        self._calls = new_calls
 
 
 def _infer_parametrize_scope(
@@ -1707,11 +1769,8 @@ def __init__(
 
         # todo: this is a hell of a hack
         # https://github.com/pytest-dev/pytest/issues/4569
-        # Note: the order of the updates is important here; indicates what
-        # takes priority (ctor argument over function attributes over markers).
-        # Take own_markers only; NodeKeywords handles parent traversal on its own.
-        self.keywords.update((mark.name, mark) for mark in self.own_markers)
-        self.keywords.update(self.obj.__dict__)
+        # Marker names, function attributes and callspec.id are derived by
+        # NodeKeywords on read. Only retain explicit keyword overrides here.
         if keywords:
             self.keywords.update(keywords)
 
@@ -1720,7 +1779,9 @@ def __init__(
             fixtureinfo = fm.getfixtureinfo(self, self.obj, self.cls)
         self._fixtureinfo: FuncFixtureInfo = fixtureinfo
         self.fixturenames = fixtureinfo.names_closure
-        self._initrequest()
+        # Defer TopRequest until first access or test setup (see _request).
+        self.funcargs: dict[str, object] = {}
+        self.__dict__["_request"] = None
 
     # todo: determine sound type limitations
     @classmethod
@@ -1729,8 +1790,27 @@ def from_parent(cls, parent, **kw) -> Self:
         return super().from_parent(parent=parent, **kw)
 
     def _initrequest(self) -> None:
-        self.funcargs: dict[str, object] = {}
-        self._request = fixtures.TopRequest(self, _ispytest=True)
+        self.funcargs = {}
+        self.__dict__["_request"] = fixtures.TopRequest(self, _ispytest=True)
+
+    @property
+    def _request(self) -> fixtures.FixtureRequest | Literal[False] | None:
+        """Fixture request for this item.
+
+        ``None`` means not yet created (lazy); first read constructs a
+        :class:`~_pytest.fixtures.TopRequest`. ``False`` means cleared after a
+        run (see :func:`_pytest.runner.runtestprotocol`) and must be restored
+        via :meth:`_initrequest` before the next run.
+        """
+        req = self.__dict__["_request"]
+        if req is None:
+            self._initrequest()
+            req = self.__dict__["_request"]
+        return cast(fixtures.FixtureRequest | Literal[False] | None, req)
+
+    @_request.setter
+    def _request(self, value: fixtures.FixtureRequest | Literal[False] | None) -> None:
+        self.__dict__["_request"] = value
 
     @property
     def function(self):
@@ -1775,7 +1855,10 @@ def runtest(self) -> None:
         self.ihook.pytest_pyfunc_call(pyfuncitem=self)
 
     def setup(self) -> None:
-        self._request._fillfixtures()
+        # _request property lazily constructs TopRequest when still deferred.
+        if self.__dict__.get("_request") is False:
+            self._initrequest()
+        self._request._fillfixtures()  # type: ignore[union-attr]
 
     def _traceback_filter(self, excinfo: ExceptionInfo[BaseException]) -> Traceback:
         if hasattr(self, "_obj") and not self.config.getoption("fulltrace", False):