Tweak fix partitioning to closer align to lint partitioning (pant…

…sbuild#19796)

pantsbuild#17403 (comment)
describes a bit of a split between `lint` and `fix` in terms of
batching.

This change fixes pantsbuild#19634 by switching `AbstractFixRequest.files` to
report `self.snapshot.files` which is already de-duplicated. (Note that
the elements in the case of `lint` were the filepaths directly
translated from addresses. Maybe it's worth de-duplicating at a higher
layer, but 🤷‍♂️ )

This change also fixes pantsbuild#17403 (which really is a duplicate of pantsbuild#15069
manifest in new and exciting ways) by de-duplicating the `fix`-specific
batching strategy. It didn't sit right to me that in the reproduction
repo we did 1 batches of 2 files for `fix`, but 1 batch of 4 files for
`lint`.

src/python/pants/core/goals/fix.py

@@ -36,6 +36,7 @@
 from pants.util.collections import partition_sequentially
 from pants.util.docutil import bin_name
 from pants.util.logging import LogLevel
+from pants.util.ordered_set import FrozenOrderedSet
 from pants.util.strutil import softwrap, strip_v2_chroot_path
 
 logger = logging.getLogger(__name__)
@@ -126,7 +127,7 @@ class Batch(AbstractLintRequest.Batch):
 
         @property
         def files(self) -> tuple[str, ...]:
-            return self.elements
+            return tuple(FrozenOrderedSet(self.elements))
 
     @classmethod
     def _get_rules(cls) -> Iterable[UnionRule]:
@@ -294,7 +295,7 @@ def batch_by_size(files: Iterable[str]) -> Iterator[tuple[str, ...]]:
             yield tuple(batch)
 
     def _make_disjoint_batch_requests() -> Iterable[_FixBatchRequest]:
-        partition_infos: Sequence[Tuple[Type[AbstractFixRequest], Any]]
+        partition_infos: Iterable[Tuple[Type[AbstractFixRequest], Any]]
         files: Sequence[str]
 
         partition_infos_by_files = defaultdict(list)
@@ -306,7 +307,8 @@ def _make_disjoint_batch_requests() -> Iterable[_FixBatchRequest]:
 
         files_by_partition_info = defaultdict(list)
         for file, partition_infos in partition_infos_by_files.items():
-            files_by_partition_info[tuple(partition_infos)].append(file)
+            deduped_partition_infos = FrozenOrderedSet(partition_infos)
+            files_by_partition_info[deduped_partition_infos].append(file)
 
         for partition_infos, files in files_by_partition_info.items():
             for batch in batch_by_size(files):

src/python/pants/core/goals/fix_test.py

@@ -36,10 +36,17 @@
     Snapshot,
 )
 from pants.engine.rules import Get, QueryRule, collect_rules, rule
-from pants.engine.target import FieldSet, MultipleSourcesField, SingleSourceField, Target
+from pants.engine.target import (
+    FieldSet,
+    MultipleSourcesField,
+    SingleSourceField,
+    StringField,
+    Target,
+)
 from pants.option.option_types import SkipOption
 from pants.option.subsystem import Subsystem
 from pants.testutil.rule_runner import RuleRunner
+from pants.testutil.rule_runner import logging as log_this
 from pants.util.logging import LogLevel
 from pants.util.meta import classproperty
 
@@ -128,9 +135,17 @@ class SmalltalkSource(SingleSourceField):
     pass
 
 
+# NB: This extra field is required to help us in `test_batches` below.
+#   With it, each `SmalltalkTarget` we instantiate will produce a different `SmalltalkFieldSet`
+#   (even with the same `source` field value), which then results in https://github.com/pantsbuild/pants/issues/17403.
+#   See https://github.com/pantsbuild/pants/pull/19796.
+class SmalltalkExtraField(StringField):
+    alias = "extra"
+
+
 class SmalltalkTarget(Target):
     alias = "smalltalk"
-    core_fields = (SmalltalkSource,)
+    core_fields = (SmalltalkSource, SmalltalkExtraField)
 
 
 @dataclass(frozen=True)
@@ -253,6 +268,7 @@ def fix_rule_runner(
     )
 
 
+@log_this(level=LogLevel.INFO)
 def run_fix(
     rule_runner: RuleRunner,
     *,
@@ -288,6 +304,30 @@ def write_files(rule_runner: RuleRunner) -> None:
     )
 
 
+def test_batches(capfd) -> None:
+    rule_runner = fix_rule_runner(
+        target_types=[SmalltalkTarget],
+        request_types=[SmalltalkNoopRequest],
+    )
+
+    rule_runner.write_files(
+        {
+            "BUILD": dedent(
+                """\
+                smalltalk(name='s1-1', source="duplicate1.st")
+                smalltalk(name='s1-2', source="duplicate1.st")
+                smalltalk(name='s2-1', source="duplicate1.st")
+                smalltalk(name='s2-2', source="duplicate2.st")
+                """,
+            ),
+            "duplicate1.st": "",
+            "duplicate2.st": "",
+        },
+    )
+    run_fix(rule_runner, target_specs=["::"])
+    assert capfd.readouterr().err.count("Smalltalk Did Not Change made no changes.") == 1
+
+
 def test_summary() -> None:
     rule_runner = fix_rule_runner(
         target_types=[FortranTarget, SmalltalkTarget],
@@ -503,7 +543,7 @@ class FixKitchenRequest(FixTargetsRequest):
         QueryRule(Partitions, [FixKitchenRequest.PartitionRequest]),
     ]
     rule_runner = RuleRunner(rules=rules)
-    print(rule_runner.write_files({"BUILD": "", "knife.utensil": "", "bowl.utensil": ""}))
+    rule_runner.write_files({"BUILD": "", "knife.utensil": "", "bowl.utensil": ""})
     partitions = rule_runner.request(Partitions, [FixKitchenRequest.PartitionRequest(field_sets)])
     assert len(partitions) == 1
     assert partitions[0].elements == ("bowl.utensil", "knife.utensil")

src/python/pants/core/goals/lint_test.py

@@ -26,7 +26,7 @@
 )
 from pants.core.util_rules.distdir import DistDir
 from pants.core.util_rules.environments import EnvironmentNameRequest
-from pants.core.util_rules.partitions import PartitionerType
+from pants.core.util_rules.partitions import PartitionerType, _EmptyMetadata
 from pants.engine.addresses import Address
 from pants.engine.environment import EnvironmentName
 from pants.engine.fs import PathGlobs, SpecsPaths, Workspace
@@ -403,6 +403,15 @@ def run_lint_rule(
         return result.exit_code, stdio_reader.get_stderr()
 
 
+def test_duplicate_files_in_fixer(rule_runner: RuleRunner) -> None:
+    assert MockFixRequest.Batch(
+        "tool_name",
+        ("a", "a"),
+        _EmptyMetadata(),
+        rule_runner.make_snapshot({"a": ""}),
+    ).files == ("a",)
+
+
 def test_invalid_target_noops(rule_runner: RuleRunner) -> None:
     exit_code, stderr = run_lint_rule(
         rule_runner, lint_request_types=[InvalidRequest], targets=[make_target()]