Avoid false unreachable and redundant-expr warnings in loops more robustly and efficiently, and avoid multiple revealed type notes for the same line.

This change is an improvement over 9685171.  Besides fixing the regression reported in python#18606 and removing the duplicates reported in python#18511, it should significantly reduce the performance regression reported in python#18991.  At least running `Measure-command {python runtests.py self}` on my computer (with removed cache) is 10 % faster.

Author: tyralla

mypy/checker.py

@@ -24,7 +24,7 @@
 from mypy.constraints import SUPERTYPE_OF
 from mypy.erasetype import erase_type, erase_typevars, remove_instance_last_known_values
 from mypy.errorcodes import TYPE_VAR, UNUSED_AWAITABLE, UNUSED_COROUTINE, ErrorCode
-from mypy.errors import Errors, ErrorWatcher, report_internal_error
+from mypy.errors import Errors, ErrorWatcher, LoopErrorWatcher, report_internal_error
 from mypy.expandtype import expand_type
 from mypy.literals import Key, extract_var_from_literal_hash, literal, literal_hash
 from mypy.maptype import map_instance_to_supertype
@@ -600,19 +600,23 @@ def accept_loop(
             # on without bound otherwise)
             widened_old = len(self.widened_vars)
 
-            # Disable error types that we cannot safely identify in intermediate iteration steps:
-            warn_unreachable = self.options.warn_unreachable
-            warn_redundant = codes.REDUNDANT_EXPR in self.options.enabled_error_codes
-            self.options.warn_unreachable = False
-            self.options.enabled_error_codes.discard(codes.REDUNDANT_EXPR)
-
             iter = 1
             while True:
                 with self.binder.frame_context(can_skip=True, break_frame=2, continue_frame=1):
                     if on_enter_body is not None:
                         on_enter_body()
 
-                    self.accept(body)
+                    # We collect errors that indicate unreachability or redundant expressions
+                    # in the first iteration and remove them in subsequent iterations if the
+                    # related statement becomes reachable or non-redundant due to changes in
+                    # partial types:
+                    with LoopErrorWatcher(self.msg.errors) as watcher:
+                        self.accept(body)
+                    if iter == 1:
+                        uselessness_errors = watcher.useless_statements
+                    else:
+                        uselessness_errors.intersection_update(watcher.useless_statements)
+
                 partials_new = sum(len(pts.map) for pts in self.partial_types)
                 widened_new = len(self.widened_vars)
                 # Perform multiple iterations if something changed that might affect
@@ -633,16 +637,15 @@ def accept_loop(
                 if iter == 20:
                     raise RuntimeError("Too many iterations when checking a loop")
 
-            # If necessary, reset the modified options and make up for the postponed error checks:
-            self.options.warn_unreachable = warn_unreachable
-            if warn_redundant:
-                self.options.enabled_error_codes.add(codes.REDUNDANT_EXPR)
-            if warn_unreachable or warn_redundant:
-                with self.binder.frame_context(can_skip=True, break_frame=2, continue_frame=1):
-                    if on_enter_body is not None:
-                        on_enter_body()
-
-                    self.accept(body)
+            # Report those unreachable and redundant expression errors that have been
+            # identified in all iteration steps, as well as the revealed types identified
+            # in the last iteration step:
+            for notes_and_errors in itertools.chain(uselessness_errors, watcher.revealed_types):
+                context = Context(line=notes_and_errors[2], column=notes_and_errors[3])
+                context.end_line = notes_and_errors[4]
+                context.end_column = notes_and_errors[5]
+                reporter = self.msg.note if notes_and_errors[0] == codes.MISC else self.msg.fail
+                reporter(notes_and_errors[1], context, code=notes_and_errors[0])
 
             # If exit_condition is set, assume it must be False on exit from the loop:
             if exit_condition:

mypy/errors.py

@@ -6,7 +6,7 @@
 from collections import defaultdict
 from collections.abc import Iterable
 from typing import Callable, Final, NoReturn, Optional, TextIO, TypeVar
-from typing_extensions import Literal, TypeAlias as _TypeAlias
+from typing_extensions import Literal, Self, TypeAlias as _TypeAlias
 
 from mypy import errorcodes as codes
 from mypy.error_formatter import ErrorFormatter
@@ -179,7 +179,7 @@ def __init__(
         self._filter_deprecated = filter_deprecated
         self._filtered: list[ErrorInfo] | None = [] if save_filtered_errors else None
 
-    def __enter__(self) -> ErrorWatcher:
+    def __enter__(self) -> Self:
         self.errors._watchers.append(self)
         return self
 
@@ -220,6 +220,45 @@ def filtered_errors(self) -> list[ErrorInfo]:
         return self._filtered
 
 
+class LoopErrorWatcher(ErrorWatcher):
+    """Error watcher that filters and separately collects unreachable errors, redundant
+    expression errors, and revealed type notes when analysing loops iteratively to help
+    avoid making too-hasty reports."""
+
+    # Meaning of the entries: ErrorCode, message, line, column, end_line, end_column:
+    useless_statements: set[tuple[ErrorCode, str, int, int, int, int]]
+    revealed_types: set[tuple[ErrorCode, str, int, int, int, int]]
+
+    def __init__(
+        self,
+        errors: Errors,
+        *,
+        filter_errors: bool | Callable[[str, ErrorInfo], bool] = False,
+        save_filtered_errors: bool = False,
+        filter_deprecated: bool = False,
+    ) -> None:
+        super().__init__(
+            errors,
+            filter_errors=filter_errors,
+            save_filtered_errors=save_filtered_errors,
+            filter_deprecated=filter_deprecated,
+        )
+        self.useless_statements = set()
+        self.revealed_types = set()
+
+    def on_error(self, file: str, info: ErrorInfo) -> bool:
+        if info.code in (codes.UNREACHABLE, codes.REDUNDANT_EXPR):
+            self.useless_statements.add(
+                (info.code, info.message, info.line, info.column, info.end_line, info.end_column)
+            )
+            return True
+        if info.code == codes.MISC and info.message.startswith("Revealed type is "):
+            self.revealed_types.add(
+                (info.code, info.message, info.line, info.column, info.end_line, info.end_column)
+            )
+            return True
+        return super().on_error(file, info)
+
 class Errors:
     """Container for compile errors.
 

test-data/unit/check-narrowing.test

@@ -2346,8 +2346,7 @@ def f() -> bool: ...
 
 y = None
 while f():
-    reveal_type(y)  # N: Revealed type is "None" \
-                    # N: Revealed type is "Union[builtins.int, None]"
+    reveal_type(y)  # N: Revealed type is "Union[builtins.int, None]"
     y = 1
 reveal_type(y)  # N: Revealed type is "Union[builtins.int, None]"
 
@@ -2370,7 +2369,7 @@ class A:
 
 [builtins fixtures/primitives.pyi]
 
-[case testAvoidFalseUnreachableInLoop]
+[case testAvoidFalseUnreachableInLoop1]
 # flags: --warn-unreachable --python-version 3.11
 
 def f() -> int | None: ...
@@ -2383,6 +2382,29 @@ while x is not None or b():
 
 [builtins fixtures/bool.pyi]
 
+[case testAvoidFalseUnreachableInLoop2]
+# flags: --warn-unreachable --python-version 3.11
+
+y = None
+while y is None:
+    if y is None:
+        y = []
+    y.append(1)
+
+[builtins fixtures/list.pyi]
+
+[case testAvoidFalseUnreachableInLoop3]
+# flags: --warn-unreachable --python-version 3.11
+
+xs: list[int | None]
+y = None
+for x in xs:
+    if x is not None:
+        if y is None:
+            y = {}  # E: Need type annotation for "y" (hint: "y: Dict[<type>, <type>] = ...")
+
+[builtins fixtures/list.pyi]
+
 [case testAvoidFalseRedundantExprInLoop]
 # flags: --enable-error-code redundant-expr --python-version 3.11
 

test-data/unit/check-redefine2.test

@@ -628,8 +628,7 @@ def f1() -> None:
 def f2() -> None:
     x = None
     while int():
-        reveal_type(x) # N: Revealed type is "None" \
-                       # N: Revealed type is "Union[None, builtins.str]"
+        reveal_type(x) # N: Revealed type is "Union[None, builtins.str]"
         if int():
             x = ""
     reveal_type(x) # N: Revealed type is "Union[None, builtins.str]"
@@ -709,8 +708,7 @@ def b() -> None:
 def c() -> None:
     x = 0
     while int():
-        reveal_type(x) # N: Revealed type is "builtins.int" \
-                       # N: Revealed type is "Union[builtins.int, builtins.str, None]"
+        reveal_type(x)  # N: Revealed type is "Union[builtins.int, builtins.str, None]"
         if int():
             x = ""
             continue