Put the ChangeCalculator implementation next to TargetRootsCalculator

src/python/pants/bin/goal_runner.py

@@ -123,7 +123,8 @@ def _init_graph(self,
     target_roots = target_roots or TargetRootsCalculator.create(
       options=self._options,
       build_root=self._root_dir,
-      change_calculator=graph_helper.change_calculator
+      session=graph_helper.scheduler_session,
+      symbol_table=graph_helper.symbol_table
     )
     graph, address_mapper = graph_helper.create_build_graph(target_roots,
                                                             self._root_dir)

src/python/pants/init/engine_initializer.py

@@ -79,12 +79,10 @@ class LegacyGraphScheduler(datatype(['scheduler', 'symbol_table'])):
 
   def new_session(self):
     session = self.scheduler.new_session()
-    scm = get_scm()
-    change_calculator = EngineChangeCalculator(session, self.symbol_table, scm) if scm else None
-    return LegacyGraphSession(session, self.symbol_table, change_calculator)
+    return LegacyGraphSession(session, self.symbol_table)
 
 
-class LegacyGraphSession(datatype(['scheduler_session', 'symbol_table', 'change_calculator'])):
+class LegacyGraphSession(datatype(['scheduler_session', 'symbol_table'])):
   """A thin wrapper around a SchedulerSession configured with @rules for a symbol table."""
 
   def warm_product_graph(self, target_roots):

src/python/pants/init/target_roots_calculator.py

@@ -9,15 +9,87 @@
 
 from twitter.common.collections import OrderedSet
 
-from pants.base.build_environment import get_buildroot
+from pants.base.build_environment import get_buildroot, get_scm
 from pants.base.cmd_line_spec_parser import CmdLineSpecParser
-from pants.base.specs import SingleAddress
+from pants.base.specs import SingleAddress, DescendantAddresses, Specs
+from pants.build_graph.address import Address
+from pants.engine.legacy.graph import TransitiveHydratedTargets, target_types_from_symbol_table
+from pants.engine.legacy.source_mapper import EngineSourceMapper
+from pants.goal.workspace import ScmWorkspace
 from pants.base.target_roots import TargetRoots
 from pants.scm.subsystems.changed import ChangedRequest
 
 
 logger = logging.getLogger(__name__)
 
+class _DependentGraph(object):
+  """A graph for walking dependent addresses of TargetAdaptor objects.
+
+  This avoids/imitates constructing a v1 BuildGraph object, because that codepath results
+  in many references held in mutable global state (ie, memory leaks).
+
+  The long term goal is to deprecate the `changed` goal in favor of sufficiently good cache
+  hit rates, such that rather than running:
+
+    ./pants --changed-parent=master test
+
+  ...you would always be able to run:
+
+    ./pants test ::
+
+  ...and have it complete in a similar amount of time by hitting relevant caches.
+  """
+
+  @classmethod
+  def from_iterable(cls, target_types, adaptor_iter):
+    """Create a new DependentGraph from an iterable of TargetAdaptor subclasses."""
+    inst = cls(target_types)
+    for target_adaptor in adaptor_iter:
+      inst.inject_target(target_adaptor)
+    return inst
+
+  def __init__(self, target_types):
+    self._dependent_address_map = defaultdict(set)
+    self._target_types = target_types
+
+  def inject_target(self, target_adaptor):
+    """Inject a target, respecting all sources of dependencies."""
+    target_cls = self._target_types[target_adaptor.type_alias]
+
+    declared_deps = target_adaptor.dependencies
+    implicit_deps = (Address.parse(s)
+                     for s in target_cls.compute_dependency_specs(kwargs=target_adaptor.kwargs()))
+
+    for dep in itertools.chain(declared_deps, implicit_deps):
+      self._dependent_address_map[dep].add(target_adaptor.address)
+
+  def dependents_of_addresses(self, addresses):
+    """Given an iterable of addresses, yield all of those addresses dependents."""
+    seen = set(addresses)
+    for address in addresses:
+      for dependent_address in self._dependent_address_map[address]:
+        if dependent_address not in seen:
+          seen.add(dependent_address)
+          yield dependent_address
+
+  def transitive_dependents_of_addresses(self, addresses):
+    """Given an iterable of addresses, yield all of those addresses dependents, transitively."""
+    addresses_to_visit = set(addresses)
+    while 1:
+      dependents = set(self.dependents_of_addresses(addresses))
+      # If we've exhausted all dependencies or visited all remaining nodes, break.
+      if (not dependents) or dependents.issubset(addresses_to_visit):
+        break
+      addresses = dependents.difference(addresses_to_visit)
+      addresses_to_visit.update(dependents)
+
+    transitive_set = itertools.chain(
+      *(self._dependent_address_map[address] for address in addresses_to_visit)
+    )
+    for dep in transitive_set:
+      yield dep
+
+
 
 class InvalidSpecConstraint(Exception):
   """Raised when invalid constraints are given via target specs and arguments like --changed*."""
@@ -39,7 +111,7 @@ def parse_specs(cls, target_specs, build_root=None):
     return OrderedSet(spec_parser.parse_spec(spec_str) for spec_str in target_specs)
 
   @classmethod
-  def create(cls, options, build_root=None, change_calculator=None):
+  def create(cls, options, session, symbol_table, build_root=None):
     """
     :param Options options: An `Options` instance to use.
     :param string build_root: The build root.
@@ -55,6 +127,8 @@ def create(cls, options, build_root=None, change_calculator=None):
 
     logger.debug('spec_roots are: %s', spec_roots)
     logger.debug('changed_request is: %s', changed_request)
+    scm = get_scm()
+    change_calculator = EngineChangeCalculator(session, symbol_table, scm) if scm else None
 
     if change_calculator and changed_request.is_actionable():
       if spec_roots:
@@ -68,3 +142,70 @@ def create(cls, options, build_root=None, change_calculator=None):
       return TargetRoots(tuple(SingleAddress(a.spec_path, a.target_name) for a in changed_addresses))
 
     return TargetRoots(spec_roots)
+
+
+
+class EngineChangeCalculator(object):
+  """A ChangeCalculator variant that uses the v2 engine for source mapping."""
+
+  def __init__(self, scheduler, symbol_table, scm, workspace=None, changes_since=None,
+               diffspec=None):
+    """
+    :param scheduler: The `Scheduler` instance to use for computing file to target mappings.
+    :param symbol_table: The symbol table.
+    :param scm: The `Scm` instance to use for change determination.
+    """
+    self._scm = scm or get_scm()
+    self._scheduler = scheduler
+    self._symbol_table = symbol_table
+    self._mapper = EngineSourceMapper(self._scheduler)
+    self._workspace = workspace or ScmWorkspace(scm)
+    self._changes_since = changes_since
+    self._diffspec = diffspec
+
+  def changed_files(self, changes_since=None, diffspec=None):
+    """Determines the files changed according to SCM/workspace and options."""
+    diffspec = diffspec or self._diffspec
+    if diffspec:
+      return self._workspace.changes_in(diffspec)
+
+    changes_since = changes_since or self._changes_since or self._scm.current_rev_identifier()
+    return self._workspace.touched_files(changes_since)
+
+  def iter_changed_target_addresses(self, changed_request):
+    """Given a `ChangedRequest`, compute and yield all affected target addresses."""
+    changed_files = self.changed_files(changed_request.changes_since, changed_request.diffspec)
+    logger.debug('changed files: %s', changed_files)
+    if not changed_files:
+      return
+
+    changed_addresses = set(address
+                            for address
+                            in self._mapper.iter_target_addresses_for_sources(changed_files))
+    for address in changed_addresses:
+      yield address
+
+    if changed_request.include_dependees not in ('direct', 'transitive'):
+      return
+
+    # TODO: For dependee finding, we technically only need to parse all build files to collect target
+    # dependencies. But in order to fully validate the graph and account for the fact that deleted
+    # targets do not show up as changed roots, we use the `TransitiveHydratedTargets` product.
+    #   see https://github.com/pantsbuild/pants/issues/382
+    specs = (DescendantAddresses(''),)
+    adaptor_iter = (t.adaptor
+                    for targets in self._scheduler.product_request(TransitiveHydratedTargets,
+                                                                   [Specs(specs)])
+                    for t in targets.roots)
+    graph = _DependentGraph.from_iterable(target_types_from_symbol_table(self._symbol_table),
+                                          adaptor_iter)
+
+    if changed_request.include_dependees == 'direct':
+      for address in graph.dependents_of_addresses(changed_addresses):
+        yield address
+    elif changed_request.include_dependees == 'transitive':
+      for address in graph.transitive_dependents_of_addresses(changed_addresses):
+        yield address
+
+  def changed_target_addresses(self, changed_request):
+    return list(self.iter_changed_target_addresses(changed_request))