Implementation:Iterative Dvc Stage Changed
| Knowledge Sources | |
|---|---|
| Domains | Pipeline_Management, Change_Detection |
| Last Updated | 2026-02-10 00:00 GMT |
Overview
Concrete tool for detecting whether a pipeline stage needs re-execution by performing multi-level hash comparison across stage definition, dependencies, and outputs, provided by the DVC library.
Description
The Stage.changed() method in DVC is the primary entry point for freshness detection. It orchestrates three sub-checks: changed_stage() compares the stage command against its lockfile-recorded value (for PipelineStage, this uses the cmd_changed flag set during lockfile loading; for single stages, it compares MD5 hashes); changed_deps() iterates over all dependencies and calls each dependency's status() method to detect modifications; and changed_outs() does the same for all outputs.
The method uses short-circuit boolean evaluation via Python's or operator: self.changed_stage() or self.changed_deps(...) or self.changed_outs(...). This ensures the cheapest check runs first and more expensive checks are skipped if an earlier check already detected a change.
Special handling exists for several edge cases: frozen stages always return False for dependency changes; callback stages (no deps/outs) and always_changed stages always return True for dependency changes; and the allow_missing parameter combined with upstream stage information enables tolerant change detection where missing dependencies are acceptable if upstream stages have not changed.
The method is decorated with @rwlocked(read=["deps", "outs"]) to ensure thread-safe read access to the stage's dependencies and outputs during concurrent operations.
Usage
Use Stage.changed() when you need to:
- Determine if a specific stage needs re-execution during pipeline reproduction.
- Implement incremental build logic that skips unchanged stages.
- Report stage status to users (via dvc status).
- Decide whether to invoke the run-cache or proceed with full execution.
Code Reference
Source Location
- Repository: DVC
- File:
dvc/stage/__init__.py - Lines: L360-373 (changed), L301-310 (changed_deps), L337-352 (changed_outs), L354-358 (changed_stage)
Signature
class Stage:
@rwlocked(read=["deps", "outs"])
def changed(
self, allow_missing: bool = False, upstream: Optional[list] = None
) -> bool:
...
def changed_deps(
self, allow_missing: bool = False, upstream: Optional[list] = None
) -> bool:
...
@rwlocked(read=["outs"])
def changed_outs(self, allow_missing: bool = False) -> bool:
...
def changed_stage(self) -> bool:
...
class PipelineStage(Stage):
def changed_stage(self) -> bool:
"""Returns self.cmd_changed (set during lockfile loading)."""
...
Import
from dvc.stage import Stage, PipelineStage
I/O Contract
Inputs
| Name | Type | Required | Description |
|---|---|---|---|
| self | Stage | Yes | The stage instance to check for changes (contains deps, outs, cmd, md5) |
| allow_missing | bool | No | If True, tolerate deleted dependencies and outputs (useful during partial reproduction) |
| upstream | Optional[list[Stage]] | No | List of upstream stages that have been identified as changed; used to distinguish truly missing deps from deps that will be recreated |
Outputs
| Name | Type | Description |
|---|---|---|
| is_changed | bool | True if the stage needs re-execution (any of the three change levels detected a difference); False if the stage is fresh |
Usage Examples
Basic Usage
from dvc.repo import Repo
repo = Repo(".")
# Check if a specific stage has changed
for stage in repo.index.stages:
if stage.changed():
print(f"Stage '{stage.addressing}' needs re-execution")
else:
print(f"Stage '{stage.addressing}' is up to date")
Detailed Change Inspection
# Inspect which specific level detected the change
stage = repo.stage.collect("train")[0]
if stage.changed_stage():
print("Stage command/definition has changed")
if stage.changed_deps():
print("One or more dependencies have changed")
if stage.changed_outs():
print("One or more outputs have changed")
# Use allow_missing for partial pipeline scenarios
if stage.changed(allow_missing=True, upstream=changed_upstream_stages):
print("Stage changed even allowing for missing files")
Status Reporting
# Get detailed status information (used by `dvc status`)
status = stage.status()
# Returns dict like: {"train": ["changed command", {"changed deps": {...}}]}