[R] Refactor do_exec_plan to return a RecordBatchReader

[asfimport]

Right now do_exec_plan returns an Arrow table because head, tail, and arrange do. If ARROW-14289 is completed and similar work is done for arrange, we may be able to alter do_exec_plan to return a RBR instead.

The map_batches() implementation (ARROW-14029) could benefit from this refactor. And it might make ARROW-15040 more useful.

Reporter: Will Jones / @wjones127
Assignee: Neal Richardson / @nealrichardson

Related issues:

• [R] Repair map_batches() (is related to)
• [R] Enable write_csv_arrow to take Dataset or arrow_dplyr_query as input (is related to)

PRs and other links:

• GitHub Pull Request #13170

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

[asfimport]

Dewey Dunnington / @paleolimbot:
Just collecting a few related code comments here:

• arrow/r/R/dataset-scan.R

  Line 89 in 03219e2

  # TODO: Is there a way to get a RecordBatchReader rather than evaluating?

• arrow/r/R/query-engine.R

  Lines 23 to 26 in 03219e2

  	# TODO (ARROW-14289): make the head/tail methods return RBR not Table
  	if (inherits(tab, "RecordBatchReader")) {
  	tab <- tab$read_table()
  	}

• arrow/r/R/dataset-scan.R

  Line 184 in 03219e2

  # TODO: ARROW-15271 possibly refactor do_exec_plan to return a RecordBatchReader

  Related is the ability to write files directly in a query plan using the WriteNode that was added in ARROW-13542. For example, there is a ticket open for using the WriteNode to write data sets (ARROW-14266). Writing files is useful but perhaps orthogonal to the ability to iterate over a RecordBatchReader, which is exemplified by the revamped map_batches() + vignette addition.

[asfimport]

Dewey Dunnington / @paleolimbot:
Just connecting ARROW-15489, since an improved RecordBatchReader class will make this more useful!

[asfimport]

Neal Richardson / @nealrichardson:
The underlying issue is that there are some operations that can't be fully supported with the current ExecPlan:

• sorting on a temporary expression (e.g. arrange(ds, x * y)): you have to project with the expression in it, collect the sorted data, and then drop the derived column, but if you drop the column using select() and run it through the ExecPlan again, you lose your sorting because sorting currently only happens in the last step (SinkNode)

• arrange %>% tail: it's implemented as a topK operation, so for head you get data in the right order, but for tail, it's done as reversing the sort and taking topK (i.e. there's no bottomK that returns in that order). So you have to re-sort the result. This could I guess be done with another ExecPlan that re-sorts, so that could yield a RBR, though awkwardly.

• ARROW-14289: head.RecordBatchReader returns Table not RBR, seems easily fixable. Also would be nice if ExecPlan had a limit node, or the SinkNode took a limit, or something.

  I'm not sure what the performance impact would be in these cases if we were to compute into a Table, do whatever finishing steps, and push back into a RBR, which in most cases is just going to be pulled back into a Table in R. But maybe these are sufficiently uncommon scenarios that we shouldn't let them shape our API to the extent that they are.

  We also would need to pass the R metadata into ExecPlan_run to attach to the output_schema in compute::MakeGeneratorReader because it does not appear that you can modify the schema of an RBR.

  P.S. I haven't checked whether there are open JIRAs for all of those ExecPlan issues but there probably should be.

[asfimport]

Neal Richardson / @nealrichardson:
Issue resolved by pull request 13170
#13170