Nested pipelines

A pipeline can be a step of another pipeline. This allows to reuse a standard analysis pipeline inside a larger workflow. The key is as.function(), which turns a pipeline into a function, and functions are what steps are made of.

Inner pipeline

We start with the same pipeline as in the previous “Split, map, and reduce” vignette, fitting a linear model and returning its coefficients. This will serve as our inner pipeline.

library(pipeflow)

inner <- pip_new("coefficients") |>
    pip_add("input", \(input_df = NULL) input_df) |>
    pip_add(
        "fit",
        \(data = ~input, xVar = "x", yVar = "y") {
            lm(paste(yVar, "~", xVar), data = data)
        }
    ) |>
    pip_add("coefs", \(fit = ~fit, digits = 3) round(coefficients(fit), digits))

inner
# <pipeflow> coefficients (3 steps)
# ---------------------------------
#     step         params depends state
# 1: input       input_df           new
# 2:   fit data,xVar,yVar   input   new
# 3: coefs     fit,digits     fit   new
# ---------------------------------
# <ready> last run: never

A pipeline as a function

as.function() returns a function whose arguments are the unbound parameters of the pipeline, with the current values as defaults, and whose result is the output of the steps defined by out.

f_inner <- as.function(inner, out = "coefs")

formals(f_inner) |> str()
# Dotted pair list of 4
#  $ input_df: NULL
#  $ xVar    : chr "x"
#  $ yVar    : chr "y"
#  $ digits  : num 3

As you see, the function version of the inner pipeline has the same default parameters as the pipeline itself.

Under the hood, the function runs a copy of the inner pipeline. Let’s see this now:

params <- list(input_df = iris, xVar = "Sepal.Length", yVar = "Sepal.Width")
inner$set_params(params)$run()
# info [2026-10-19 13:46:43.462 UTC]: Starting run of pipeflow 'coefficients'
# info [2026-10-19 13:46:43.462 UTC]: Step 1/3 input
# info [2026-10-19 13:46:43.463 UTC]: Step 2/3 fit
# info [2026-10-19 13:46:43.465 UTC]: Step 3/3 coefs
# info [2026-10-19 13:46:43.466 UTC]: Finished run of pipeflow 'coefficients'

inner[["coefs", "out"]]
#  (Intercept) Sepal.Length 
#        3.419       -0.062

do.call(f_inner, params)
#  (Intercept) Sepal.Length 
#        3.419       -0.062

Note that since f_inner gets its own copy of the inner pipeline, later changes to the inner pipeline will not change the function.

Outer pipeline

As any other function, the function can be added as a step of another pipeline. We start with the simplest case, where the outer pipeline provides the data and the inner pipeline analyses it as a whole.

outer_all <- pip_new("analysis") |>
    pip_add("df", \(df = NULL) df) |>
    pip_add(
        "inner_run",
        as.function(
            inner,
            bind = list(input_df = ~df),   # df is passed to the inner pipeline
            out = "coefs"                  # coefs is returned from inner run
        )
    )

outer_all
# <pipeflow> analysis (2 steps)
# -----------------------------
#         step                    params depends state
# 1:        df                        df           new
# 2: inner_run input_df,xVar,yVar,digits      df   new
# -----------------------------
# <ready> last run: never

The argument bind wires the function into the outer pipeline. list(input_df = ~df) reads: the argument input_df of the function is bound to the output of the step df from the outer pipeline. This is the same as writing input_df = ~df as an argument of a step function, so the step inner_run now depends on df.

The remaining arguments of the function, xVar, yVar and digits, are not bound. They stay unbound parameters, and as such are parameters of the outer pipeline, without having to forward them:

pip_get_params(outer_all) |> str()
# List of 4
#  $ df    : NULL
#  $ xVar  : chr "Sepal.Length"
#  $ yVar  : chr "Sepal.Width"
#  $ digits: num 3

The df listed here belongs to the outer step df.

Let’s set the analysis parameters and run the outer pipeline:

outer_all |>
    pip_set_params(
        list(df = iris, xVar = "Sepal.Length", yVar = "Sepal.Width")
    ) |>
    pip_run()
# info [2026-10-19 13:46:43.536 UTC]: Starting run of pipeflow 'analysis'
# info [2026-10-19 13:46:43.537 UTC]: Step 1/2 df
# info [2026-10-19 13:46:43.537 UTC]: Step 2/2 inner_run
# info [2026-10-19 13:46:43.551 UTC]: Finished run of pipeflow 'analysis'

