TypeForm: Add instrumentation of full parses done in semanal.py's try_parse_as_type_expression()

Specifically:

- If you set MYPY_TYPEFORM_PROFILE_FULL_PARSE environment variable,
  mypy will output a .tsv to that filepath which characterizes the
  kinds of Expressions that try_parse_as_type_expression() in semanal.py
  was forced to do a full parse of, which was not rejected early.

- A misc/analyze_typeform_full_parse_profile.py script is added which
  takes those .tsvs and prints an expression-time summary (by total time)
  plus top-N descriptors per FAIL class.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: davidfstr

misc/analyze_typeform_full_parse_profile.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""
+Aggregate the full-parse profile log produced by mypy's
+SemanticAnalyzer.try_parse_as_type_expression() when run with
+MYPY_TYPEFORM_PROFILE_FULL_PARSE set.
+
+Usage:
+    # 1. Run mypy with the profile env var set; per-PID log files are
+    #    written as "<path>.<pid>":
+    MYPY_TYPEFORM_PROFILE_FULL_PARSE=/tmp/tf.log \\
+        python3 -m mypy --no-incremental -p your_package
+
+    # 2. Aggregate one or more per-PID files:
+    python3 misc/analyze_typeform_full_parse_profile.py /tmp/tf.log.*
+
+    # Optional: limit per-descriptor breakdown to top N rows per class.
+    python3 misc/analyze_typeform_full_parse_profile.py --top 20 /tmp/tf.log.*
+
+The script summarizes which (outcome, kind, subkind) classes account for
+the most full-parse time, and lists the top descriptors within each
+FAIL class -- the populations worth targeting with cheaper pre-filters
+upstream in try_parse_as_type_expression.
+
+See also:
+    - mypy/semanal.py: SemanticAnalyzer.try_parse_as_type_expression()
+    - mypy/semanal.py: _log_typeform_full_parse() (TSV schema docstring)
+    - misc/analyze_typeform_stats.py (aggregate counters via --dump-build-stats)
+"""
+
+from __future__ import annotations
+
+import argparse
+import statistics
+import sys
+from collections import defaultdict
+from collections.abc import Iterable
+
+
+def read_rows(paths: Iterable[str]) -> list[tuple[str, str, str, str, int]]:
+    rows: list[tuple[str, str, str, str, int]] = []
+    for path in paths:
+        with open(path) as f:
+            for line in f:
+                # Skip header lines (each per-PID file starts with one).
+                if line.startswith("outcome\t"):
+                    continue
+                parts = line.rstrip("\n").split("\t")
+                if len(parts) < 5:
+                    continue
+                outcome, kind, subkind, desc, dur_ns_str = parts[:5]
+                try:
+                    dur_ns = int(dur_ns_str)
+                except ValueError:
+                    continue
+                rows.append((outcome, kind, subkind, desc, dur_ns))
+    return rows
+
+
+def print_class_summary(rows: list[tuple[str, str, str, str, int]]) -> None:
+    buckets: dict[tuple[str, str, str], list[int]] = defaultdict(list)
+    total_ns = 0
+    for outcome, kind, subkind, _desc, dur_ns in rows:
+        buckets[(outcome, kind, subkind)].append(dur_ns)
+        total_ns += dur_ns
+
+    print("Class summary (by total time):")
+    print("=" * 80)
+    print(f"{'count':>7} {'total_ms':>10} {'mean_us':>9} {'med_us':>9} {'pct':>6}  class")
+    print("-" * 80)
+    ordered = sorted(
+        (
+            (sum(d), len(d), statistics.mean(d), statistics.median(d), key)
+            for key, d in buckets.items()
+        ),
+        reverse=True,
+    )
+    for total, n, mean, med, key in ordered:
+        pct = (100 * total / total_ns) if total_ns else 0
+        outcome, kind, subkind = key
+        print(
+            f"{n:>7} {total/1e6:>10.2f} {mean/1e3:>9.1f} {med/1e3:>9.1f} "
+            f"{pct:>5.1f}%  {outcome} {kind} {subkind}"
+        )
+    print("-" * 80)
+    print(f"TOTAL: {len(rows):,} events, {total_ns/1e6:.2f} ms")
+
+
+def print_fail_descriptors(rows: list[tuple[str, str, str, str, int]], top_n: int) -> None:
+    # Group FAIL rows by (kind, subkind) class, then by descriptor within each.
+    by_class: dict[tuple[str, str, str], dict[str, list[int]]] = defaultdict(
+        lambda: defaultdict(list)
+    )
+    for outcome, kind, subkind, desc, dur_ns in rows:
+        if outcome != "FAIL":
+            continue
+        by_class[(outcome, kind, subkind)][desc].append(dur_ns)
+
+    # Order classes by total FAIL time, descending.
+    class_totals = sorted(
+        ((sum(sum(d) for d in descs.values()), key, descs) for key, descs in by_class.items()),
+        reverse=True,
+    )
+    for total_ns, key, descs in class_totals:
+        outcome, kind, subkind = key
+        print()
+        print(
+            f"Top {top_n} descriptors in {outcome} {kind} {subkind} "
+            f"(class total {total_ns/1e6:.2f} ms):"
+        )
+        print("-" * 80)
+        print(f"{'count':>6} {'total_ms':>10} {'mean_us':>9}  descriptor")
+        rows_d = sorted(
+            ((sum(d), len(d), statistics.mean(d), desc) for desc, d in descs.items()), reverse=True
+        )
+        for tot, n, mean, desc in rows_d[:top_n]:
+            print(f"{n:>6} {tot/1e6:>10.3f} {mean/1e3:>9.1f}  {desc!r}")
+        if len(rows_d) > top_n:
+            print(f"... {len(rows_d) - top_n} more descriptors")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        formatter_class=argparse.RawDescriptionHelpFormatter, description=__doc__
+    )
+    parser.add_argument(
+        "files", nargs="+", help="One or more per-PID profile files (e.g. /tmp/tf.log.*)"
+    )
+    parser.add_argument(
+        "--top",
+        type=int,
+        default=20,
+        help="Max number of descriptors to list per FAIL class (default: 20)",
+    )
+    args = parser.parse_args()
+
+    rows = read_rows(args.files)
+    if not rows:
+        print("No data rows found in input files.", file=sys.stderr)
+        sys.exit(1)
+
+    print_class_summary(rows)
+    print_fail_descriptors(rows, args.top)
+
+
+if __name__ == "__main__":
+    main()

misc/analyze_typeform_stats.py

@@ -26,7 +26,6 @@
 
 See also:
     - mypy/semanal.py: SemanticAnalyzer.try_parse_as_type_expression()
-    - mypy/semanal.py: DEBUG_TYPE_EXPRESSION_FULL_PARSE_FAILURES
 """
 
 import re
