Why a step recomputed

Call explain_key(params) on a node to list every part of its cache key. Run it before and after a change: the part whose digest differs is the cause.

Explain one node

from dataclasses import dataclass

from pydantic import BaseModel

from adag import Config, InputParam, Step


class Number(BaseModel):
    value: int


@dataclass(frozen=True)
class Scale(Step[Number]):
    factor: int = 2
    workers: Config[int] = Config(4)

    def run(self, x: int, factor: int, workers: int) -> Number:
        return Number(value=x * factor)


node = Scale()(x=InputParam[int]("x"))
print(node.explain_key({"x": 3}))

Scale  key=9ba856cd8cc6  format=2
  identity  -        Scale         88caac9df9ad
  code      -        663765373931  3c5791a5408e
  field     factor   2             01e6dab02f90
  skipped   workers  Config(4)     -
  param     x        3             412a6e3e02d7

Each row is one component: kind, name, value, and a 12-hex digest of the bytes it adds to the key. A skipped row is not in the key and has no digest.

Kind	What it is
identity	the step class name, or its kind
code	a digest of the step class source
pinned	the class sets cache_key: it replaces identity, code and fields
field	a step field, by value
file	a FileSource field, by file content
nested_step	a field that holds another step, by that step's hash
upstream	an input wired to another node, by that node's key
param	an input wired to an InputParam, by the run's value or the default
deferred	an input wired to another source, such as a FileSource
raw	an input passed as a literal value
skipped	not in the key: a Config field, node_label, or a value the key cannot encode when strict_content_hash = False

The return value is a KeyExplanation. str() gives the table above, to_dict() gives JSON.

• key is None when a component blocks it, and blocked_by names that component. A FileSource field whose file is missing does this. A missing FileSource wired as an input does not: it still gets a key, and the run fails with FileNotFoundError.
• key_format is adag.hashing.KEY_FORMAT. A release that changes every key bumps it.
• A step that overrides content_hash (FanOutStep does) shows one identity row: adag cannot see its parts.

Find what changed

Pipeline.explain_keys(params) explains every node, upstream nodes first. Compare two graphs component by component:

"""Find which part of a cache key changed, and read the record adag saved with a result."""

import json
import tempfile
from dataclasses import dataclass
from pathlib import Path

from pydantic import BaseModel

from adag import InputParam, KeyExplanation, NullObserver, Pipeline, Step


class Number(BaseModel):
    value: int


@dataclass(frozen=True)
class Scale(Step[Number]):
    factor: int = 2

    def run(self, x: int, factor: int) -> Number:
        return Number(value=x * factor)


@dataclass(frozen=True)
class AddOne(Step[Number]):
    def run(self, n: Number) -> Number:
        return Number(value=n.value + 1)


def build(factor: int, cache: Path) -> Pipeline[Number]:
    x = InputParam[int]("x")
    return Pipeline(AddOne()(n=Scale(factor=factor)(x=x)), cache_dir=cache)


cache = Path(tempfile.mkdtemp())
params = {"x": 3}
old, new = build(2, cache), build(3, cache)

for before, after in zip(old.explain_keys(params), new.explain_keys(params), strict=True):
    for a, b in zip(before.components, after.components, strict=True):
        if a.digest != b.digest:
            print(f"{after.step}: {b.kind.value} {b.name} changed: {a.value} -> {b.value}")

old.execute(observer=NullObserver(), **params)
key = old.output.content_hash(params)
record = json.loads((cache / ".keys" / f"{key}.json").read_text())
saved = KeyExplanation.from_dict(record)
print(saved.step, "record has adag_version:", "adag_version" in record)
print("record equals explain_key:", saved == old.output.explain_key(params))

Scale: field factor changed: 2 -> 3
AddOne: upstream n changed: Scale cc7c4b4e33b2 -> Scale a926f5aa6596
AddOne record has adag_version: True
record equals explain_key: True

The first line that is not upstream is the cause. Each upstream line after it is a consequence.

Key records

After each save adag writes <cache_dir>/.keys/<key>.json: the node's explain_key output plus record_version, adag_version and saved_at. The end of the example above reads one back with KeyExplanation.from_dict.

• Size: under 1 KB for a small step. Each value is cut to 80 characters.
• A skipped component keeps its name and loses its value. A Config field can hold a client or a secret, and the record goes to disk or to a shared store.
• A failed write never fails the run. adag sends a step.key_record.failed event instead.
• With store=, the record is the store key .keys/<key>.json.
• adag does not read these records. Use them to compare the key a result was saved under with a new key.