outer_all[["inner_run", "out"]]
#  (Intercept) Sepal.Length 
#        3.419       -0.062

The inner pipeline was run as a single step and returned the same coefficients as above where we called the function directly.

Outer pipeline with split data

The real strength shows when the inner pipeline is used multiple times in a “split-data” scenario. Here, the outer pipeline splits the data into subsets (see also the previous vignette), and input_df is bound to the output of the split step. Then the function is called once per subset.

outer <- pip_new("full analysis") |>
    pip_add("df", \(df = NULL) df) |>
    pip_add(
        "split_df",
        \(df = ~df, byVar = "by") split(df, f = df[[byVar]]),
        exec = "split"
    ) |>
    pip_add(
        "inner_run",
        as.function(
            inner,
            bind = list(input_df = ~split_df),    # bind to split data
            out = "coefs"
        )
    ) |>
    pip_add(
        "combine",
        \(coefs = ~inner_run) as.data.frame(do.call(rbind, coefs)),
        exec = "reduce"
    )

outer
# <pipeflow> full analysis (4 steps)
# ----------------------------------
#         step                    params   depends state   exec
# 1:        df                        df             new   auto
# 2:  split_df                  df,byVar        df   new  split
# 3: inner_run input_df,xVar,yVar,digits  split_df   new   auto
# 4:   combine                     coefs inner_run   new reduce
# ----------------------------------
# <ready> last run: never

The split output is partitioned, so the function is called once per subset, and its output is partitioned as well. The combine step is a reduce step and receives all partitions. As before, the unbound parameters of the inner pipeline are parameters of the outer pipeline, now next to byVar of the split step:

pip_get_params(outer) |> str()
# List of 5
#  $ df    : NULL
#  $ byVar : chr "by"
#  $ xVar  : chr "Sepal.Length"
#  $ yVar  : chr "Sepal.Width"
#  $ digits: num 3

Let’s set the analysis parameters and run the full pipeline:

outer |>
    pip_set_params(
        list(
            df = iris,
            xVar = "Sepal.Length",
            yVar = "Sepal.Width",
            byVar = "Species"
        )
    ) |>
    pip_run()
# info [2026-10-19 13:46:43.616 UTC]: Starting run of pipeflow 'full analysis'
# info [2026-10-19 13:46:43.616 UTC]: Step 1/4 df
# info [2026-10-19 13:46:43.616 UTC]: Step 2/4 split_df
# info [2026-10-19 13:46:43.618 UTC]: Step 3/4 inner_run
# info [2026-10-19 13:46:43.637 UTC]: Step 4/4 combine
# info [2026-10-19 13:46:43.639 UTC]: Finished run of pipeflow 'full analysis'

The output of the inner_run step is a partitioned list of coefficient vectors, one for each species,

outer[["inner_run", "out"]]
# $setosa
#  (Intercept) Sepal.Length 
#       -0.569        0.799 
# 
# $versicolor
#  (Intercept) Sepal.Length 
#        0.872        0.320 
# 
# $virginica
#  (Intercept) Sepal.Length 
#        1.446        0.232 
# 
# attr(,"class")
# [1] "list"                 "pipeflow_partitioned"

and the combine step returns the expected combined table.

outer[["combine", "out"]]
#            (Intercept) Sepal.Length
# setosa          -0.569        0.799
# versicolor       0.872        0.320
# virginica        1.446        0.232

Now suppose we want to change one of the model settings, say use Petal.Length instead of Sepal.Length as the predictor.

pip_set_params(outer, params = list(xVar = "Petal.Length"))

outer
# <pipeflow> full analysis (4 steps)
# ----------------------------------
#         step                    params   depends    state                 out   exec
# 1:        df                        df               done <data.frame[150x5]>   auto
# 2:  split_df                  df,byVar        df     done           <list[3]>  split
# 3: inner_run input_df,xVar,yVar,digits  split_df outdated           <list[3]>   auto
# 4:   combine                     coefs inner_run outdated   <data.frame[3x2]> reduce
# ----------------------------------
# <ready> last run: 2026-10-19 13:46:43

Since xVar is a parameter of the inner_run step, this step and its downstream steps have been marked as “outdated”.

