ENH(PLAN) hide Recompute_TILL from user; +1.6TC; ...

+ enh: recompute_TILL := Outputs.
+ refact: pass a copy-graph in recompute,
    for client-code to inspect it if need.
+ fix: empties TC ok, now parametrixzed.

Author: ankostis

graphtik/base.py

@@ -784,7 +784,6 @@ def compute(
         named_inputs,
         outputs=None,
         recompute_from=None,
-        recompute_till=None,
         *kw,
     ):
         """
@@ -801,9 +800,6 @@ def compute(
         :param recompute_from:
             recompute all downstream from those dependencies,
             see :meth:`.Pipeline.compute()`.
-        :param recompute_till:
-            (UNSTABLE) recompute all upstream from those dependencies,
-            see :meth:`.Pipeline.compute()`.
 
         :returns list:
             Should return a list values representing

graphtik/pipeline.py

@@ -322,7 +322,6 @@ def compile(
         inputs=None,
         outputs=UNSET,
         recompute_from=None,
-        recompute_till=None,
         *,
         predicate: "NodePredicate" = UNSET,
     ) -> "ExecutionPlan":
@@ -337,8 +336,6 @@ def compile(
             If not given, those set by a previous call to :meth:`withset()` or cstor are used.
         :param recompute_from:
             Described in :meth:`.Pipeline.compute()`.
-        :param recompute_till:
-            (UNSTABLE) Described in :meth:`.Pipeline.compute()`.
         :param predicate:
             Will be stored and applied on the next :meth:`compute()` or :meth:`compile()`.
             If not given, those set by a previous call to :meth:`withset()` or cstor are used.
@@ -361,16 +358,13 @@ def compile(
         if predicate == UNSET:
             predicate = self.predicate
 
-        return self.net.compile(
-            inputs, outputs, recompute_from, recompute_till, predicate=predicate
-        )
+        return self.net.compile(inputs, outputs, recompute_from, predicate=predicate)
 
     def compute(
         self,
         named_inputs: Mapping = None,
         outputs: Items = UNSET,
         recompute_from: Items = None,
-        recompute_till: Items = None,
         *,
         predicate: "NodePredicate" = UNSET,
         pre_callback=None,
@@ -397,19 +391,13 @@ def compute(
             *strictly downstream (excluding themselves)* from the dependencies
             listed here, as missing from `named_inputs`.
 
-            * If also `recompute_till` is given, traversing downstream stops
-              when arriving in any dependency contained in that list.
+            * Traversing downstream stops when arriving at any dep in `outputs`.
             * Any dependencies here unreachable downstreams from values in `named_inputs`
               are ignored, but logged.
             * Any dependencies here unreachable upstreams from `outputs` (if given)
               are ignored, but logged.
             * Results may differ even if graph is unchanged, in the presence
               of :term:`overwrite`\\s.
-        :param recompute_till:
-            (UNSTABLE) Refresh all computations arriving to these (string or list) dependencies.
-            In effect, it clears all values in `named_inputs` reachable UPstreams.
-
-            * See bullet notes in `recompute_from`, above.
         :param predicate:
             filter-out nodes before compiling
             If not given, those set by a previous call to :meth:`withset()` or cstor are used.
@@ -471,7 +459,6 @@ def compute(
                 named_inputs.keys(),
                 outputs,
                 recompute_from,
-                recompute_till,
                 predicate=predicate,
             )
 

graphtik/planning.py

@@ -17,6 +17,7 @@
     Mapping,
     NamedTuple,
     Optional,
+    Sequence,
     Tuple,
     Union,
 )
@@ -239,18 +240,23 @@ def _topo_sort_nodes(dag) -> iset:
         raise nx.NetworkXUnfeasible(msg).with_traceback(tb)
 
 
-def recompute_inputs(
-    graph, inputs, recompute_from, recompute_till
+def inputs_for_recompute(
+    graph,
+    inputs: Sequence[str],
+    recompute_from: Sequence[str],
+    recompute_till: Sequence[str] = None,
 ) -> Tuple[iset, iset]:
     """
     Clears the inputs between `recompute_from >--<= recompute_till` to clear.
 
+    :param graph:
+        MODIFIED, at most 2 helper nodes inserted
     :param inputs:
-        None or a sequence
+        a sequence
     :param recompute_from:
-        None or a sequence
+        None or a sequence, including any out-of-graph deps (logged))
     :param recompute_till:
-        (UNSTABLE) None or a sequence
+        (optional) a sequence, only in-graph deps.
 
     :return:
         a 2-tuple with the reduced `inputs` by the dependencies that must
@@ -260,51 +266,40 @@ def recompute_inputs(
 
         strict-descendants(recompute_from) & ancestors(recompute_till)
 
-    FIXME: Should recompute() while travesing unsatisfied?  Is `till` relevant??
+    FIXME: merge recompute() with travesing unsatisfied (see ``test_recompute_NEEDS_FIX``)
+    bc it clears inputs of unsatisfied ops (cannot be replaced later)
     """
-    start, stop = "_TMP.RECOMPUTE_FROM", "_TMP.RECOMPUTE_TILL"
-    graph = graph.copy()
-
-    datanodes = iset(yield_datanodes(graph.nodes))
-    downstreams_strict = datanodes
-    if recompute_from is not None:
-        recompute_from = iset(recompute_from)  # traversed in logs
-        bad = recompute_from - datanodes
-        if bad:
-            log.info(
-                "... ignoring unknown `recompute_from` dependencies: %s", list(bad)
-            )
-            recompute_from = recompute_from & datanodes
-        if recompute_from:
-            graph.add_edges_from((start, i) for i in recompute_from)
+    START, STOP = "_TMP.RECOMPUTE_FROM", "_TMP.RECOMPUTE_TILL"
 
-            downstreams_strict = (
-                iset(yield_datanodes(nx.descendants(graph, start))) - recompute_from
-            )
+    deps = set(yield_datanodes(graph.nodes))
+    recompute_from = iset(recompute_from)  # traversed in logs
+    inputs = iset(inputs)  # returned
+    bad = recompute_from - deps
+    if bad:
+        log.info("... ignoring unknown `recompute_from` dependencies: %s", list(bad))
+        recompute_from = recompute_from & deps  # avoid sideffect in `recompute_from`
+    assert recompute_from, f"Given unknown-only `recompute_from` {locals()}"
 
-    if recompute_till is None:
-        upstreams = datanodes
-    else:
-        recompute_till = iset(recompute_till)  # traversed in logs
-        bad = recompute_till - datanodes
-        if bad:
-            log.info(
-                "... ignoring unknown `recompute_till` dependencies: %s", list(bad)
-            )
-            recompute_till = recompute_till & datanodes
-        graph.add_edges_from((i, stop) for i in recompute_till)  # edge reversed!
+    graph.add_edges_from((START, i) for i in recompute_from)
+
+    # strictly-downstreams from START
+    between_deps = iset(nx.descendants(graph, START)) & deps - recompute_from
+
+    if recompute_till:
+        graph.add_edges_from((i, STOP) for i in recompute_till)  # edge reversed!
 
-        upstreams = iset(yield_datanodes(nx.ancestors(graph, stop)))
+        # upstreams from STOP
+        upstreams = set(nx.ancestors(graph, STOP)) & deps
+        between_deps &= upstreams
 
-    between = downstreams_strict & upstreams
-    recomputes = [i for i in inputs if i in between]
-    new_inputs = iset(inputs) - between
+    recomputes = between_deps & inputs
+    new_inputs = iset(inputs) - recomputes
 
     if log.isEnabledFor(logging.DEBUG):
         log.debug(
             "... recompute x%i data%s means deleting x%i inputs%s, to arrive from x%i %s -> x%i %s.",
-            len(between),
-            list(between),
+            len(between_deps),
+            list(between_deps),
             len(recomputes),
             list(recomputes),
             len(inputs),
@@ -866,7 +861,6 @@ def compile(
         inputs: Items = None,
         outputs: Items = None,
         recompute_from=None,
-        recompute_till=None,
         *,
         predicate=None,
     ) -> "ExecutionPlan":
@@ -886,8 +880,6 @@ def compile(
             If string, it is converted to a single-element collection.
         :param recompute_from:
             Described in :meth:`.Pipeline.compute()`.
-        :param recompute_till:
-            (UNSTABLE) Described in :meth:`.Pipeline.compute()`.
         :param predicate:
             the :term:`node predicate` is a 2-argument callable(op, node-data)
             that should return true for nodes to include; if None, all nodes included.
@@ -909,15 +901,15 @@ def compile(
 
         ok = False
         try:
-            ## Make a stable cache-key.
+            ## Make a stable cache-key,
+            #  ignoring out-of-graph nodes (2nd results).
             #
             inputs, k1 = self._deps_tuplized(inputs, "inputs")
             outputs, k2 = self._deps_tuplized(outputs, "outputs")
             recompute_from, k3 = self._deps_tuplized(recompute_from, "recompute_from")
-            recompute_till, k4 = self._deps_tuplized(recompute_till, "recompute_till")
             if not predicate:
                 predicate = None
-            cache_key = (k1, k2, k3, k4, predicate, is_skip_evictions())
+            cache_key = (k1, k2, k3, predicate, is_skip_evictions())
 
             ## Build (or retrieve from cache) execution plan
             #  for the given dep-lists (excluding any unknown node-names).
@@ -926,9 +918,9 @@ def compile(
                 log.debug("... compile cache-hit key: %s", cache_key)
                 plan = self._cached_plans[cache_key]
             else:
-                if recompute_from is not None or recompute_till is not None:
-                    inputs, recomputes = recompute_inputs(
-                        self.graph, inputs, recompute_from, recompute_till
+                if recompute_from:
+                    inputs, recomputes = inputs_for_recompute(
+                        self.graph.copy(), inputs, recompute_from, k2
                     )
 
                 _prune_results = self._prune_graph(inputs, outputs, predicate)

test/test_graphtik.py

@@ -21,7 +21,7 @@
     vararg,
     varargs,
 )
-from graphtik.base import IncompleteExecutionError
+from graphtik.base import IncompleteExecutionError, UNSET
 from graphtik.config import debug_enabled, evictions_skipped, operations_endured
 
 from .helpers import addall, exe_params
@@ -1014,17 +1014,26 @@ def test_rerun_resched(quarantine_pipeline):
     assert exe_ops(sol) == ["get_out_or_stay_home", "exercise"]
 
 
-@pytest.mark.xfail(reason="Badly-specified `recompute` empties AND algorithm!")
-def test_recompute_empties(samplenet):
+@pytest.fixture
+def recompute_sol(samplenet):
     pipe = compose(..., samplenet, excludes="sum_op1")
     sol = pipe.compute({"c": 3, "d": 4}, recompute_from=())
     assert sol == {"c": 3, "d": 4, "sum2": 7, "sum3": 10}
     assert exe_ops(sol) == ["sum_op2", "sum_op3"]
+
+    return pipe, sol
+
+
+@pytest.mark.parametrize("outs", [None, UNSET, ()])
+@pytest.mark.parametrize("recomputes", [None, ()])
+def test_recompute_empties(recompute_sol, outs, recomputes):
+    pipe, sol = recompute_sol
     exp = dict(sol)
 
-    sol = pipe.compute(sol, recompute_from=())
-    assert sol == exp
-    assert exe_ops(sol) == ["sum_op2", "sum_op3"]
+    with pytest.raises(ValueError, match="^Unsolvable"):
+        pipe.compute(sol, outputs=outs, recompute_from=recomputes)
+    with pytest.raises(ValueError, match="^Unsolvable"):
+        pipe.compute({}, outputs=outs, recompute_from=recomputes)
 
 
 @pytest.mark.parametrize(
@@ -1131,6 +1140,35 @@ def test_recompute_resched_false(quarantine_pipeline, recompute, exp_ops):
     assert exe_ops(sol) == exp_ops
 
 
+def test_recompute_till():
+    def by2(n):
+        return 2 * n
+
+    pipe = compose(
+        ...,
+        operation(by2, "f0", "a0", "a1"),
+        operation(by2, "f1", "a1", "a2"),
+        operation(by2, "f2", "a2", "a3"),
+        operation(by2, "f3", "a3", "a4"),
+    )
+    sol = pipe(a0=1)
+    assert exe_ops(sol) == ["f0", "f1", "f2", "f3"]
+    assert sol == {"a0": 1, "a1": 2, "a2": 4, "a3": 8, "a4": 16}
+
+    inp = dict(sol)
+    inp["a1"] = 3
+
+    sol = pipe.compute(inp, outputs="a3", recompute_from="a1")
+    assert exe_ops(sol) == ["f1", "f2"]
+    assert sol == {"a3": 12}
+
+    with evictions_skipped(True):
+        sol = pipe.compute(inp, outputs="a3", recompute_from="a1")
+        sol.plot("t.pdf")
+        assert exe_ops(sol) == ["f1", "f2"]
+        assert sol == {"a0": 1, "a1": 3, "a2": 6, "a3": 12, "a4": 16}
+
+
 def test_recompute_NEEDS_FIX():
     pipe = compose(
         ...,