---
title: "Nested pipelines"
output:
  rmarkdown::html_vignette:
    toc: true
    toc_depth: 4
description: >
  Shows how to run a pipeline as a step of another pipeline with
  as.function(), so that changing its parameters re-executes the outer step
  and only the inner steps that are affected.
vignette: >
  %\VignetteIndexEntry{Nested pipelines}
  %\VignetteEngine{knitr::rmarkdown}
  %\VignetteEncoding{UTF-8}
---

```{r knitr-setup, include = FALSE}
knitr::opts_chunk$set(
    comment = "#",
    prompt = FALSE,
    tidy = FALSE,
    cache = FALSE,
    collapse = TRUE
)

old <- options(width = 100L)
```


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"](v05a-split-map-reduce.html) vignette,
fitting a linear model and returning its coefficients. This will
serve as our inner pipeline.

```{r 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
```


### 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`.

```{r as-function}
f_inner <- as.function(inner, out = "coefs")

formals(f_inner) |> str()
```

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:

```{r}
params <- list(input_df = iris, xVar = "Sepal.Length", yVar = "Sepal.Width")
inner$set_params(params)$run()

inner[["coefs", "out"]]

do.call(f_inner, params)
```

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.

```{r outer-all}
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
```

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:

```{r outer-all-params}
pip_get_params(outer_all) |> str()
```

The `df` listed here belongs to the outer step `df`.

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

```{r run-outer-all}
outer_all |>
    pip_set_params(
        list(df = iris, xVar = "Sepal.Length", yVar = "Sepal.Width")
    ) |>
    pip_run()

outer_all[["inner_run", "out"]]
```

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](v05a-split-map-reduce.html)), and `input_df` is bound
to the output of the split step. Then the function is called once per subset.

```{r outer-pipeline}
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
```

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:

```{r outer-params}
pip_get_params(outer) |> str()
```

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

```{r run-outer}
outer |>
    pip_set_params(
        list(
            df = iris,
            xVar = "Sepal.Length",
            yVar = "Sepal.Width",
            byVar = "Species"
        )
    ) |>
    pip_run()
```

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

```{r inner-run-out}
outer[["inner_run", "out"]]
```

and the `combine` step returns the expected combined table.

```{r combine-out}
outer[["combine", "out"]]
```

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

```{r change-param}
pip_set_params(outer, params = list(xVar = "Petal.Length"))

outer
```

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.

```{r toy-inner}
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.

```{r toy-outer}
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()
```

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:

```{r toy-param}
pip_set_params(toy_outer, list(factor = 10))
pip_run(toy_outer, lgr = NULL)
n_runs()
```

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.

```{r toy-input}
pip_set_params(toy_outer, list(values = list(a = 1:4, b = 4:6)))
pip_run(toy_outer, lgr = NULL)
n_runs()
```

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:

```{r toy-back}
pip_set_params(toy_outer, list(values = list(a = 1:3, b = 4:6)))
pip_run(toy_outer, lgr = NULL)
n_runs()

toy_outer[["toy_run", "out"]]
```

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"](v05c-error-handling.html)).
* 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.

```{r manual}
# 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
```

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](v06-self-modify-pipeline.html) 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.

```{r manual-run}
manual |>
    pip_set_params(
        list(
            df = iris,
            xVar = "Sepal.Length",
            yVar = "Sepal.Width",
            byVar = "Species"
        )
    ) |>
    pip_run()

manual[["combine", "out"]]
```

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](v05a-split-map-reduce.html) 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.

```{r, include = FALSE}
options(old)
```