GH-35649: [R] Always call RecordBatchReader::ReadNext() from DuckDB from the main R thread

[paleolimbot]

Rationale for this change

When passing a DuckDB result to Arrow via to_arrow() whose input was an Arrow dataset, calls to R code from other threads can occur in some DuckDB operations. This caused a crash or hang on Linux when attempting to combine pivot_longer() and write_dataset().

What changes are included in this PR?

• Added a wrapper class around the RecordBatchReader that routes calls to ReadNext() through SafeCallIntoR().

Are these changes tested?

I can't find a new case that isn't covered by our existing tests, although I did remove a skip that was causing a similar problem at one point (#33033). Because it's difficult to predict/test where duckdb evaluates R code, it's hard to know exactly what to test here (I would have expected R code to be evaluated/a crash to occur with many of our existing tests, but even the pivot_longer() example does not crash on MacOS and Windows 🤷 ).

I did verify on Ubuntu 22.04 that the reprex kindly provided by @PMassicotte errors before this PR and does not error after this PR:

library(tidyverse)
library(arrow)

one_level_tree <- tempfile()

mtcars |>
  to_duckdb() |>
  pivot_longer(everything()) |>
  to_arrow() |>
  # collect() |> # collecting make it work, otherwise, it hangs on write_dataset()
  write_dataset(one_level_tree, partitioning = "name")

list.files(one_level_tree, recursive = TRUE)

Are there any user-facing changes?

There are no user facing changes.

• Closes: [R][C++] write_dataset() hangs when pivot_() operation is performed before #35649

[github-actions]

⚠️ GitHub issue #35649 has been automatically assigned in GitHub to PR creator.

[PMassicotte]

Thank you very much @paleolimbot for looking and fixing the issue! Really love the Arrow community!

[thisisnic]

This all looks good, so will approve shortly, but @paleolimbot, mainly for the sake of my own understanding, would you mind expanding on why routing calls to ReadNext() through SafeCallIntoR() is needed? What is SafeCallIntoR() doing?

[paleolimbot]

It's a good thing for all of our sanity!

SafeCallIntoR() either evaluates the expression it contains once on the main R thread OR errors it can't. The RunWithCapturedR() wrapper launches a thread (via the IO thread pools), sets up a "listener" on the main R thread that waits for requests to evaluate R code from SafeCallIntoR(). It's not a perfect demo, but there's some test code that demonstrates this:

arrow/r/src/safe-call-into-r-impl.cpp

Lines 62 to 76 in 8b4a548

	auto result = RunWithCapturedR<std::string>([&thread, r_fun_that_returns_a_string]() {
	auto fut = arrow::Future<std::string>::Make();
	thread = std::thread([&fut, r_fun_that_returns_a_string]() {
	auto result = SafeCallIntoR<std::string>(
	[&] { return cpp11::as_cpp<std::string>(r_fun_that_returns_a_string()); });
	fut.MarkFinished(result);
	});
	return fut;
	});
	if (thread.joinable()) {
	thread.join();
	}

The main difference with actual code is that the call to SafeCallIntoR() is usually buried deep in some utility code (e.g., read bytes from an R connection by calling the base R readBin() function) and the top-level call usually knows nothing about this (e.g., "Read this parquet file").

If SafeCallIntoR() is invoked from the R thread, it just evaluates the function without any of that dance. Basically, it is used everywhere we call anything from the R API (as in #include <R.h>) or cpp11 (because cpp11 wraps a lot of calls to R.h functions with the assumption that all of this is happening on the main thread).

[thisisnic]

Thanks for making these changes!

[conbench-apache-arrow]

Conbench analyzed the 6 benchmark runs on commit a75299d8.

There were 9 benchmark results indicating a performance regression:

• Commit Run on ursa-thinkcentre-m75q at 2023-07-02 07:20:16Z
  • params=<SubtractChecked, Int64Type>/size:524288/inverse_null_proportion:0, source=cpp-micro, suite=arrow-compute-scalar-arithmetic-benchmark
  • source=cpp-micro, suite=parquet-bloom-filter-benchmark
• and 7 more (see the report linked below)

The full Conbench report has more details.