24103052497

Committed 07 Apr 2026 08:33PM UTC coverage: 52.311% (-40.6%) from 92.909%

Build # 24103052497

Build Type

Pull #23228

github

Committed by

web-flow

Commit Message

Merge 18fdfb0fd into b05152cd9

Pull Request Pull Request #23228: Add persistent dependency inference cache for incremental --changed-dependents

Coverage Stats

31 of 136 new or added lines in 2 files covered. (22.79%)

23028 existing lines in 605 files now uncovered.

31671 of 60544 relevant lines covered (52.31%)

1.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

38.46

/src/python/pants/backend/project_info/dependents.py

# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import json
import logging
import time
from collections import defaultdict
from collections.abc import Iterable
from dataclasses import dataclass
from enum import Enum

from pants.backend.project_info.incremental_dependents import (
    CachedEntry,
    IncrementalDependents,
    compute_source_fingerprint,
    get_cache_path,
    load_persisted_graph,
    save_persisted_graph,
)
from pants.base.build_environment import get_buildroot
from pants.engine.addresses import Address, Addresses
from pants.engine.collection import DeduplicatedCollection
from pants.engine.console import Console
from pants.engine.goal import Goal, GoalSubsystem, LineOriented
from pants.engine.internals.graph import resolve_dependencies
from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule
from pants.engine.target import (
    AllUnexpandedTargets,
    AlwaysTraverseDeps,
    Dependencies,
    DependenciesRequest,
)
from pants.option.option_types import BoolOption, EnumOption
from pants.util.frozendict import FrozenDict
from pants.util.logging import LogLevel
from pants.util.ordered_set import FrozenOrderedSet

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class AddressToDependents:
    mapping: FrozenDict[Address, FrozenOrderedSet[Address]]


class DependentsOutputFormat(Enum):
    """Output format for listing dependents.

    text: List all dependents as a single list of targets in plain text.
    json: List all dependents as a mapping `{target: [dependents]}`.
    """

    text = "text"
    json = "json"


@rule(desc="Map all targets to their dependents", level=LogLevel.DEBUG)
async def map_addresses_to_dependents(
    all_targets: AllUnexpandedTargets,
    incremental_cfg: IncrementalDependents,
) -> AddressToDependents:
    """Build a reverse dependency map (target -> set of its dependents).

    When incremental mode is enabled via `--incremental-dependents-enabled`, the forward
    dependency graph is persisted to disk. On subsequent runs, only targets whose source
    files have changed need their dependencies re-resolved, dramatically reducing wall time
    for large repos.
    """
    if not incremental_cfg.enabled:
        # Original behavior: resolve all dependencies from scratch.
        dependencies_per_target = await concurrently(
            resolve_dependencies(
                DependenciesRequest(
                    tgt.get(Dependencies),
                    should_traverse_deps_predicate=AlwaysTraverseDeps(),
                ),
                **implicitly(),
            )
            for tgt in all_targets
        )

        address_to_dependents = defaultdict(set)
        for tgt, dependencies in zip(all_targets, dependencies_per_target):
            for dependency in dependencies:
                address_to_dependents[dependency].add(tgt.address)
        return AddressToDependents(
            FrozenDict(
                {
                    addr: FrozenOrderedSet(dependents)
                    for addr, dependents in address_to_dependents.items()
                }
            )
        )

    # --- Incremental mode ---
    start_time = time.time()
    buildroot = get_buildroot()
    cache_path = get_cache_path()

    # Step 1: Load previous graph
    previous = load_persisted_graph(cache_path, buildroot)
    logger.warning(
        "Incremental dep graph: loaded %d cached entries from %s",
        len(previous),
        cache_path,
    )

    # Step 2: Classify targets as cached or changed
    changed_targets = []
    cached_results: list[tuple[Address, CachedEntry]] = []

    for tgt in all_targets:
        spec = tgt.address.spec
        fingerprint = compute_source_fingerprint(tgt.address, buildroot)

        cached_entry = previous.get(spec)
        if cached_entry is not None and cached_entry.fingerprint == fingerprint:
            cached_results.append((tgt.address, cached_entry))
        else:
            changed_targets.append(tgt)

    cache_hits = len(cached_results)
    cache_misses = len(changed_targets)
    logger.warning(
        "Incremental dep graph: %d cached, %d changed (out of %d total targets)",
        cache_hits,
        cache_misses,
        len(all_targets),
    )

    # Step 3: Resolve deps only for changed targets
    if changed_targets:
        fresh_deps_per_target = await concurrently(
            resolve_dependencies(
                DependenciesRequest(
                    tgt.get(Dependencies),
                    should_traverse_deps_predicate=AlwaysTraverseDeps(),
                ),
                **implicitly(),
            )
            for tgt in changed_targets
        )
    else:
        fresh_deps_per_target = []

    # Step 4: Build the reverse dependency map from merged results
    address_to_dependents: dict[Address, set[Address]] = defaultdict(set)

    # Build a spec → Address lookup from all_targets for resolving cached specs
    spec_to_address: dict[str, Address] = {tgt.address.spec: tgt.address for tgt in all_targets}

    # Process cached results (deps stored as address spec strings)
    for addr, entry in cached_results:
        for dep_spec in entry.deps:
            dep_addr = spec_to_address.get(dep_spec)
            if dep_addr is not None:
                address_to_dependents[dep_addr].add(addr)

    # Process freshly resolved results
    for tgt, deps in zip(changed_targets, fresh_deps_per_target):
        for dep_addr in deps:
            address_to_dependents[dep_addr].add(tgt.address)

    # Step 5: Save the updated forward graph for next run
    new_entries: dict[str, CachedEntry] = {}

    # Carry forward cached entries
    for addr, entry in cached_results:
        new_entries[addr.spec] = entry

    # Add fresh entries
    for tgt, deps in zip(changed_targets, fresh_deps_per_target):
        spec = tgt.address.spec
        fingerprint = compute_source_fingerprint(tgt.address, buildroot)
        new_entries[spec] = CachedEntry(
            fingerprint=fingerprint,
            deps=tuple(dep.spec for dep in deps),
        )

    save_persisted_graph(cache_path, buildroot, new_entries)

    elapsed = time.time() - start_time
    logger.warning(
        "Incremental dep graph: completed in %.1fs (%d from cache, %d resolved fresh)",
        elapsed,
        cache_hits,
        cache_misses,
    )

    return AddressToDependents(
        FrozenDict(
            {
                addr: FrozenOrderedSet(dependents)
                for addr, dependents in address_to_dependents.items()
            }
        )
    )


