{pipeflow} aims to offer a lean and intuitive interface that enables new users to get started quickly without having to learn a lot of new functions. At the same time, it was designed to provide easy access to the underlying data structures to allow advanced users to modify the pipeline basically in any way they want.
To see this, let’s briefly inspect the internal structure of a pipeline object.
pip <- pip_new("my-pipeline") |>
pip_add("init", \(xInit = 0) xInit) |>
pip_add("f1", \(x = ~init) x + 1) |>
pip_add("f2", \(x = ~f1) x + 2) |>
pip_add("f3", \(x = ~f2) x + 3)
str(pip)
# List of 3
# $ name : chr "my-pipeline"
# $ view : NULL
# $ pipenv:<environment: 0x5bb0698a66e8>There is the name of the pipeline, which is just a
character string, as well as a view entry, which is
undefined initially.
The most “interesting” part is the pipenv: an
environment that holds the pipeline’s actual state. It is shared by the
pipeline and all of its views, which is why operations on a view write
through to the underlying pipeline.
Listing the pipenv1 reveals the dataentry, which
contains the pipeline’s step table as a data.table with one
row per step:
pip$pipenv$data # or pip_data(pip)
# step fun params out state tags locked exec time depends unbound nodeId
# <char> <list> <list> <list> <char> <list> <lgcl> <char> <POSc> <list> <list> <int>
# 1: init <function[1]> <list[1]> [NULL] new FALSE auto 2026-10-20 10:26:06 xInit 0
# 2: f1 <function[1]> <list[1]> [NULL] new FALSE auto 2026-10-20 10:26:06 init 1
# 3: f2 <function[1]> <list[1]> [NULL] new FALSE auto 2026-10-20 10:26:06 f1 2
# 4: f3 <function[1]> <list[1]> [NULL] new FALSE auto 2026-10-20 10:26:06 f2 3Many of the columns should be already familar to you and most of them
can be manipulated safely via the [<- and
[[<- operators, for example:
pip[["f2", "step"]] <- "my_pretty_f2" # updates downstream 'depends'
pip
# <pipeflow> my-pipeline (4 steps)
# --------------------------------
# step params depends state
# 1: init xInit new
# 2: f1 x init new
# 3: my_pretty_f2 x f1 new
# 4: f3 x my_pretty_f2 new
# --------------------------------
# <ready> last run: neverpip[step %like% "f", "locked"] <- TRUE
pip
# <pipeflow> my-pipeline (4 steps)
# --------------------------------
# step params depends state locked
# 1: init xInit new FALSE
# 2: f1 x init new TRUE
# 3: my_pretty_f2 x f1 new TRUE
# 4: f3 x my_pretty_f2 new TRUE
# --------------------------------
# <ready> last run: neverAn exception are the last three columns (depends,
unbound, nodeId): these are usually derived
indirectly from the step definition and therefore protected against
direct assignment.
pip[["init", "nodeId"]] <- 99L
# Error in `[[<-.pipeflow`:
# ! direct assignment to column 'nodeId' is not supported.Of course, you can even skip the [<- and
[[<- operators and work directly on the
data.table object, but with an increased risk to invalidate
the internal consistency of the overall pipeline structure, for
example:
dat <- pip$pipenv$data
dat[2, "step"] <- "new f1 name" # fails to update downstream 'depends'
dat[3, "nodeId"] <- 99L # breaks link to internal DAG node
assign("data", dat, envir = pip$pipenv)
pip
# <pipeflow> my-pipeline (4 steps)
# --------------------------------
# step params depends state locked
# 1: init xInit new FALSE
# 2: new f1 name x init new TRUE
# 3: my_pretty_f2 x f1 new TRUE
# 4: f3 x my_pretty_f2 new TRUE
# --------------------------------
# <ready> last run: neverFor this reason, direct manipulation is useful during debugging and you should mostly stick to the provided operators or functions unless you really know what you are doing.
After this little excursion let’s next see how to safely modify our pipeline structure at runtime.
pip <- pip_new("my-pipeline") |>
pip_add("init", \(xInit = 0) xInit) |>
pip_add("f1", \(x = ~init) x + 1) |>
pip_add("f2", \(x = ~f1) x + 2) |>
pip_add("f3", \(x = ~f2) x + 3)
(pip_run(pip))
# info [2026-10-20 10:26:06.143 UTC]: Starting run of pipeflow 'my-pipeline'
# info [2026-10-20 10:26:06.144 UTC]: Step 1/4 init
# info [2026-10-20 10:26:06.144 UTC]: Step 2/4 f1
# info [2026-10-20 10:26:06.145 UTC]: Step 3/4 f2
# info [2026-10-20 10:26:06.146 UTC]: Step 4/4 f3
# info [2026-10-20 10:26:06.147 UTC]: Finished run of pipeflow 'my-pipeline'
# <pipeflow> my-pipeline (4 steps)
# --------------------------------
# step params depends state out
# 1: init xInit done 0
# 2: f1 x init done 1
# 3: f2 x f1 done 3
# 4: f3 x f2 done 6
# --------------------------------
# <ready> last run: 2026-10-20 10:26:06This pipeline just adds 1, 2, and 3 to the initial value,
respectively. Let’s modify step f2 that in turn will modify
f3 at runtime based on the interim result passed into
f2.
pip |> pip_replace(
"f2",
\(x = ~f1) {
if (x > 10) {
.self$replace("f3", \(x = ~f1) x * 3)
return(x / 2)
}
x + 2
}
)Basically, step f2 now checks if the input is greater
than 10, and if so, it replaces step f3 with a new step now
referencing f1 that multiplies the input passed from
f1 by 3 and returns half of the input.
Note that .self does not need to be declared as an
argument. {pipeflow} provides it each time it calls a step function
whose body refers to .self (also within functions defined
in the body). Access that is not visible in the body, such as
get(".self"), is not detected.
To see this, let’s try it with an input of 15.
pip |>
pip_set_params(list(xInit = 15)) |>
pip_run()
# info [2026-10-20 10:26:06.180 UTC]: Starting run of pipeflow 'my-pipeline'
# info [2026-10-20 10:26:06.180 UTC]: Step 1/4 init
# info [2026-10-20 10:26:06.181 UTC]: Step 2/4 f1
# info [2026-10-20 10:26:06.183 UTC]: Step 3/4 f2
# info [2026-10-20 10:26:06.189 UTC]: Step 4/4 f3
# info [2026-10-20 10:26:06.190 UTC]: Finished run of pipeflow 'my-pipeline'
pip
# <pipeflow> my-pipeline (4 steps)
# --------------------------------
# step params depends state out
# 1: init xInit done 15
# 2: f1 x init done 16
# 3: f2 x f1 done 8
# 4: f3 x f1 done 48
# --------------------------------
# <ready> last run: 2026-10-20 10:26:06We see that both the output of the pipeline and the dependencies of the last step have changed. Let’s confirm by inspecting the function of the last step.
Next, we get even more hacky and instead of just replacing, we will go a bit further to insert and remove steps. The pipeline definition is as follows:
pip <- pip_new("hicky-hacky") |>
pip_add("init", \(xInit = 0) xInit) |>
pip_add("f1", \(x = ~init) x + 1) |>
pip_add(
"f2",
\(x = ~f1) {
if (x > 10) {
.self |>
pip_add("f2a", \(x = ~f1) x + 21, after = "f1") |>
pip_add("f2b", \(x = ~f2a) x + 22, after = "f2a") |>
pip_replace("f3", \(x = ~f2b) x + 30) |>
pip_remove("f2")
}
x + 2
}
) |>
pip_add("f3", \(x = ~f2) x + 3)If the input is greater than 10, we insert two new steps
f2a and f2b after f1, remove
f2, and replace f3 with a new step that adds
30 to the input. Let’s first run with the initial value of 0 to see the
original output.
pip_run(pip)
# info [2026-10-20 10:26:06.237 UTC]: Starting run of pipeflow 'hicky-hacky'
# info [2026-10-20 10:26:06.237 UTC]: Step 1/4 init
# info [2026-10-20 10:26:06.238 UTC]: Step 2/4 f1
# info [2026-10-20 10:26:06.239 UTC]: Step 3/4 f2
# info [2026-10-20 10:26:06.240 UTC]: Step 4/4 f3
# info [2026-10-20 10:26:06.241 UTC]: Finished run of pipeflow 'hicky-hacky'
pip
# <pipeflow> hicky-hacky (4 steps)
# --------------------------------
# step params depends state out
# 1: init xInit done 0
# 2: f1 x init done 1
# 3: f2 x f1 done 3
# 4: f3 x f2 done 6
# --------------------------------
# <ready> last run: 2026-10-20 10:26:06Next, we set the initial value to 11 to trigger the changes.
pip |>
pip_set_params(list(xInit = 11)) |>
pip_run()
# info [2026-10-20 10:26:06.259 UTC]: Starting run of pipeflow 'hicky-hacky'
# info [2026-10-20 10:26:06.259 UTC]: Step 1/4 init
# info [2026-10-20 10:26:06.259 UTC]: Step 2/4 f1
# info [2026-10-20 10:26:06.260 UTC]: Step 3/4 f2
# info [2026-10-20 10:26:06.264 UTC]: Step 4/4 f3
# info [2026-10-20 10:26:06.265 UTC]: Finished run of pipeflow 'hicky-hacky'
pip
# <pipeflow> hicky-hacky (5 steps)
# --------------------------------
# step params depends state out
# 1: init xInit done 11
# 2: f1 x init done 12
# 3: f2a x f1 new [NULL]
# 4: f2b x f2a done
# 5: f3 x f2b new [NULL]
# --------------------------------
# <ready> last run: 2026-10-20 10:26:06While the structure has changed as expected, some steps were not yet
run. In fact, since originally step f3came after
f2, and in contrast to what the log is showing, instead of
step f3, actually the new step f2b was run2 , albeit
with x = NULL as input.
So to have the true results, we need to re-init the parameter and need to re-run the pipeline.
pip |>
pip_set_params(list(xInit = 11)) |>
pip_run()
# info [2026-10-20 10:26:06.282 UTC]: Starting run of pipeflow 'hicky-hacky'
# info [2026-10-20 10:26:06.283 UTC]: Step 1/5 init - skipping done step
# info [2026-10-20 10:26:06.283 UTC]: Step 2/5 f1 - skipping done step
# info [2026-10-20 10:26:06.283 UTC]: Step 3/5 f2a
# info [2026-10-20 10:26:06.284 UTC]: Step 4/5 f2b
# info [2026-10-20 10:26:06.285 UTC]: Step 5/5 f3
# info [2026-10-20 10:26:06.286 UTC]: Finished run of pipeflow 'hicky-hacky'
pip
# <pipeflow> hicky-hacky (5 steps)
# --------------------------------
# step params depends state out
# 1: init xInit done 11
# 2: f1 x init done 12
# 3: f2a x f1 done 33
# 4: f2b x f2a done 55
# 5: f3 x f2b done 85
# --------------------------------
# <ready> last run: 2026-10-20 10:26:06Now the output of all steps is as expected. If we want to use
{pipeflow} in production, obviously, having to re-run the pipeline and
temporarily showing a wrong log is not ideal. That is, ideally, the
pipeline run would be aborted once away after all changes were made in
f2 and then re-run right away from the beginning. Also,
this process potentially should be repeated recursively until the
structure does not change anymore.
Luckily, with some minimal changes, this behaviour can be achieved
with {pipeflow}. First, for any step where you want to restart the
pipeline run, you need to call .self$restart(), so we adapt
the f2 function as follows:
pip <- pip_new("hacky-with-restart") |>
pip_add("init", \(xInit = 0) xInit) |>
pip_add("f1", \(x = ~init) x + 1) |>
pip_add(
"f2",
\(x = ~f1) {
if (x > 10) {
.self |>
pip_add("f2a", \(x = ~f1) x + 21, after = "f1") |>
pip_add("f2b", \(x = ~f2a) x + 22, after = "f2a") |>
pip_replace("f3", \(x = ~f2b) x + 30) |>
pip_remove("f2")
.self$restart() # <-- restart the run
}
x + 2
}
) |>
pip_add("f3", \(x = ~f2) x + 3)Second, you just run the pipeline as usual.
pip |>
pip_set_params(list(xInit = 11)) |>
pip_run()
# info [2026-10-20 10:26:06.320 UTC]: Starting run of pipeflow 'hacky-with-restart'
# info [2026-10-20 10:26:06.320 UTC]: Step 1/4 init
# info [2026-10-20 10:26:06.321 UTC]: Step 2/4 f1
# info [2026-10-20 10:26:06.322 UTC]: Step 3/4 f2
# info [2026-10-20 10:26:06.325 UTC]: Restarting pipeline execution.
# info [2026-10-20 10:26:06.326 UTC]: Restarting run of pipeflow 'hacky-with-restart'
# info [2026-10-20 10:26:06.326 UTC]: Step 1/5 init
# info [2026-10-20 10:26:06.326 UTC]: Step 2/5 f1
# info [2026-10-20 10:26:06.327 UTC]: Step 3/5 f2a
# info [2026-10-20 10:26:06.328 UTC]: Step 4/5 f2b
# info [2026-10-20 10:26:06.329 UTC]: Step 5/5 f3
# info [2026-10-20 10:26:06.329 UTC]: Finished run of pipeflow 'hacky-with-restart'As you can see, the run was aborted right after step f2
and re-run from the start based on the new structure. As a result, the
log now is fully aligned with the performed pipeline run.
Looking at the final pipeline overview, we see that the output matches the expected output of the modified pipeline.
pip
# <pipeflow> hacky-with-restart (5 steps)
# ---------------------------------------
# step params depends state out
# 1: init xInit done 11
# 2: f1 x init done 12
# 3: f2a x f1 done 33
# 4: f2b x f2a done 55
# 5: f3 x f2b done 85
# ---------------------------------------
# <ready> last run: 2026-10-20 10:26:06Of course, this was just a toy example to show some possibilities, but I have made use of this feature already in various projects and may present one of them in a more sophisticated example in the future.
Lastly note that since you have full access to the pipeline object,
of course, you can get even more hacky, but be aware that some
additional operations are done under the hood when steps are added or
removed. It is therefore not recommended to “manually” manipulate the
internal data.table object in terms of removing or adding rows, or
changing important columns such as depends or
nodeId as this immediately would invalidate the internal
consistency of the dependency graph.
On the other hand, changing entries in columns such as
tags, time, state or
output is generally not critical. If in doubt, just try and
see what works.
.self.self is provided for the duration of the step call and
released as soon as the step returns. This keeps the pipeline out of the
step’s output: a step that returns a closure, a formula or any other
object that captures its call frame would otherwise carry the whole
pipeline along, which shows up when the output is serialized, cached or
sent to another process.
As a consequence, .self must be used while the step
runs. A function that is returned by a step and refers to
.self fails when it is called later.
pip <- pip_new("lifetime") |>
pip_add("make", \(x = 1) function() .self[["name"]])
pip_run(pip, lgr = NULL)
getName <- pip[["make", "out"]]
tryCatch(getName(), error = \(e) conditionMessage(e))
# [1] "object '.self' not found"If a step needs pipeline information later on, read it during the run and return the value instead of the pipeline.