Fix left-plus graph operator to take ALL ancestors (#293)

integration_tests/features/flow_run.feature

@@ -23,6 +23,24 @@ Feature: `flow run` command
     And the following scripts are ran:
       | agent_wait_time.before.py |
 
+  Scenario: fal flow run command with complex selectors
+    Given the project 001_flow_run_with_selectors
+    When the data is seeded
+
+    When the following command is invoked:
+      """
+      fal flow run --profiles-dir $profilesDir --project-dir $baseDir --select +intermediate_model_3 --threads 1
+      """
+    Then the following models are calculated:
+      | agent_wait_time | intermediate_model_1 | intermediate_model_2 | intermediate_model_3 |
+
+    When the following command is invoked:
+      """
+      fal flow run --profiles-dir $profilesDir --project-dir $baseDir --select intermediate_model_1+ --threads 1
+      """
+    Then the following models are calculated:
+      | intermediate_model_1 | intermediate_model_2 | intermediate_model_3 |
+
   Scenario: fal flow run command with selectors
     Given the project 001_flow_run_with_selectors
     When the data is seeded

src/fal/cli/selectors.py

@@ -58,7 +58,12 @@ def create_plan_from_graph(cls, parsed, nodeGraph: NodeGraph, fal_dbt: FalDbt):
         return cls(list(set(ids_to_execute)), fal_dbt.project_name)
 
 
-def _filter_node_ids(unique_ids, fal_dbt, selected_nodes, nodeGraph) -> List[str]:
+def _filter_node_ids(
+    unique_ids: List[str],
+    fal_dbt: FalDbt,
+    selected_nodes: List[str],
+    nodeGraph: NodeGraph,
+) -> List[str]:
     """Filter list of unique_ids according to a selector."""
     output = []
     selector_plans = list(
@@ -74,7 +79,7 @@ def _filter_node_ids(unique_ids, fal_dbt, selected_nodes, nodeGraph) -> List[str
                 children = list(nodeGraph.get_descendants(id))
                 output.extend(children)
             if selector_plan.parents:
-                parents = list(nodeGraph.get_predecessors(id))
+                parents = list(nodeGraph.get_ancestors(id))
                 output.extend(parents)
     return output
 

src/fal/node_graph.py

@@ -132,12 +132,18 @@ def __init__(self, graph: nx.DiGraph, node_lookup: Dict[str, FalFlowNode]):
         self.graph = graph
         self.node_lookup = node_lookup
 
+    def get_successors(self, id: str) -> List[str]:
+        return list(self.graph.successors(id))
+
     def get_descendants(self, id: str) -> List[str]:
         return list(nx.descendants(self.graph, id))
 
     def get_predecessors(self, id: str) -> List[str]:
         return list(self.graph.predecessors(id))
 
+    def get_ancestors(self, id: str) -> List[str]:
+        return list(nx.ancestors(self.graph, id))
+
     def get_node(self, id: str) -> FalFlowNode | None:
         return self.node_lookup.get(id)