@@ -70,6 +69,7 @@ def analyze_stats(output: str) -> None:
         print(
             f"  - Expensive failed full parses: {failures:,} ({(failures / total * 100):.1f}% of all calls)"
         )
+        print("    - Analyze further with misc/analyze_typeform_full_parse_profile.py")
 
 
 if __name__ == "__main__":

mypy/semanal.py

@@ -50,10 +50,12 @@
 
 from __future__ import annotations
 
+import os
 import re
+import time
 from collections.abc import Callable, Collection, Iterable, Iterator
 from contextlib import contextmanager
-from typing import Any, Final, TypeAlias as _TypeAlias, TypeGuard, TypeVar, cast
+from typing import Any, Final, TextIO, TypeAlias as _TypeAlias, TypeGuard, TypeVar, cast
 from typing_extensions import assert_never
 
 from mypy import errorcodes as codes, message_registry
@@ -318,11 +320,18 @@
 T = TypeVar("T")
 
 
-# Whether to print diagnostic information for failed full parses
-# in SemanticAnalyzer.try_parse_as_type_expression().
+# Instrumentation: If non-None, every expression that reaches the expensive
+# full-parse block of SemanticAnalyzer.try_parse_as_type_expression()
+# is logged to a .tsv by log_typeform_full_parse().
 #
