pip_add: Add a step

In pipeflow: Fast Interactive Data Analysis Pipelines

Add a step

Description

Adds a named step to the pipeline. Each step is a function whose parameters either hold constant defaults or reference the output of a prior step using formula notation (~step_name). Dependencies are validated when the step is added.

Usage

pip_add(
  x,
  step,
  fun,
  tags = character(0),
  after = length(x),
  params = list(),
  exec = "auto"
)

Arguments

x	A pipeflow pipeline object.
step	Unique step name.
fun	Function to execute for the step. Each function parameter must have a default value. Default values that are simple constants are resolved immediately. Default values that are formulas like ~other_step are treated as dependencies to those steps and resolved to the respective output values at runtime once the step is executed. A reference wrapped in try(), like ~try(other_step), accepts a failed input when the pipeline is run with on_error = "continue": instead of the output of the referenced step, the argument then receives a condition of class pipeflow_failure (see section 'Continuing after errors' in pip_run()). The function can refer to the running pipeline as .self (see section 'The pipeline as .self' in pip_add()).
tags	Optional character vector of tags belonging to the step. Can also be adjusted later using ⁠[pip_tag()]⁠.
after	Optional position after which the new step should be inserted (defaults to last position). Can be a step name or an integer index. If set to 0, the new step will be inserted at the beginning of the pipeline.
params	Optional named list of parameter values, which will be merged with the defaults of fun (if overlapping names, the default values in fun take precedence). There are two use cases for params: Provide param values programmatically when adding steps at runtime Provide extra param values to "mark" dependencies that are defined in pipelines nested in a step, which ensures that the step (and with that the pipeline in the step) is re-executed when one of the respective param values change.
exec	Execution mode for this step. One of "auto", "split", "reduce" or "plain". Using execution mode exec = split, the output of the step is marked as partitioned output. In this mode, any step that depends on the split step (directly or indirectly) will have its output automatically mapped partition-wise during step execution. The reduce mode expects partitioned input and passes it through without mapping, while plain mode only accepts non-partitioned input and always intends to execute a single call. If a split pipeline fails in a run with on_error = "continue", a reduce step also accepts the resulting failure object when it takes it via ~try(). In summary: auto: map if partitioned input appears, otherwise single call split: single call, then mark output as partitioned reduce: single call, but only valid with partitioned input or, via ~try(), a failure object plain: single call, only valid with non-partitioned input

Details

If after was specified, the new step will be inserted after the given step or position. Be aware that in contrast to adding a step at the end, inserting a step in the middle is a rather expensive operation as it requires re-wiring parts of the internal pipeline structure, especially if the new step is inserted at an early position.

The pipeline as .self

A step function can refer to the pipeline that is being run as .self without declaring it as an argument, for example to set parameters or add steps at runtime, or to call .self$restart() or .self$halt() (see pip_run()). .self is always the full pipeline, also when a view is run. pipeflow provides it each time the step runs, but only if the body of the step function contains the name .self (also within functions defined in the body). Indirect access, such as get(".self"), is not detected. .self is a reserved name: it can't be an argument of fun or a name in params.

Value

The updated pipeline, invisibly.

Examples

# --- Tags, and view filtering ---
p <- pip_new("analysis") |>
  pip_add("load", \(n = 5) seq_len(n), tags = c("io", "raw")) |>
  pip_add("clean", \(x = ~load) x * 2, tags = c("io", "process")) |>
  pip_add("fit", \(x = ~clean) sum(x), tags = c("model", "core", "daily")) |>
  pip_add("report", \(x = ~fit) paste("result:", x), tags = "report")

pip_run(p)
p

# Filter by tag using pip_view — keeps steps with any matching tag
pip_view(p, tags = "daily")
pip_view(p, tags = "core")
pip_view(p, tags = c("raw", "report"))

# --- Split / reduce execution modes ---
q <- pip_new("split-demo") |>
  pip_add("data", \(x = iris) x) |>
  pip_add("split", \(x = ~data) split(x, x$Species),
    exec = "split"
  ) |>
  pip_add("stats", \(x = ~split) summary(x)) |>
  pip_add("combine", \(x = ~stats) do.call(rbind, x),
    exec = "reduce"
  )

pip_run(q)
q[["stats", "out"]]   # partitioned list — one summary per species
q[["combine", "out"]] # combined table

# --- Insert a step at a specific position with 'after' ---
p2 <- pip_new("insert-demo") |>
  pip_add("load", \(x = 1) x) |>
  pip_add("fit", \(x = ~load) x + 1)
p2
pip_add(p2, "clean", \(x = ~load) x * 2, after = "load")
p2  # "load", "clean", "fit" — inserted after "load"

# after = 0 inserts the new step at the very beginning
pip_add(p2, "preload", \(x = 1) x, after = 0)
p2

# --- Provide parameter values programmatically with 'params' ---
p3 <- pip_new("params-demo") |>
  pip_add("load", \(x = 1, ...) c(x, ...), params = list(size = 42))
pip_run(p3)
p3
p3[["load", "out"]] # c(1, size = 42) — extra params passed via `...`

pipeflow documentation built on Oct. 13, 2026, 5:06 p.m.