ARROW-17252: [R] Intermittent valgrind failure

r/R/compute.R

@@ -344,7 +344,7 @@ cast_options <- function(safe = TRUE, ...) {
 #' @return `NULL`, invisibly
 #' @export
 #'
-#' @examplesIf arrow_with_dataset()
+#' @examplesIf arrow_with_dataset() && identical(Sys.getenv("R_ARROW_COLLECT_WITH_UDF"), "true")
 #' library(dplyr, warn.conflicts = FALSE)
 #'
 #' some_model <- lm(mpg ~ disp + cyl, data = mtcars)
@@ -358,10 +358,15 @@ cast_options <- function(safe = TRUE, ...) {
 #'   auto_convert = TRUE
 #' )
 #'
+#' # User-defined functions require some special handling
+#' # in the query engine which currently require an opt-in using
+#' # the R_ARROW_COLLECT_WITH_UDF environment variable.
+#' Sys.setenv(R_ARROW_COLLECT_WITH_UDF = "true")
 #' as_arrow_table(mtcars) %>%
 #'   transmute(mpg, mpg_predicted = mtcars_predict_mpg(disp, cyl)) %>%
 #'   collect() %>%
 #'   head()
+#' Sys.unsetenv("R_ARROW_COLLECT_WITH_UDF")
 #'
 register_scalar_function <- function(name, fun, in_type, out_type,
                                      auto_convert = FALSE) {

r/R/table.R

@@ -331,5 +331,12 @@ as_arrow_table.arrow_dplyr_query <- function(x, ...) {
   # See query-engine.R for ExecPlan/Nodes
   plan <- ExecPlan$create()
   final_node <- plan$Build(x)
-  plan$Run(final_node, as_table = TRUE)
+
+  run_with_event_loop <- identical(
+    Sys.getenv("R_ARROW_COLLECT_WITH_UDF", ""),
+    "true"
+  )
+
+  result <- plan$Run(final_node, as_table = run_with_event_loop)
+  as_arrow_table(result)
 }

r/tests/testthat/test-compute.R

@@ -81,6 +81,8 @@ test_that("arrow_scalar_function() works with auto_convert = TRUE", {
 
 test_that("register_scalar_function() adds a compute function to the registry", {
   skip_if_not(CanRunWithCapturedR())
+  # until R_ARROW_COLLECT_WITH_UDF is no longer needed to slience valgrind
+  skip_on_linux_devel()
 
   register_scalar_function(
     "times_32",
@@ -103,12 +105,14 @@ test_that("register_scalar_function() adds a compute function to the registry",
     Scalar$create(32L, float64())
   )
 
-  expect_identical(
-    record_batch(a = 1L) %>%
-      dplyr::mutate(b = times_32(a)) %>%
-      dplyr::collect(),
-    tibble::tibble(a = 1L, b = 32.0)
-  )
+  withr::with_envvar(list(R_ARROW_COLLECT_WITH_UDF = "true"), {
+    expect_identical(
+      record_batch(a = 1L) %>%
+        dplyr::mutate(b = times_32(a)) %>%
+        dplyr::collect(),
+      tibble::tibble(a = 1L, b = 32.0)
+    )
+  })
 })
 
 test_that("arrow_scalar_function() with bad return type errors", {
@@ -237,24 +241,26 @@ test_that("user-defined functions work during multi-threaded execution", {
   )
   on.exit(unregister_binding("times_32", update_cache = TRUE))
 
-  # check a regular collect()
-  result <- open_dataset(tf_dataset) %>%
-    dplyr::mutate(fun_result = times_32(value)) %>%
-    dplyr::collect() %>%
-    dplyr::arrange(row_num)
+  withr::with_envvar(list(R_ARROW_COLLECT_WITH_UDF = "true"), {
+    # check a regular collect()
+    result <- open_dataset(tf_dataset) %>%
+      dplyr::mutate(fun_result = times_32(value)) %>%
+      dplyr::collect() %>%
+      dplyr::arrange(row_num)
 
-  expect_identical(result$fun_result, example_df$value * 32)
+    expect_identical(result$fun_result, example_df$value * 32)
 
-  # check a write_dataset()
-  open_dataset(tf_dataset) %>%
-    dplyr::mutate(fun_result = times_32(value)) %>%
-    write_dataset(tf_dest)
+    # check a write_dataset()
+    open_dataset(tf_dataset) %>%
+      dplyr::mutate(fun_result = times_32(value)) %>%
+      write_dataset(tf_dest)
 
-  result2 <- dplyr::collect(open_dataset(tf_dest)) %>%
-    dplyr::arrange(row_num) %>%
-    dplyr::collect()
+    result2 <- dplyr::collect(open_dataset(tf_dest)) %>%
+      dplyr::arrange(row_num) %>%
+      dplyr::collect()
 
-  expect_identical(result2$fun_result, example_df$value * 32)
+    expect_identical(result2$fun_result, example_df$value * 32)
+  })
 })
 
 test_that("user-defined error when called from an unsupported context", {