-# See also: misc/analyze_typeform_stats.py
-DEBUG_TYPE_EXPRESSION_FULL_PARSE_FAILURES: Final = False
+# See also:
+# - misc/analyze_typeform_full_parse_profile.py
+# - misc/analyze_typeform_stats.py
+_TYPEFORM_PROFILE_FULL_PARSE_PATH: Final = os.environ.get("MYPY_TYPEFORM_PROFILE_FULL_PARSE")
+_typeform_full_parse_log_file: TextIO | None = None
+
+# TSV column names for the full-parse profile log
+_TYPEFORM_PROFILE_FULL_PARSE_HEADER = "outcome\tkind\tsubkind\tdescriptor\tdur_ns\n"
 
 
 FUTURE_IMPORTS: Final = {
@@ -8124,6 +8133,9 @@ def try_parse_as_type_expression(self, maybe_type_expr: Expression) -> None:
         else:
             assert_never(maybe_type_expr)
 
+        full_parse_t0 = (
+            time.perf_counter_ns() if _TYPEFORM_PROFILE_FULL_PARSE_PATH is not None else 0
+        )
         with self.isolated_error_analysis():
             try:
                 t = self.expr_to_analyzed_type(maybe_type_expr)
@@ -8133,17 +8145,6 @@ def try_parse_as_type_expression(self, maybe_type_expr: Expression) -> None:
                 # Not a type expression
                 t = None
 
-            if DEBUG_TYPE_EXPRESSION_FULL_PARSE_FAILURES and t is None:
-                original_flushed_files = set(self.errors.flushed_files)  # save
-                try:
-                    errors = self.errors.new_messages()  # capture
-                finally:
-                    self.errors.flushed_files = original_flushed_files  # restore
-
-                print(
-                    f"SA.try_parse_as_type_expression: Full parse failure: {maybe_type_expr}, errors={errors!r}"
-                )
-
         # Count full parse attempts for profiling
         if t is not None:
             self.type_expression_full_parse_success_count += 1
@@ -8152,6 +8153,12 @@ def try_parse_as_type_expression(self, maybe_type_expr: Expression) -> None:
 
         maybe_type_expr.as_type = t
 
+        if _TYPEFORM_PROFILE_FULL_PARSE_PATH is not None:
+            full_parse_t1 = time.perf_counter_ns()
+            self.log_typeform_full_parse(
+                maybe_type_expr, t is not None, full_parse_t1 - full_parse_t0
+            )
+
     @staticmethod
     def var_is_typing_special_form(var: Var) -> bool:
         return var.fullname.startswith("typing") and var.fullname in [
@@ -8168,6 +8175,92 @@ def var_is_typing_special_form(var: Var) -> bool:
             "typing.Union",
         ]
 
+    @staticmethod
+    def log_typeform_full_parse(expr: Expression, ok: bool, dur_ns: int) -> None:
+        """Log one entry into the full-parse block of try_parse_as_type_expression.
+
+        Active only when the MYPY_TYPEFORM_PROFILE_FULL_PARSE environment variable
+        is set to a file path. Each mypy process (worker) writes to its own file
+        named "<path>.<pid>" to avoid contention; concatenating those files yields
+        the complete profile. Aggregate with misc/analyze_typeform_full_parse_profile.py.
+
+        Output is tab-separated with one row per full-parse attempt:
+
+        outcome     "OK" if as_type was set, "FAIL" if the full parse rejected
+                    the expression (either by raising TypeTranslationError or by
+                    emitting errors during analysis).
+        kind        AST node kind: StrExpr | IndexExpr | OpExpr | (other).
+        subkind     For StrExpr: "ident", "dotident", or "other" (based on the
+                    string's shape). For IndexExpr: "Name" or "Member" (base
+                    kind). For OpExpr: always "|" (no other op reaches here).
+        descriptor  Short, type-specific identifier for the expression:
+                        StrExpr   -> the string value, truncated to 80 chars
+                                    (with " (N)" suffix when truncated).
+                        IndexExpr -> the full stringified expression (str(expr),
+                                    with tabs/newlines escaped).
+                        OpExpr    -> the full stringified expression (str(expr),
+                                    with tabs/newlines escaped).
+        dur_ns      Wall-clock nanoseconds spent in the full-parse block for
+                    this expression (measured around expr_to_analyzed_type
+                    plus the surrounding isolated_error_analysis ctx).
+
+        The first line of each file is the column header (same as above).
+        """
+        global _typeform_full_parse_log_file
+        if _typeform_full_parse_log_file is None:
+            assert _TYPEFORM_PROFILE_FULL_PARSE_PATH is not None
+            _typeform_full_parse_log_file = open(
+                f"{_TYPEFORM_PROFILE_FULL_PARSE_PATH}.{os.getpid()}", "a", buffering=1
+            )
+            _typeform_full_parse_log_file.write(_TYPEFORM_PROFILE_FULL_PARSE_HEADER)
+        outcome = "OK" if ok else "FAIL"
+        if isinstance(expr, StrExpr):
+            raw = expr.value
+            val = (
+                raw[:80]
+                .replace("\\", "\\\\")
+                .replace("\t", "\\t")
+                .replace("\n", "\\n")
+                .replace("\r", "\\r")
+            )
+            if len(raw) > 80:
+                val += f" ({len(raw)})"
+            if _IDENTIFIER_RE.fullmatch(raw):
+                subkind = "ident"
+            elif _DOTTED_IDENTIFIER_RE.fullmatch(raw):
+                subkind = "dotident"
+            else:
+                subkind = "other"
+            line = f"{outcome}\tStrExpr\t{subkind}\t{val}\t{dur_ns}\n"
+        elif isinstance(expr, IndexExpr):
+            base = expr.base
+            if isinstance(base, NameExpr):
+                subkind = "Name"
+            elif isinstance(base, MemberExpr):
+                subkind = "Member"
+            else:
+                subkind = type(base).__name__
+            desc = (
+                str(expr)
+                .replace("\\", "\\\\")
+                .replace("\t", "\\t")
+                .replace("\n", "\\n")
+                .replace("\r", "\\r")
+            )
+            line = f"{outcome}\tIndexExpr\t{subkind}\t{desc}\t{dur_ns}\n"
+        elif isinstance(expr, OpExpr):
+            desc = (
+                str(expr)
+                .replace("\\", "\\\\")
+                .replace("\t", "\\t")
+                .replace("\n", "\\n")
+                .replace("\r", "\\r")
+            )
+            line = f"{outcome}\tOpExpr\t|\t{desc}\t{dur_ns}\n"
+        else:
+            line = f"{outcome}\t{type(expr).__name__}\t\t\t{dur_ns}\n"
+        _typeform_full_parse_log_file.write(line)
+
     @contextmanager
     def isolated_error_analysis(self) -> Iterator[None]:
         """