Annotated step outputs

Packages that build on pipeflow may want to know where a result came from: which step produced it, which partition it belongs to, and when it was computed. A cache, for example, can only tell two runs apart if the results carry that information.

Without further help, every step function would have to add it itself, or be wrapped in a function that does so. The generic pip_output_value() is a single hook for this. You write one method for your class, and the pipeline applies it whenever a step returns.

An annotated output

Let’s use a class that carries the step it came from and the time of the run.

library(pipeflow)

stamped <- function(x) structure(x, class = "stamped")

print.stamped <- function(x, ...) {
    cat("<stamped>", unclass(x), "\n")
    cat("  step:", attr(x, "step"), "\n")
    cat("  key :", attr(x, "key"), "\n")
    invisible(x)
}

A step marks its result as stamped, but it doesn’t know its own name or the time of the run.

p <- pip_new("report") |>
    pip_add("load", \(n = 3) stamped(seq_len(n))) |>
    pip_add("total", \(x = ~load) stamped(sum(x)))

pip_run(p, lgr = NULL)
p[["total", "out"]]
# <stamped> 6 
#   step: 
#   key :

Nothing was filled in, because no method is registered yet.

Write and register a method

pip_output_value() is an S3 generic, which has a default method that returns the output unchanged. For our class we add a method that fills in the context.

pip_output_value.stamped <- function(x, ctx = list(), ...) {
    attr(x, "step") <- ctx[["step"]]
    attr(x, "key") <- ctx[["key"]]
    attr(x, "time") <- ctx[["time"]]
    x
}

Inside a script, register the method with registerS3method():

registerS3method("pip_output_value", "stamped", pip_output_value.stamped)

In a package, register it in the NAMESPACE file instead, either directly with S3method(pipeflow::pip_output_value, stamped) or, if you use roxygen, by tagging the method with @exportS3Method pipeflow::pip_output_value. Methods are meant for your own classes, not for base classes such as numeric or character.

Now every output is stamped when its step returns.

pip_run(p, force = TRUE, lgr = NULL)
p[["load", "out"]]
# <stamped> 1 2 3 
#   step: load 
#   key :
p[["total", "out"]]
# <stamped> 6 
#   step: total 
#   key :

The time stamp is the one the pipeline stores for the step, so it matches what the step table shows.

identical(
    as.numeric(attr(p[["load", "out"]], "time")),
    as.numeric(p[["load", "time"]])
)
# [1] TRUE

Where the method is applied

The generic is applied once to the output of a step, right after the step function returned, and its result is what is stored as the output. A downstream step therefore receives the annotated value.

p2 <- pip_new("downstream") |>
    pip_add("make", \(v = 1) stamped(v)) |>
    pip_add("inspect", \(x = ~make) attr(x, "step"))

pip_run(p2, lgr = NULL)
p2[["inspect", "out"]]
# [1] "make"

A partitioned output is the one exception: there the generic is applied once per partition, and each partition gets its key in ctx$key. This holds both for a step with exec = "split" and for a step that was mapped over a partitioned input.

p3 <- pip_new("partitions") |>
    pip_add(
        "split",
        \(x = 1) list(low = stamped(1), high = stamped(2)),
        exec = "split"
    ) |>
    pip_add("double", \(x = ~split) stamped(unclass(x) * 2))

pip_run(p3, lgr = NULL)
p3[["double", "out"]][["high"]]
# <stamped> 4 
#   step: double 
#   key : high

Two partitions of one step share the step name and the time stamp, so ctx$key is what tells them apart.

The generic is not applied to

Annotating the elements of an output

Since the generic sees the output as a whole, a plain list is passed to a method as it is, and its elements are not visited. If you want to annotate the elements, write a method that walks the output itself and applies the generic to the elements.

bundle <- function(...) structure(list(...), class = "bundle")

pip_output_value.bundle <- function(x, ctx = list(), ...) {
    x[] <- lapply(x, FUN = pip_output_value, ctx = ctx)
    x
}
registerS3method("pip_output_value", "bundle", pip_output_value.bundle)

p4 <- pip_new("bundle") |>
    pip_add("make", \(v = 1) bundle(a = stamped(v), b = stamped(v + 1)))

pip_run(p4, lgr = NULL)
p4[["make", "out"]][["b"]]
# <stamped> 2 
#   step: make 
#   key :

Errors in a method

A method runs as part of the step call. An error in a method is therefore an error of the step, handled like any other step error as described in the vignette on handling failures. For a partitioned output, the error names the partition the method failed for, just as for a step function that fails for one key.

checked <- function(x) structure(x, class = "checked")

pip_output_value.checked <- function(x, ctx = list(), ...) {
    if (length(x) == 0L) {
        stop("empty output of step '", ctx[["step"]], "'")
    }
    x
}
registerS3method("pip_output_value", "checked", pip_output_value.checked)

p5 <- pip_new("errors") |>
    pip_add("empty", \(x = 1) checked(numeric(0))) |>
    pip_add("fine", \(x = 1) x + 1)

By default, the run stops and the message tells which step failed:

tryCatch(
    pip_run(p5, lgr = NULL),
    error = \(e) conditionMessage(e)
)
# [1] "step 'empty': empty output of step 'empty'"

With on_error = "continue", the step is marked as failed and the run goes on with the steps that don’t depend on it. The run also signals a pipeflow_run_failed warning, which we suppress here.

suppressWarnings(
    pip_run(p5, lgr = NULL, on_error = "continue")
)

pip_data(p5)[, c("step", "state")]
#      step  state
#    <char> <char>
# 1:  empty failed
# 2:   fine   done

As usual, the condition of the failed step is stored as its output, and the hook is not applied to it.

conditionMessage(p5[["empty", "out"]])
# [1] "empty output of step 'empty'"

Summary