@dataclass(frozen=True)
class DependentsRequest:
    addresses: FrozenOrderedSet[Address]
    transitive: bool
    include_roots: bool

    def __init__(
        self, addresses: Iterable[Address], *, transitive: bool, include_roots: bool
    ) -> None:
        object.__setattr__(self, "addresses", FrozenOrderedSet(addresses))
        object.__setattr__(self, "transitive", transitive)
        object.__setattr__(self, "include_roots", include_roots)


class Dependents(DeduplicatedCollection[Address]):
    sort_input = True


@rule(level=LogLevel.DEBUG)
async def find_dependents(
    request: DependentsRequest, address_to_dependents: AddressToDependents
) -> Dependents:
    check = set(request.addresses)
    known_dependents: set[Address] = set()
    while True:
        dependents = set(known_dependents)
        for target in check:
            target_dependents = address_to_dependents.mapping.get(target, FrozenOrderedSet())
            dependents.update(target_dependents)
        check = dependents - known_dependents
        if not check or not request.transitive:
            result = (
                dependents | set(request.addresses)
                if request.include_roots
                else dependents - set(request.addresses)
            )
            return Dependents(result)
        known_dependents = dependents


class DependentsSubsystem(LineOriented, GoalSubsystem):
    name = "dependents"
    help = "List all targets that depend on any of the input files/targets."

    transitive = BoolOption(
        default=False,
        help="List all transitive dependents. If unspecified, list direct dependents only.",
    )
    closed = BoolOption(
        default=False,
        help="Include the input targets in the output, along with the dependents.",
    )
    format = EnumOption(
        default=DependentsOutputFormat.text,
        help="Output format for listing dependents.",
    )


class DependentsGoal(Goal):
    subsystem_cls = DependentsSubsystem
    environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY


async def list_dependents_as_plain_text(
    addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
) -> None:
    """Get dependents for given addresses and list them in the console as a single list."""
    dependents = await find_dependents(
        DependentsRequest(
            addresses,
            transitive=dependents_subsystem.transitive,
            include_roots=dependents_subsystem.closed,
        ),
        **implicitly(),
    )
    with dependents_subsystem.line_oriented(console) as print_stdout:
        for address in dependents:
            print_stdout(address.spec)


async def list_dependents_as_json(
    addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
) -> None:
    """Get dependents for given addresses and list them in the console in JSON."""
    dependents_group = await concurrently(
        find_dependents(
            DependentsRequest(
                (address,),
                transitive=dependents_subsystem.transitive,
                include_roots=dependents_subsystem.closed,
            ),
            **implicitly(),
        )
        for address in addresses
    )
    iterated_addresses = []
    for dependents in dependents_group:
        iterated_addresses.append(sorted([str(address) for address in dependents]))
    mapping = dict(zip([str(address) for address in addresses], iterated_addresses))
    output = json.dumps(mapping, indent=4)
    with dependents_subsystem.line_oriented(console) as print_stdout:
        print_stdout(output)


