fix: force check all required outputs (#3341)

Hocnonsense · johanneskoester · coderabbitai[bot] · web-flow · commit 495a4e78a93f · 2025-03-11T19:30:39.000Z
This may fix #3036 This fix discards totaly change `future_output` to `created_output`, and directly check if all output the rule wanted are created. I'm somehow doubt if `future_output` used some elsewhere. Is it needed to add it back? ### QC  * [x] The PR contains test case <`tests/tests.py::test_checkpoints_many`> for the changes. * [x] the change does neither modify the language nor the behavior or functionalities of Snakemake.  ## Summary by CodeRabbit - **New Features** - Enhanced workflow execution with refined job ordering, checkpoint validation, file parsing, checksum extraction, and updated CLI help text. - **Documentation** - Updated examples and guidelines for input parsing and configuration validations, with typographical corrections for improved clarity. - **Tests** - Introduced new test cases for checkpoint and data format validation, and updated expected outputs and schema definitions. - **Chores** - Revised CI workflows with conditional execution and updated environment dependencies to streamline the build process.  --------- Co-authored-by: Johannes Köster <johannes.koester@uni-due.de> Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> Co-authored-by: Johannes Köster <johannes.koester@tu-dortmund.de>
diff --git a/snakemake/checkpoints.py b/snakemake/checkpoints.py
@@ -1,13 +1,15 @@
-from snakemake.common import async_run
+from typing import TYPE_CHECKING
 from snakemake.exceptions import IncompleteCheckpointException, WorkflowError
 from snakemake.io import checkpoint_target
 
+if TYPE_CHECKING:
+    from snakemake.rules import Rule
+
 
 class Checkpoints:
     """A namespace for checkpoints so that they can be accessed via dot notation."""
 
     def __init__(self):
-        self.future_output = None
         self.created_output = None
 
     def register(self, rule, fallback_name=None):
@@ -20,7 +22,7 @@ def register(self, rule, fallback_name=None):
 class Checkpoint:
     __slots__ = ["rule", "checkpoints"]
 
-    def __init__(self, rule, checkpoints):
+    def __init__(self, rule: "Rule", checkpoints: Checkpoints):
         self.rule = rule
         self.checkpoints = checkpoints
 
@@ -32,11 +34,9 @@ def get(self, **wildcards):
             )
 
         output, _ = self.rule.expand_output(wildcards)
-        if self.checkpoints.future_output is not None:
-            for iofile in output:
-                if iofile in self.checkpoints.future_output:
-                    break
-            else:
+        if self.checkpoints.created_output is not None:
+            may_not_created = set(output) - set(self.checkpoints.created_output)
+            if not may_not_created:
                 return CheckpointJob(self.rule, output)
 
         raise IncompleteCheckpointException(self.rule, checkpoint_target(output[0]))
@@ -45,6 +45,6 @@ def get(self, **wildcards):
 class CheckpointJob:
     __slots__ = ["rule", "output"]
 
-    def __init__(self, rule, output):
+    def __init__(self, rule: "Rule", output):
         self.output = output
         self.rule = rule
diff --git a/snakemake/dag.py b/snakemake/dag.py
@@ -267,17 +267,16 @@ def checkpoint_jobs(self):
         return self._checkpoint_jobs
 
     @property
-    def finished_checkpoint_jobs(self):
-        for job in self.finished_jobs:
-            if job.is_checkpoint:
+    def finished_and_not_needrun_checkpoint_jobs(self):
+        for job in self.jobs:
+            if job.is_checkpoint and (self.finished(job) or not self.needrun(job)):
                 yield job
 
     def update_checkpoint_outputs(self):
-        workflow.checkpoints.future_output = set(
-            f for job in self.checkpoint_jobs for f in job.output
-        )
         workflow.checkpoints.created_output = set(
-            f for job in self.finished_checkpoint_jobs for f in job.output
+            f
+            for job in self.finished_and_not_needrun_checkpoint_jobs
+            for f in job.output
         )
 
     def update_jobids(self):
diff --git a/tests/test_checkpoints_many/Snakefile b/tests/test_checkpoints_many/Snakefile
@@ -0,0 +1,73 @@
+import glob
+import random
+from pathlib import Path
+
+ALL_SAMPLES = ["s1"]
+
+
+rule all:
+    input:
+        expand("collect/{sample}/all_done.txt", sample=ALL_SAMPLES),
+
+
+rule before:
+    output:
+        ALL_SAMPLES,
+    run:
+        for sample in ALL_SAMPLES:
+            Path(sample).touch()
+
+
+checkpoint first:
+    input:
+        expand("{sample}", sample=ALL_SAMPLES),
+    output:
+        directory("first/{sample}"),
+    run:
+        for i in range(1, 5):
+            Path(f"{output[0]}/{i}").mkdir(parents=True, exist_ok=True)
+            Path(f"{output[0]}/{i}/test.txt").touch()
+
+
+checkpoint second:
+    input:
+        "first/{sample}/{i}/test.txt",
+    output:
+        directory("second/{sample}/{i}"),
+    run:
+        for j in range(6, 10):
+            Path(f"{output[0]}/{j}").mkdir(parents=True, exist_ok=True)
+            Path(f"{output[0]}/{j}/test2.txt").touch()
+
+
+rule copy:
+    input:
+        "second/{sample}/{i}/{j}/test2.txt",
+    output:
+        touch("copy/{sample}/{i}/{j}/test2.txt"),
+
+
+def aggregate(wildcards):
+
+    outputs_i = glob.glob(f"{checkpoints.first.get(**wildcards).output}/*/")
+
+    outputs_i = [output.split("/")[-2] for output in outputs_i]
+
+    split_files = []
+    for i in outputs_i:
+        s2out = checkpoints.second.get(**wildcards, i=i).output[0]
+        assert Path(s2out).exists()
+        output_j = glob.glob(f"{s2out}/*/")
+        outputs_j = [output.split("/")[-2] for output in output_j]
+        for j in outputs_j:
+            split_files.extend(
+                expand(f"copy/{{sample}}/{i}/{j}/test2.txt", sample=wildcards.sample)
+            )
+    return split_files
+
+
+rule collect:
+    input:
+        aggregate,
+    output:
+        touch("collect/{sample}/all_done.txt"),
diff --git a/tests/test_checkpoints_many/expected-results/collect/s1/all_done.txt b/tests/test_checkpoints_many/expected-results/collect/s1/all_done.txt
diff --git a/tests/test_issue1092/Snakefile b/tests/test_issue1092/Snakefile
@@ -29,8 +29,8 @@ rule aggregate:
         aggregate_input
     output:
         "aggregated.txt"
-    #shell:
-    #    "cat {input} > {output}"
+    # shell:
+    #     "cat {input} > {output}"
     run:
         shell("cat {input} > {output}")
 
diff --git a/tests/tests.py b/tests/tests.py
@@ -1167,6 +1167,10 @@ def test_checkpoints_dir():
     run(dpath("test_checkpoints_dir"))
 
 
+def test_checkpoints_many():
+    run(dpath("test_checkpoints_many"))
+
+
 def test_issue1092():
     run(dpath("test_issue1092"))
 
