[engine] Limit matches for FilesystemNode to only cases where lhs/rhs…

… match

Limit matches for FilesystemNode to only cases where lhs/rhs match; in other cases, we will recurse.

This simplifies FilesystemNode, and ensures that FilesystemNodes will only be created for precise types and fixes [pantsbuild#3117](pantsbuild#3117).

Testing Done:
https://travis-ci.org/pantsbuild/pants/builds/122859482

Bugs closed: 3117, 3185

Reviewed at https://rbcommons.com/s/twitter/r/3688/

src/python/pants/engine/exp/nodes.py

@@ -375,6 +375,11 @@ class FilesystemNode(datatype('FilesystemNode', ['subject', 'product', 'variants
   def is_filesystem_product(cls, product):
     return product in cls._FS_PRODUCT_TYPES
 
+  @classmethod
+  def is_filesystem_pair(cls, subject_type, product):
+    """True if the given subject type and product type should be computed using a FileystemNode."""
+    return subject_type is cls._FS_PRODUCT_TYPES.get(product, None)
+
   @property
   def is_cacheable(self):
     """Native node should not be cached."""
@@ -387,33 +392,20 @@ def _input_node(self):
     return SelectNode(self.subject, self._input_type(), self.variants, None)
 
   def step(self, dependency_states, step_context):
-    # Request the relevant input product for the output product.
-    input_node = self._input_node()
-    input_state = dependency_states.get(input_node, None)
-    if input_state is None or type(input_state) == Waiting:
-      return Waiting([input_node])
-    elif type(input_state) == Throw:
-      return input_state
-    elif type(input_state) == Noop:
-      return Noop('Could not compute {} in order to make filesystem request.'.format(input_node))
-    elif type(input_state) != Return:
-      State.raise_unrecognized(input_state)
-
-    # If an input product was available, execute.
-    input_value = input_state.value
     try:
       # TODO: https://github.com/pantsbuild/pants/issues/3121
-      assert not os.path.islink(input_value.path), (
-        'cannot perform native filesystem operations on a symlink!')
+      assert not os.path.islink(self.subject.path), (
+          'cannot perform native filesystem operations on symlink!: {}'.format(self.subject.path))
 
       if self.product is Paths:
-        return Return(path_exists(step_context.project_tree, input_value))
+        return Return(path_exists(step_context.project_tree, self.subject))
       elif self.product is FileContent:
-        return Return(file_content(step_context.project_tree, input_value))
+        return Return(file_content(step_context.project_tree, self.subject))
       elif self.product is DirectoryListing:
-        return Return(list_directory(step_context.project_tree, input_value))
+        return Return(list_directory(step_context.project_tree, self.subject))
       else:
-        return Throw('Mismatched input value {} for {}'.format(input_value, self))
+        # This would be caused by a mismatch between _FS_PRODUCT_TYPES and the above switch.
+        raise ValueError('Mismatched input value {} for {}'.format(self.subject, self))
     except Exception as e:
       return Throw(e)
 

src/python/pants/engine/exp/scheduler.py

@@ -189,8 +189,6 @@ def _generate_fsnode_subjects(self, filenames):
     for file_path, parent_dir_path in file_paths:
       yield Path(file_path)
       yield Path(parent_dir_path)  # Invalidate the parent dirs DirectoryListing.
-      # TODO: See https://github.com/pantsbuild/pants/issues/3117.
-      yield DescendantAddresses(parent_dir_path)
 
   def invalidate_files(self, filenames):
     """Given a set of changed filenames, invalidate all related FilesystemNodes in the graph."""
@@ -350,13 +348,13 @@ def __init__(self, tasks):
     self._tasks = tasks
 
   def gen_nodes(self, subject, product, variants):
-    # Native filesystem operations.
-    if FilesystemNode.is_filesystem_product(product):
+    if FilesystemNode.is_filesystem_pair(type(subject), product):
+      # Native filesystem operations.
       yield FilesystemNode(subject, product, variants)
-
-    # Tasks.
-    for task, anded_clause in self._tasks[product]:
-      yield TaskNode(subject, product, variants, task, anded_clause)
+    else:
+      # Tasks.
+      for task, anded_clause in self._tasks[product]:
+        yield TaskNode(subject, product, variants, task, anded_clause)
 
   def select_node(self, selector, subject, variants):
     """Constructs a Node for the given Selector and the given Subject/Variants.