@goal_rule
async def dependents_goal(
    specified_addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
) -> DependentsGoal:
    if DependentsOutputFormat.text == dependents_subsystem.format:
        await list_dependents_as_plain_text(
            addresses=specified_addresses,
            dependents_subsystem=dependents_subsystem,
            console=console,
        )
    elif DependentsOutputFormat.json == dependents_subsystem.format:
        await list_dependents_as_json(
            addresses=specified_addresses,
            dependents_subsystem=dependents_subsystem,
            console=console,
        )
    return DependentsGoal(exit_code=0)


def rules():
    return collect_rules()

1	# Copyright 2020 Pants project contributors (see CONTRIBUTORS.md).
2	# Licensed under the Apache License, Version 2.0 (see LICENSE).
3	import json	2✔
4	import logging	2✔
5	import time	2✔
6	from collections import defaultdict	2✔
7	from collections.abc import Iterable	2✔
8	from dataclasses import dataclass	2✔
9	from enum import Enum	2✔
10
11	from pants.backend.project_info.incremental_dependents import (	2✔
12	CachedEntry,
13	IncrementalDependents,
14	compute_source_fingerprint,
15	get_cache_path,
16	load_persisted_graph,
17	save_persisted_graph,
18	)
19	from pants.base.build_environment import get_buildroot	2✔
20	from pants.engine.addresses import Address, Addresses	2✔
21	from pants.engine.collection import DeduplicatedCollection	2✔
22	from pants.engine.console import Console	2✔
23	from pants.engine.goal import Goal, GoalSubsystem, LineOriented	2✔
24	from pants.engine.internals.graph import resolve_dependencies	2✔
25	from pants.engine.rules import collect_rules, concurrently, goal_rule, implicitly, rule	2✔
26	from pants.engine.target import (	2✔
27	AllUnexpandedTargets,
28	AlwaysTraverseDeps,
29	Dependencies,
30	DependenciesRequest,
31	)
32	from pants.option.option_types import BoolOption, EnumOption	2✔
33	from pants.util.frozendict import FrozenDict	2✔
34	from pants.util.logging import LogLevel	2✔
35	from pants.util.ordered_set import FrozenOrderedSet	2✔
36
37	logger = logging.getLogger(__name__)	2✔
38
39
40	@dataclass(frozen=True)	2✔
41	class AddressToDependents:	2✔
42	mapping: FrozenDict[Address, FrozenOrderedSet[Address]]
43
44
45	class DependentsOutputFormat(Enum):	2✔
46	"""Output format for listing dependents.
47
48	text: List all dependents as a single list of targets in plain text.
49	json: List all dependents as a mapping `{target: [dependents]}`.
50	"""
51
52	text = "text"	2✔
53	json = "json"	2✔
54
55
56	@rule(desc="Map all targets to their dependents", level=LogLevel.DEBUG)	2✔
57	async def map_addresses_to_dependents(	2✔
58	all_targets: AllUnexpandedTargets,
59	incremental_cfg: IncrementalDependents,
60	) -> AddressToDependents:
61	"""Build a reverse dependency map (target -> set of its dependents).
62
63	When incremental mode is enabled via `--incremental-dependents-enabled`, the forward
64	dependency graph is persisted to disk. On subsequent runs, only targets whose source
65	files have changed need their dependencies re-resolved, dramatically reducing wall time
66	for large repos.
67	"""
NEW 68	if not incremental_cfg.enabled:	×
69	# Original behavior: resolve all dependencies from scratch.
NEW 70	dependencies_per_target = await concurrently(	×
71	resolve_dependencies(
72	DependenciesRequest(
73	tgt.get(Dependencies),
74	should_traverse_deps_predicate=AlwaysTraverseDeps(),
75	),
76	**implicitly(),
77	)
78	for tgt in all_targets
79	)
80
NEW 81	address_to_dependents = defaultdict(set)	×
NEW 82	for tgt, dependencies in zip(all_targets, dependencies_per_target):	×
NEW 83	for dependency in dependencies:	×
NEW 84	address_to_dependents[dependency].add(tgt.address)	×
NEW 85	return AddressToDependents(	×
86	FrozenDict(
87	{
88	addr: FrozenOrderedSet(dependents)
89	for addr, dependents in address_to_dependents.items()
90	}
91	)
92	)
93
94	# --- Incremental mode ---
NEW 95	start_time = time.time()	×
NEW 96	buildroot = get_buildroot()	×
NEW 97	cache_path = get_cache_path()	×
98
99	# Step 1: Load previous graph
NEW 100	previous = load_persisted_graph(cache_path, buildroot)	×
NEW 101	logger.warning(	×
102	"Incremental dep graph: loaded %d cached entries from %s",
103	len(previous),
104	cache_path,
105	)
106
107	# Step 2: Classify targets as cached or changed
NEW 108	changed_targets = []	×
NEW 109	cached_results: list[tuple[Address, CachedEntry]] = []	×
110
NEW 111	for tgt in all_targets:	×
NEW 112	spec = tgt.address.spec	×
NEW 113	fingerprint = compute_source_fingerprint(tgt.address, buildroot)	×
114
NEW 115	cached_entry = previous.get(spec)	×
NEW 116	if cached_entry is not None and cached_entry.fingerprint == fingerprint:	×
NEW 117	cached_results.append((tgt.address, cached_entry))	×
118	else:
NEW 119	changed_targets.append(tgt)	×
120
NEW 121	cache_hits = len(cached_results)	×
NEW 122	cache_misses = len(changed_targets)	×
NEW 123	logger.warning(	×
124	"Incremental dep graph: %d cached, %d changed (out of %d total targets)",
125	cache_hits,
126	cache_misses,
127	len(all_targets),
128	)
129
130	# Step 3: Resolve deps only for changed targets
NEW 131	if changed_targets:	×
NEW 132	fresh_deps_per_target = await concurrently(	×
133	resolve_dependencies(
134	DependenciesRequest(
135	tgt.get(Dependencies),
136	should_traverse_deps_predicate=AlwaysTraverseDeps(),
137	),
138	**implicitly(),
139	)
140	for tgt in changed_targets
141	)
142	else:
NEW 143	fresh_deps_per_target = []	×
144
145	# Step 4: Build the reverse dependency map from merged results
NEW 146	address_to_dependents: dict[Address, set[Address]] = defaultdict(set)	×
147
148	# Build a spec → Address lookup from all_targets for resolving cached specs
NEW 149	spec_to_address: dict[str, Address] = {tgt.address.spec: tgt.address for tgt in all_targets}	×
150
151	# Process cached results (deps stored as address spec strings)
NEW 152	for addr, entry in cached_results:	×
NEW 153	for dep_spec in entry.deps:	×
NEW 154	dep_addr = spec_to_address.get(dep_spec)	×
NEW 155	if dep_addr is not None:	×
NEW 156	address_to_dependents[dep_addr].add(addr)	×
157
158	# Process freshly resolved results
NEW 159	for tgt, deps in zip(changed_targets, fresh_deps_per_target):	×
NEW 160	for dep_addr in deps:	×
NEW 161	address_to_dependents[dep_addr].add(tgt.address)	×
162
163	# Step 5: Save the updated forward graph for next run
NEW 164	new_entries: dict[str, CachedEntry] = {}	×
165
166	# Carry forward cached entries
NEW 167	for addr, entry in cached_results:	×
NEW 168	new_entries[addr.spec] = entry	×
169
170	# Add fresh entries
NEW 171	for tgt, deps in zip(changed_targets, fresh_deps_per_target):	×
NEW 172	spec = tgt.address.spec	×
NEW 173	fingerprint = compute_source_fingerprint(tgt.address, buildroot)	×
NEW 174	new_entries[spec] = CachedEntry(	×
175	fingerprint=fingerprint,
176	deps=tuple(dep.spec for dep in deps),
177	)
178
NEW 179	save_persisted_graph(cache_path, buildroot, new_entries)	×
180
NEW 181	elapsed = time.time() - start_time	×
NEW 182	logger.warning(	×
183	"Incremental dep graph: completed in %.1fs (%d from cache, %d resolved fresh)",
184	elapsed,
185	cache_hits,
186	cache_misses,
187	)
188
UNCOV 189	return AddressToDependents(	×
190	FrozenDict(
191	{
192	addr: FrozenOrderedSet(dependents)
193	for addr, dependents in address_to_dependents.items()
194	}
195	)
196	)
197
198
199	@dataclass(frozen=True)	2✔
200	class DependentsRequest:	2✔
201	addresses: FrozenOrderedSet[Address]
202	transitive: bool
203	include_roots: bool
204
205	def __init__(	2✔
206	self, addresses: Iterable[Address], *, transitive: bool, include_roots: bool
207	) -> None:
UNCOV 208	object.__setattr__(self, "addresses", FrozenOrderedSet(addresses))	×
UNCOV 209	object.__setattr__(self, "transitive", transitive)	×
UNCOV 210	object.__setattr__(self, "include_roots", include_roots)	×
211
212
213	class Dependents(DeduplicatedCollection[Address]):	2✔
214	sort_input = True	2✔
215
216
217	@rule(level=LogLevel.DEBUG)	2✔
218	async def find_dependents(	2✔
219	request: DependentsRequest, address_to_dependents: AddressToDependents
220	) -> Dependents:
UNCOV 221	check = set(request.addresses)	×
UNCOV 222	known_dependents: set[Address] = set()	×
UNCOV 223	while True:	×
UNCOV 224	dependents = set(known_dependents)	×
UNCOV 225	for target in check:	×
UNCOV 226	target_dependents = address_to_dependents.mapping.get(target, FrozenOrderedSet())	×
UNCOV 227	dependents.update(target_dependents)	×
UNCOV 228	check = dependents - known_dependents	×
UNCOV 229	if not check or not request.transitive:	×
UNCOV 230	result = (	×
231	dependents \| set(request.addresses)
232	if request.include_roots
233	else dependents - set(request.addresses)
234	)
UNCOV 235	return Dependents(result)	×
UNCOV 236	known_dependents = dependents	×
237
238
239	class DependentsSubsystem(LineOriented, GoalSubsystem):	2✔
240	name = "dependents"	2✔
241	help = "List all targets that depend on any of the input files/targets."	2✔
242
243	transitive = BoolOption(	2✔
244	default=False,
245	help="List all transitive dependents. If unspecified, list direct dependents only.",
246	)
247	closed = BoolOption(	2✔
248	default=False,
249	help="Include the input targets in the output, along with the dependents.",
250	)
251	format = EnumOption(	2✔
252	default=DependentsOutputFormat.text,
253	help="Output format for listing dependents.",
254	)
255
256
257	class DependentsGoal(Goal):	2✔
258	subsystem_cls = DependentsSubsystem	2✔
259	environment_behavior = Goal.EnvironmentBehavior.LOCAL_ONLY	2✔
260
261
262	async def list_dependents_as_plain_text(	2✔
263	addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
264	) -> None:
265	"""Get dependents for given addresses and list them in the console as a single list."""
UNCOV 266	dependents = await find_dependents(	×
267	DependentsRequest(
268	addresses,
269	transitive=dependents_subsystem.transitive,
270	include_roots=dependents_subsystem.closed,
271	),
272	**implicitly(),
273	)
UNCOV 274	with dependents_subsystem.line_oriented(console) as print_stdout:	×
UNCOV 275	for address in dependents:	×
UNCOV 276	print_stdout(address.spec)	×
277
278
279	async def list_dependents_as_json(	2✔
280	addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
281	) -> None:
282	"""Get dependents for given addresses and list them in the console in JSON."""
UNCOV 283	dependents_group = await concurrently(	×
284	find_dependents(
285	DependentsRequest(
286	(address,),
287	transitive=dependents_subsystem.transitive,
288	include_roots=dependents_subsystem.closed,
289	),
290	**implicitly(),
291	)
292	for address in addresses
293	)
UNCOV 294	iterated_addresses = []	×
UNCOV 295	for dependents in dependents_group:	×
UNCOV 296	iterated_addresses.append(sorted([str(address) for address in dependents]))	×
UNCOV 297	mapping = dict(zip([str(address) for address in addresses], iterated_addresses))	×
UNCOV 298	output = json.dumps(mapping, indent=4)	×
UNCOV 299	with dependents_subsystem.line_oriented(console) as print_stdout:	×
UNCOV 300	print_stdout(output)	×
301
302
303	@goal_rule	2✔
304	async def dependents_goal(	2✔
305	specified_addresses: Addresses, dependents_subsystem: DependentsSubsystem, console: Console
306	) -> DependentsGoal:
UNCOV 307	if DependentsOutputFormat.text == dependents_subsystem.format:	×
UNCOV 308	await list_dependents_as_plain_text(	×
309	addresses=specified_addresses,
310	dependents_subsystem=dependents_subsystem,
311	console=console,
312	)
UNCOV 313	elif DependentsOutputFormat.json == dependents_subsystem.format:	×
UNCOV 314	await list_dependents_as_json(	×
315	addresses=specified_addresses,
316	dependents_subsystem=dependents_subsystem,
317	console=console,
318	)
UNCOV 319	return DependentsGoal(exit_code=0)	×
320
321
322	def rules():	2✔
323	return collect_rules()	2✔

pantsbuild / pants / 24103052497

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous