T1.32e: the contract suite within CI's memory; a hang fails fast

CI's contract-python (ref-pandas3) job (run 38508814868 at 77bdc12) went
silent at 83% and was killed after 14 minutes. Reproduced on Linux
(python:3.12, the cell's pins, CI's command, two workers as on CI's
runner, about 8 GB): test_t132e_time_feature_family.py alone needed about
13 GB. Its module fixture held both of the family's runs parsed (600 MB of
observations, about 6.5 GB of Python objects) while a runner subprocess
made the second (3.5 GB), and capture_check.py, run on the same files,
parsed both again (6.6 GB). Workers were killed mid-file, and xdist then
waited on them until the job was killed. On macOS, with more memory, the
suite passed.

- capture_check.py reads the two files one pair of observations at a
  time (check_files, by each line's offset), giving the same problems and
  summary lines in the same order; check() over lists is unchanged. Peak
  on the family's runs: 6.6 GB down to 239 MB.
- The family test's fixture makes both runs before it parses the
  captured one, and never parses the capture-off run, which only the
  capture check reads, from its file; no assertion changes.
- contract/tests/conftest.py: under --dist loadfile, the two files whose
  module fixtures hold a family run parsed (the time and feature family
  and the multicategorical vectors, about 3.5 GB and 4 GB) share one
  worker, one after the other, so they are never in memory together;
  every other file goes to a worker of its own as before. No test is
  dropped (an --ignore would be, and l0's skips check refuses it).
- ci.yml: pytest-timeout fails a test past 10 minutes with its stack,
  and --max-worker-restart=0 ends the run when a worker dies instead of
  waiting on it.

Measured on Linux, two workers: 2,503 passed in 8 minutes, the machine's
memory in use at most 5.5 GB (7.8 GB before the fixture's order, about
13 GB needed before any of this).

Signed-off-by: Adi Wadaskar <awadaskar@nvidia.com>

Author: adi-kumo

.github/workflows/ci.yml

@@ -651,14 +651,21 @@ jobs:
         run: |
           printf '%s\n' $CELL_PINS duckdb==1.5.6 > "$RUNNER_TEMP/cell.txt"
           pip install --upgrade pip
-          pip install -c "$RUNNER_TEMP/cell.txt" $CELL_PINS duckdb==1.5.6 pyyaml jsonschema pytest pytest-xdist \
+          pip install -c "$RUNNER_TEMP/cell.txt" $CELL_PINS duckdb==1.5.6 pyyaml jsonschema pytest pytest-xdist pytest-timeout \
             ./packages/structured-data-models-api ./packages/structured-data-models-pql \
             './packages/structured-data-models-connectors[duckdb]' \
             ./packages/structured-data-models-relational-engine "$(ls native/*.whl)[test]"
       # In parallel, each test file on one worker so its module fixtures are
-      # built once; the results match a serial run's, test by test.
+      # built once; the results match a serial run's, test by test. The
+      # files whose module fixtures hold a family run parsed share one
+      # worker, one after the other (contract/tests/conftest.py), so this
+      # runner's memory holds one at a time. A test past 10 minutes fails
+      # with its stack, and a worker that dies ends the run rather than
+      # leaving xdist waiting on it.
       - name: The runner's controls and the route tests
-        run: python -m pytest contract/tests -q -n auto --dist loadfile
+        run: |
+          python -m pytest contract/tests -q -n auto --dist loadfile \
+            --timeout=600 --max-worker-restart=0
       # As in rust-test, whether the gate runs fails closed (T1.18); off
       # the phase branch, the candidate's own runners and comparator run.
       # Every cell gates, the floor's too (T1.18's own run alone reported

contract/runners/capture_check.py

@@ -41,6 +41,7 @@
 import json
 import re
 import sys
+from collections.abc import Iterable, Iterator
 from pathlib import Path
 from typing import Any
 
@@ -341,8 +342,7 @@ def _first_difference(on: Any, off: Any, path: str) -> str | None:
     return None if on == off else f'{path}: differs'
 
 
-def _cases(run: list[dict[str, Any]], name: str) -> list[str]:
-    ids = [item['case'] for item in run]
+def _cases(ids: list[str], name: str) -> list[str]:
     problems = []
     if not ids:
         problems.append(f'the {name} run holds no case')
@@ -352,24 +352,30 @@ def _cases(run: list[dict[str, Any]], name: str) -> list[str]:
     return problems
 
 
-def check(
-    on: list[dict[str, Any]],
-    off: list[dict[str, Any]],
-    required: tuple[str, ...] | list[str] = (),
+#: One captured observation and its capture-off twin (``None`` when the
+#: capture-off run lacks the case).
+Pair = tuple[dict[str, Any], dict[str, Any] | None]
+
+
+def _check(
+    on_ids: list[str],
+    off_ids: list[str],
+    required: tuple[str, ...] | list[str],
+    pairs: Iterable[Pair],
 ) -> list[str]:
-    r"""Every problem with the capture over two runs of the same cases,
-    each case in ``required`` among them."""
-    problems = _cases(on, 'captured') + _cases(off, 'uncaptured')
-    by_case = {item['case']: item for item in off}
-    if [item['case'] for item in on] != [item['case'] for item in off]:
+    r""":func:`check` over the two runs' case ids and their observations
+    in pairs, taken one pair at a time, so a caller reading files holds
+    one pair in memory, never a whole run."""
+    problems = _cases(on_ids, 'captured') + _cases(off_ids, 'uncaptured')
+    if on_ids != off_ids:
         problems.append('the two runs hold different cases')
+    on_set, off_set = set(on_ids), set(off_ids)
     for case in required:
-        if case not in by_case or case not in {i['case'] for i in on}:
+        if case not in off_set or case not in on_set:
             problems.append(f'{case}: required, and not in both runs')
-    for item in on:
+    for item, other in pairs:
         case = item['case']
         problems.extend(f'{case}: {p}' for p in complete(item))
-        other = by_case.get(case)
         if other is None:
             continue
         if other['driver']['capture'] is not None:
@@ -382,12 +388,88 @@ def check(
     return problems
 
 
-def _read(path: Path) -> list[dict[str, Any]]:
-    return [
-        json.loads(line)
-        for line in path.read_text(encoding='utf-8').splitlines()
-        if line
-    ]
+def check(
+    on: list[dict[str, Any]],
+    off: list[dict[str, Any]],
+    required: tuple[str, ...] | list[str] = (),
+) -> list[str]:
+    r"""Every problem with the capture over two runs of the same cases,
+    each case in ``required`` among them."""
+    by_case = {item['case']: item for item in off}
+    return _check(
+        [item['case'] for item in on],
+        [item['case'] for item in off],
+        required,
+        ((item, by_case.get(item['case'])) for item in on),
+    )
+
+
+def _lines(path: Path) -> Iterator[tuple[int, bytes]]:
+    r"""Each non-empty line of ``path`` with the offset it starts at."""
+    with path.open('rb') as handle:
+        while True:
+            offset = handle.tell()
+            line = handle.readline()
+            if not line:
+                return
+            if line.strip(b'\r\n'):
+                yield offset, line
+
+
+def _index(path: Path) -> list[tuple[str, int]]:
+    r"""Each observation's case id and the offset of its line, in file
+    order; one line is parsed at a time."""
+    return [(json.loads(line)['case'], offset) for offset, line in _lines(path)]
+
+
+def _at(path: Path, offset: int) -> dict[str, Any]:
+    with path.open('rb') as handle:
+        handle.seek(offset)
+        loaded: dict[str, Any] = json.loads(handle.readline())
+        return loaded
+
+
+def _summary(item: dict[str, Any]) -> str | None:
+    capture = item['driver']['capture']
+    logged = item['driver']['query_log']
+    if capture is None:
+        return None
+    return (
+        f'{item["case"]}: {len(capture["statements"])} statements and '
+        f'{len(capture["bindings"])} bindings captured on '
+        f'{capture["cursors"]} derived cursors; query log '
+        f'{logged["statements"]} and {logged["bindings"]}; '
+        f'{logged["compared"]} statement texts compared with it'
+    )
+
+
+def check_files(
+    on: Path, off: Path, required: tuple[str, ...] | list[str] = ()
+) -> tuple[list[str], list[str]]:
+    r""":func:`check` over two observation files, and each captured
+    case's summary line, reading one pair of observations at a time: a
+    family's runs reach hundreds of megabytes, which parsed whole took
+    gigabytes (T1.32e's time and feature family: 6.6 GB on Linux)."""
+    on_index, off_index = _index(on), _index(off)
+    off_at = dict(off_index)
+    summaries: list[str] = []
+
+    def pairs() -> Iterator[Pair]:
+        for case, offset in on_index:
+            item = _at(on, offset)
+            summary = _summary(item)
+            if summary is not None:
+                summaries.append(summary)
+            where = off_at.get(case)
+            yield item, None if where is None else _at(off, where)
+
+    problems = _check(
+        [case for case, _ in on_index],
+        [case for case, _ in off_index],
+        required,
+        pairs(),
+    )
+    return problems, summaries
 
 
 def main(argv: list[str] | None = None) -> int:
@@ -396,22 +478,13 @@ def main(argv: list[str] | None = None) -> int:
     parser.add_argument('--off', type=Path, required=True)
     parser.add_argument('--require', action='append', default=[])
     arguments = parser.parse_args(argv)
-    on, off = _read(arguments.on), _read(arguments.off)
-    problems = check(on, off, arguments.require)
+    problems, summaries = check_files(
+        arguments.on, arguments.off, arguments.require
+    )
     for problem in problems:
         print(problem, file=sys.stderr)
-    for item in on:
-        capture = item['driver']['capture']
-        logged = item['driver']['query_log']
-        if capture is None:
-            continue
-        print(
-            f'{item["case"]}: {len(capture["statements"])} statements and '
-            f'{len(capture["bindings"])} bindings captured on '
-            f'{capture["cursors"]} derived cursors; query log '
-            f'{logged["statements"]} and {logged["bindings"]}; '
-            f'{logged["compared"]} statement texts compared with it'
-        )
+    for summary in summaries:
+        print(summary)
     print('capture: ' + ('FAILED' if problems else 'complete and conserving'))
     return 1 if problems else 0
 

contract/tests/conftest.py

@@ -0,0 +1,54 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES.
+# All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+r"""How pytest-xdist shares the contract suite out under ``--dist loadfile``.
+
+Each test file goes to one worker, as ``loadfile`` sends it, except the
+files in :data:`ONE_AT_A_TIME`, whose module fixtures hold a whole family
+run parsed (about 3.5 GB and 4 GB on Linux). They go to one worker together,
+which runs them one after the other, so no two of them are in memory at
+once: side by side on CI's two workers and about 8 GB they ran the runner
+out of memory, and the worker killed mid-file left xdist waiting until the
+job was killed. Every test still runs; only which worker runs it changes.
+"""
+
+from __future__ import annotations
+
+from pathlib import PurePosixPath
+from typing import Any
+
+import pytest
+
+#: The files run on one worker, one after the other.
+ONE_AT_A_TIME = frozenset(
+    {
+        'test_t132e_multicategorical.py',
+        'test_t132e_time_feature_family.py',
+    }
+)
+
+#: The scope the files in :data:`ONE_AT_A_TIME` share.
+SHARED_SCOPE = 'contract/tests: a family run in memory'
+
+
+def scope_of(nodeid: str) -> str:
+    r"""The unit a worker takes: the test's file, as ``loadfile`` has it,
+    or :data:`SHARED_SCOPE` for a file in :data:`ONE_AT_A_TIME`."""
+    path = nodeid.split('::', 1)[0]
+    if PurePosixPath(path).name in ONE_AT_A_TIME:
+        return SHARED_SCOPE
+    return path
+
+
+@pytest.hookimpl(optionalhook=True)
+def pytest_xdist_make_scheduler(config: pytest.Config, log: Any) -> Any:
+    if config.getvalue('dist') != 'loadfile':
+        return None
+    from xdist.scheduler import LoadScopeScheduling
+
+    class _Scheduling(LoadScopeScheduling):  # type: ignore[misc]
+        def _split_scope(self, nodeid: str) -> str:
+            return scope_of(nodeid)
+
+    return _Scheduling(config, log)

contract/tests/test_t132e_time_feature_family.py

@@ -176,7 +176,7 @@ def _forced(tmp: Path) -> Path:
     return out
 
 
-def _run(cases: Path, out: Path, *extra: str) -> list[dict[str, Any]]:
+def _execute(cases: Path, out: Path, *extra: str) -> None:
     done = subprocess.run(
         [
             sys.executable,
@@ -194,6 +194,9 @@ def _run(cases: Path, out: Path, *extra: str) -> list[dict[str, Any]]:
         cwd=ROOT,
     )
     assert done.returncode == 0, done.stdout + done.stderr
+
+
+def _read(out: Path) -> list[dict[str, Any]]:
     return [
         json.loads(line)
         for line in out.read_text(encoding='utf-8').splitlines()
@@ -204,11 +207,14 @@ def _run(cases: Path, out: Path, *extra: str) -> list[dict[str, Any]]:
 def runs(tmp_path_factory: pytest.TempPathFactory) -> dict[str, Any]:
     tmp = tmp_path_factory.mktemp('t132e2')
     cases = _forced(tmp)
-    on = _run(cases, tmp / 'on.jsonl')
-    off = _run(cases, tmp / 'off.jsonl', '--no-capture')
+    # The capture-off run is read from its file by the capture check alone,
+    # never parsed here, and runs before the captured run is parsed: on
+    # Linux a run parsed holds about 3.5 GB and the runner making one about
+    # as much, past CI's runner when they overlap.
+    _execute(cases, tmp / 'on.jsonl')
+    _execute(cases, tmp / 'off.jsonl', '--no-capture')
     return {
-        'on': {o['case']: o for o in on},
-        'off': {o['case']: o for o in off},
+        'on': {o['case']: o for o in _read(tmp / 'on.jsonl')},
         'on_path': tmp / 'on.jsonl',
         'off_path': tmp / 'off.jsonl',
         'tmp': tmp,