Caching of inner pipelines

When converting a pipeline into a function, {pipeflow} caches the pipelines it has run (up to cache of them, 16 by default) and finds the one that belongs to a call by the bound arguments. A later call sets all parameters and inputs in that pipeline, so that exactly the steps that are outdated by the change are run again.

To see this at work, we use a minimal toy example. The inner pipeline squares numbers and scales the result. The counters tell us how often each step has actually run.

runs <- new.env()
runs$square <- runs$scale <- 0L
count <- function(step) runs[[step]] <- runs[[step]] + 1L
n_runs <- function() unlist(mget(c("square", "scale"), envir = runs))

toy <- pip_new("toy") |>
    pip_add("input", \(x = NULL) x) |>
    pip_add("square", \(x = ~input) {
        count("square")
        x^2
    }) |>
    pip_add("scale", \(sq = ~square, factor = 1) {
        count("scale")
        sq * factor
    })

The outer pipeline splits a list of two number vectors, a and b, and runs the toy pipeline on each of them.

toy_outer <- pip_new("toy outer") |>
    pip_add(
        "numbers",
        \(values = list(a = 1:3, b = 4:6)) values,
        exec = "split"
    ) |>
    pip_add(
        "toy_run",
        as.function(toy, bind = list(x = ~numbers), out = "scale")
    )

pip_run(toy_outer, lgr = NULL)
n_runs()
# square  scale 
#      2      2

Each step ran twice, once in the inner pipeline of a and once in the one of b. Now we change factor, which only the last step uses. Only this step is outdated in the inner pipelines, so only this step runs again:

pip_set_params(toy_outer, list(factor = 10))
pip_run(toy_outer, lgr = NULL)
n_runs()
# square  scale 
#      2      4

Next we change the input data and give a other numbers while b stays the same. The new numbers are a new input for the function, so they get a new inner pipeline, in which all steps run. For b nothing runs.

pip_set_params(toy_outer, list(values = list(a = 1:4, b = 4:6)))
pip_run(toy_outer, lgr = NULL)
n_runs()
# square  scale 
#      3      5

The cache now holds three inner pipelines, because the one for the old numbers of a is still there. When we go back to the original numbers, nothing needs to run at all:

pip_set_params(toy_outer, list(values = list(a = 1:3, b = 4:6)))
pip_run(toy_outer, lgr = NULL)
n_runs()
# square  scale 
#      3      5

toy_outer[["toy_run", "out"]]
# $a
# [1] 10 40 90
# 
# $b
# [1] 160 250 360
# 
# attr(,"class")
# [1] "list"                 "pipeflow_partitioned"

A few things to know:

cache should be at least the number of subsets. If it is smaller, the least recently used pipelines are dropped before they are needed again, and nothing is reused (cache = 0 turns the reuse off). The kept pipelines hold their inputs and outputs, so they need memory.
Each call has an overhead of a few milliseconds for setting the parameters and running the pipeline, so reuse pays when the inner steps are expensive.
A call that is matched to the “wrong” pipeline is only slower, never wrong.
Inputs are compared as they are, so a change made in place to an environment or an R6 object is not seen (as in a plain pipeline), and a forced run of the outer pipeline does not force the inner pipelines. Use cache = 0 for such inputs.
The pipelines belong to the function and are dropped with it. Serialising a function that has been called (e.g. to send it to a parallel worker) copies them, inputs and outputs included. A function that has not been called yet, or one made with cache = 0, is small.
Warnings and messages of the inner steps reach the outer run. If an inner step fails, the outer step fails with a message that names the partition and the inner step, unless the inner pipeline handles the failure itself (see “Handling failures”).
The inner pipelines are run silently. To follow their steps while debugging, set as.function(lgr = pipeflow_lgr) (or any other logger): each subset then logs a run of its own.

Full control: running the inner pipeline yourself

as.function() shares the parameters of the inner pipeline between all subsets. If you need more control, e.g. different settings per group or mapping over something else than the output of a split step, you can always “manually” clone and run the inner pipeline in a step function yourself.

# Helper to run inner pipeline
run_inner_pip <- function(pip, name, df) {
    pip$name <- sprintf("coefs for *%s*", name)

    # Set data subset for inner pipeline and run it
    pip_set_params(pip, list(input_df = df)) |> pip_run(lgr = NULL)

    pip[["coefs", "out"]]
}

