[R] Use WriteNode to write queries

[asfimport]

Following ARROW-13542. Any query that has a join or an aggregation currently has to first evaluate the query and hold it in memory before creating a Scanner to write it. We could improve that by using a WriteNode inside write_dataset() (and maybe that improves the other cases too, or at least allows us to delete some code).

Reporter: Neal Richardson / @nealrichardson
Assignee: Neal Richardson / @nealrichardson

Related issues:

• [R] Use WriteNode in write_dataset() (duplicates)

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

[asfimport]

Dewey Dunnington / @paleolimbot:
I'd be happy to take a look at this but need a bit more background on what changes you envision in (approximately) which parts of the code.

Some example code with a simple join and aggregation + write_dataset:

library(arrow, warn.conflicts = FALSE)
library(dplyr, warn.conflicts = FALSE)

df1 <- data.frame(a = letters, b = 1:26)
df2 <- data.frame(b = 1:5, c = LETTERS[1:5])

tf1 <- tempfile()
tf2 <- tempfile()

record_batch(df2) %>% 
  left_join(df1) %>% 
  write_dataset(tf1)

open_dataset(tf1) %>% 
  collect()
#>   b c a
#> 1 1 A a
#> 2 2 B b
#> 3 3 C c
#> 4 4 D d
#> 5 5 E e


record_batch(df1) %>% 
  summarise(col = mean(b)) %>% 
  write_dataset(tf2)

open_dataset(tf2) %>% 
  collect()
#> # A tibble: 1 × 1
#>     col
#>   <dbl>
#> 1  13.5

[asfimport]

Jonathan Keane / @jonkeane:
I'm sure it'll end up being more complicated than this, but the first step would seem to be replacing the scanner bits inside of write_dataset (https://github.com/apache/arrow/blob/master/r/R/dataset-write.R#L126-L134, and then also in dataset___Dataset_write adding something like MakeWriteNode https://github.com/apache/arrow/pull/11017/files#diff-2caf4e9bd3f139e05e55dca80725d8a9c436f5ccf65c76a37cebfa6ee9b36a6aR411-R422 )

https://github.com/apache/arrow/blob/master/r/R/query-engine.R has a lot of the other ExecNode implementation for creating the exec plans and running them and https://github.com/apache/arrow/blob/master/r/src/compute-exec.cpp has the C++ that links to libarrow

[asfimport]

Weston Pace / @westonpace:
I think write_dataset's first argument changes from dataset to dataset_or_whatever_we_call_a_dplyr_query and then if it is a dataset keep the existing logic but if it is the other thing then slap a write node on the end and call something like do_exec_plan