sketch parallel aggregation of dynamic branches within static branches

Author: wlandau

R/tar_map_rep.R

@@ -26,8 +26,14 @@
 #' @param columns A tidyselect expression to select which columns of `values`
 #'   to append to the output. Columns already in the target output
 #'   are not appended.
-#' @param combine Logical of length 1, whether to statically combine
-#'   all the results into a single target downstream.
+#' @param combine Logical of length 1, whether to create additional downstream
+#'   targets to combine the results of static branches.
+#'   The `values` argument must not be `NULL`
+#'   for this combining to take effect.
+#'   If `combine` is `TRUE` and `values` is not `NULL`,
+#'   then separate targets aggregate all dynamic branches within each
+#'   static branch, and then a final target combines all
+#'   the static branches together.
 #' @examples
 #' if (identical(Sys.getenv("TAR_LONG_EXAMPLES"), "true")) {
 #' targets::tar_dir({ # tar_dir() runs code from a temporary directory.

R/tar_map_rep_raw.R

@@ -112,7 +112,7 @@ tar_map_rep_raw <- function(
     is.null(values),
     target_dynamic,
     do.call(
-      tar_map,
+      what = tar_map,
       args = list(
         target_dynamic,
         values = values,
@@ -126,12 +126,58 @@ tar_map_rep_raw <- function(
   if (!unlist && is.list(target_static)) {
     target_static <- target_static[[1L]]
   }
-  target_combine <- if_any(
-    is.null(values) || !combine,
-    NULL,
-    tar_combine_raw(
+  target_combine <- NULL
+  target_combine_dynamic <- NULL
+  if (combine && !is.null(values)) {
+    expr <- substitute(
+      targets::tar_target_raw(
+        name = paste0(name, "_combine"),
+        command = as.symbol(name),
+        packages = packages,
+        library = library,
+        format = format,
+        repository = repository,
+        iteration = "vector",
+        error = error,
+        memory = memory,
+        garbage_collection = garbage_collection,
+        deployment = deployment,
+        priority = priority,
+        resources = resources,
+        storage = storage,
+        retrieval = retrieval,
+        cue = cue,
+        description = description
+      ),
+      env = list(
+        packages = packages,
+        library = library,
+        format = format,
+        repository = repository,
+        iteration = "vector",
+        error = error,
+        memory = memory,
+        garbage_collection = garbage_collection,
+        deployment = deployment,
+        priority = priority,
+        resources = resources,
+        storage = storage,
+        retrieval = retrieval,
+        cue = cue,
+        description = description
+      )
+    )
+    target_combine_dynamic <- tar_eval_raw(
+      expr = expr,
+      values = list(name = map_chr(target_static, ~.x$settings$name))
+    )
+    names(target_combine_dynamic) <- map_chr(
+      target_combine_dynamic,
+      ~.x$settings$name
+    )
+    target_combine <- tar_combine_raw(
       name = name,
-      target_static,
+      target_combine_dynamic,
       command = tar_map_combine_command,
       use_names = TRUE,
       packages = character(0),
@@ -141,15 +187,16 @@ tar_map_rep_raw <- function(
       error = error,
       memory = memory,
       garbage_collection = garbage_collection,
-      deployment = "main",
+      deployment = deployment,
       priority = priority,
       cue = cue,
       description = description
     )
-  )
+  }
   out <- list(
     batch_index = target_batch,
     static_branches = target_static,
+    combine_dynamic = target_combine_dynamic,
     combine = target_combine
   )
   if (unlist) {