manual <- pip_new("full analysis") |>
    pip_add("df", \(df = NULL) df) |>
    pip_add(
        "split_df",
        \(df = ~df, byVar = "by") split(df, f = df[[byVar]])
    ) |>
    pip_add(
        "inner_run",
        \(dfList = ~split_df, ...) {
            p <- pip_clone(inner)

            # Forward all parameters (from outer and inner)
            pip_set_params(p, .self$get_params(), unknown = "ignore")

            Map(
                f = run_inner_pip,
                name = names(dfList),
                df = dfList,
                MoreArgs = list(pip = p)
            )
        },
        params = pip_get_params(inner) # <--- default parameters of inner
    ) |>
    pip_add(
        "combine",
        \(coefs = ~inner_run) as.data.frame(do.call(rbind, coefs))
    )

manual
# <pipeflow> full analysis (4 steps)
# ----------------------------------
#         step                           params   depends state
# 1:        df                               df             new
# 2:  split_df                         df,byVar        df   new
# 3: inner_run dfList,input_df,xVar,yVar,digits  split_df   new
# 4:   combine                            coefs inner_run   new
# ----------------------------------
# <ready> last run: never

In this version, the inner_run step does not declare the parameters of the inner pipeline as arguments. Instead, its parameters are seeded with the inner pipeline’s parameters via params = pip_get_params(inner), and the step forwards the combined set of all existing parameters at run time. Three aspects are worth spelling out:

.self refers to the pipeline that is currently being run and is available inside every step function that refers to it, without having to declare it. It exposes the usual pipeline interface, so .self$get_params() returns the current unbound parameters of the outer pipeline. For more examples on this mechanism see the self-modifying pipelines vignette.
params = pip_get_params(inner) registers the inner pipeline’s unbound parameters (input_df, xVar, yVar, digits) as parameters of the inner_run step. They show up in the step’s params column, become part of the outer pipeline’s parameter set, and changing them outdates the step.
... takes the forwarded parameters: the parameters in params are passed to the step function when the step runs, so a function that does not declare them as arguments needs .... Where the names overlap, the values in params override the defaults of the function: had the inner pipeline an unbound parameter dfList, its value would replace the reference ~split_df of the inner_run step.

Let us now run the manual pipeline to see it in action.

manual |>
    pip_set_params(
        list(
            df = iris,
            xVar = "Sepal.Length",
            yVar = "Sepal.Width",
            byVar = "Species"
        )
    ) |>
    pip_run()
# info [2026-10-19 13:46:43.911 UTC]: Starting run of pipeflow 'full analysis'
# info [2026-10-19 13:46:43.911 UTC]: Step 1/4 df
# info [2026-10-19 13:46:43.912 UTC]: Step 2/4 split_df
# info [2026-10-19 13:46:43.913 UTC]: Step 3/4 inner_run
# info [2026-10-19 13:46:43.934 UTC]: Step 4/4 combine
# info [2026-10-19 13:46:43.936 UTC]: Finished run of pipeflow 'full analysis'

manual[["combine", "out"]]
#            (Intercept) Sepal.Length
# setosa          -0.569        0.799
# versicolor       0.872        0.320
# virginica        1.446        0.232

This version runs each subset from scratch every time the step is outdated. With as.function(), the same result is obtained with less code and with caching.

Built-in exec modes vs. nested pipelines

Both this and the previous vignette solve the same “split, apply, and combine” problem. The built-in execution modes (exec = "split"/"reduce") are the recommended default whenever the work fits into the steps of a single pipeline, while nested pipelines are the more general tool. As a rule of thumb:

Code: the built-in modes are declarative and need no extra code. as.function() needs one step and the bind list. Running the inner pipeline yourself requires cloning, mapping, and forwarding parameters.
Map shape: the built-in modes handle a single split point in the pipeline, while nested pipelines can be mapped over the partitions of a split step and can contain their own splits (nested splits). Running the inner pipeline yourself can map over arbitrary inputs (e.g. cross-validation folds or several data sets).
Run log and state: the built-in modes produce one unified run over all partitions, while a nested pipeline is run as one step of the outer pipeline, with its own log silenced.
Parallelism: the built-in modes are handled by {pipeflow}, whereas with a manual loop you control it and can parallelize it yourself.