[R] RecordBatchReaderHead from ExecPlan with UDF cannot be read

[asfimport]

  register_scalar_function(
    "times_32",
    function(context, x) x * 32.0,
    int32(),
    float64(),
    auto_convert = TRUE
  )
  record_batch(a = 1:1000) %>%
    dplyr::mutate(b = times_32(a)) %>%
    as_record_batch_reader() %>%
    head(11) %>%
    as_arrow_table()

# Error: NotImplemented: Call to R (resolve scalar user-defined function output data type) from a non-R thread from an unsupported context
# /arrow/cpp/src/arrow/compute/exec.cc:649  kernel_->signature->out_type().Resolve(kernel_ctx_, args.inputs)
# /arrow/cpp/src/arrow/compute/exec/expression.cc:602  executor->Init(&kernel_context, {kernel, types, options})
# /arrow/cpp/src/arrow/compute/exec/project_node.cc:91  ExecuteScalarExpression(simplified_expr, target, plan()->exec_context())
# /arrow/cpp/src/arrow/record_batch.cc:336  ReadNext(&batch)
# /arrow/cpp/src/arrow/record_batch.cc:350  ToRecordBatches()

It works fine if you don't call as_record_batch_reader() in the middle. Oddly, it also works fine if you add as_adq() (aka collapse()) after head() and before evaluating to table--that is, if you run it through an ExecPlan again, it doesn't error.

Reporter: Neal Richardson / @nealrichardson
Assignee: Dewey Dunnington / @paleolimbot

PRs and other links:

• GitHub Pull Request #14518

_{Note: This issue was originally created as ARROW-18101. Please see the migration documentation for further details.}

[asfimport]

Dewey Dunnington / @paleolimbot:
After some sleuthing, it seems that's because an ExecPlanReader is collected to a table explicitly using RunWithCapturedR:

arrow/r/src/compute-exec.cpp

Lines 191 to 198 in 5984d8a

	// [[arrow::export]]
	std::shared_ptr<arrow::Table> Table__from_ExecPlanReader(
	const std::shared_ptr<arrow::RecordBatchReader>& reader) {
	auto result = RunWithCapturedRIfPossible<std::shared_ptr<arrow::Table>>(
	[&]() { return reader->ToTable(); });
	return ValueOrStop(result);
	}

...but a regular RecordBatchReader (like the one created by head.RecordBatchReader()) is not:

arrow/r/src/recordbatchreader.cpp

Lines 113 to 117 in 5984d8a

	// [[arrow::export]]
	std::shared_ptr<arrow::Table> Table__from_RecordBatchReader(
	const std::shared_ptr<arrow::RecordBatchReader>& reader) {
	return ValueOrStop(reader->ToTable());
	}

I believe this was on purpose to limit number of places where RunWithCapturedR() was used since it might be causing problems in valgrind/autobrew nightlies (ARROW-17879). Ideally, RunWithCapturedR() causes no problems and it's safe to use it to collect any RecordBatchReader into a table.

[asfimport]

Dewey Dunnington / @paleolimbot:
Issue resolved by pull request 14518
#14518