ARROW-3490: [R] streaming of arrow objects to streams

r/DESCRIPTION

@@ -24,10 +24,12 @@ Imports:
     vctrs (>= 0.0.0.9000),
     fs,
     tibble,
-    crayon
+    crayon,
+    withr
 Remotes:
     r-lib/vctrs, 
-    RcppCore/Rcpp
+    RcppCore/Rcpp, 
+    romainfrancois/withr@bug-79/defer
 Roxygen: list(markdown = TRUE)
 RoxygenNote: 6.1.0.9000
 Suggests:
@@ -43,6 +45,8 @@ Collate:
     'List.R'
     'RcppExports.R'
     'RecordBatch.R'
+    'RecordBatchReader.R'
+    'RecordBatchWriter.R'
     'Schema.R'
     'Struct.R'
     'Table.R'
@@ -51,5 +55,6 @@ Collate:
     'dictionary.R'
     'io.R'
     'memory_pool.R'
+    'on_exit.R'
     'reexports-tibble.R'
     'zzz.R'

r/NAMESPACE

@@ -13,19 +13,43 @@ S3method(buffer,numeric)
 S3method(buffer,raw)
 S3method(buffer_reader,"arrow::Buffer")
 S3method(buffer_reader,default)
+S3method(fixed_size_buffer_writer,"arrow::Buffer")
+S3method(fixed_size_buffer_writer,default)
 S3method(length,"arrow::Array")
 S3method(names,"arrow::RecordBatch")
 S3method(print,"arrow-enum")
 S3method(read_record_batch,"arrow::io::BufferReader")
 S3method(read_record_batch,"arrow::io::RandomAccessFile")
+S3method(read_record_batch,"arrow::ipc::RecordBatchFileReader")
+S3method(read_record_batch,"arrow::ipc::RecordBatchStreamReader")
 S3method(read_record_batch,character)
 S3method(read_record_batch,fs_path)
 S3method(read_record_batch,raw)
 S3method(read_table,"arrow::io::BufferReader")
 S3method(read_table,"arrow::io::RandomAccessFile")
+S3method(read_table,"arrow::ipc::RecordBatchFileReader")
+S3method(read_table,"arrow::ipc::RecordBatchStreamReader")
 S3method(read_table,character)
 S3method(read_table,fs_path)
 S3method(read_table,raw)
+S3method(record_batch_file_reader,"arrow::io::RandomAccessFile")
+S3method(record_batch_file_reader,character)
+S3method(record_batch_file_reader,fs_path)
+S3method(record_batch_stream_reader,"arrow::io::InputStream")
+S3method(record_batch_stream_reader,raw)
+S3method(write_arrow,"arrow::RecordBatch")
+S3method(write_arrow,"arrow::Table")
+S3method(write_arrow,data.frame)
+S3method(write_record_batch,"arrow::io::OutputStream")
+S3method(write_record_batch,"arrow::ipc::RecordBatchWriter")
+S3method(write_record_batch,character)
+S3method(write_record_batch,fs_path)
+S3method(write_record_batch,raw)
+S3method(write_table,"arrow::io::OutputStream")
+S3method(write_table,"arrow::ipc::RecordBatchWriter")
+S3method(write_table,character)
+S3method(write_table,fs_path)
+S3method(write_table,raw)
 export(DateUnit)
 export(FileMode)
 export(StatusCode)
@@ -35,13 +59,16 @@ export(array)
 export(as_tibble)
 export(boolean)
 export(buffer)
+export(buffer_output_stream)
 export(buffer_reader)
 export(chunked_array)
 export(date32)
 export(date64)
 export(decimal)
 export(dictionary)
 export(file_open)
+export(file_output_stream)
+export(fixed_size_buffer_writer)
 export(float16)
 export(float32)
 export(float64)
@@ -52,11 +79,16 @@ export(int8)
 export(list_of)
 export(mmap_create)
 export(mmap_open)
+export(mock_output_stream)
 export(null)
 export(read_arrow)
 export(read_record_batch)
 export(read_table)
 export(record_batch)
+export(record_batch_file_reader)
+export(record_batch_file_writer)
+export(record_batch_stream_reader)
+export(record_batch_stream_writer)
 export(schema)
 export(struct)
 export(table)
@@ -69,17 +101,16 @@ export(uint64)
 export(uint8)
 export(utf8)
 export(write_arrow)
+export(write_record_batch)
+export(write_table)
 importFrom(R6,R6Class)
 importFrom(Rcpp,sourceCpp)
 importFrom(assertthat,assert_that)
 importFrom(glue,glue)
 importFrom(purrr,map)
 importFrom(purrr,map2)
-importFrom(purrr,map_chr)
 importFrom(purrr,map_int)
 importFrom(rlang,dots_n)
-importFrom(rlang,quo_name)
-importFrom(rlang,seq2)
-importFrom(rlang,set_names)
 importFrom(tibble,as_tibble)
+importFrom(withr,defer_parent)
 useDynLib(arrow, .registration = TRUE)

r/R/R6.R

@@ -40,6 +40,10 @@
     },
     pointer_address = function(){
       Object__pointer_address(self$pointer())
+    },
+
+    is_null = function(){
+      Object__is_null(self)
     }
   )
 )

r/R/RecordBatch.R

@@ -22,8 +22,6 @@
     num_columns = function() RecordBatch__num_columns(self),
     num_rows = function() RecordBatch__num_rows(self),
     schema = function() `arrow::Schema`$new(RecordBatch__schema(self)),
-    to_file = function(path) invisible(RecordBatch__to_file(self, fs::path_abs(path))),
-    to_stream = function() RecordBatch__to_stream(self),
     column = function(i) `arrow::Array`$new(RecordBatch__column(self, i)),
     column_name = function(i) RecordBatch__column_name(self, i),
     names = function() RecordBatch__names(self),
@@ -40,7 +38,9 @@
       } else {
         `arrow::RecordBatch`$new(RecordBatch__Slice2(self, offset, length))
       }
-    }
+    },
+
+    serialize = function(output_stream, ...) write_record_batch(self, output_stream, ...)
   )
 )
 
@@ -67,43 +67,3 @@
 record_batch <- function(.data){
   `arrow::RecordBatch`$new(RecordBatch__from_dataframe(.data))
 }
-
-#' Read a single record batch from a stream
-#'
-#' @param stream input stream
-#'
-#' @details `stream` can be a `arrow::io::RandomAccessFile` stream as created by [file_open()] or [mmap_open()] or a path.
-#'
-#' @export
-read_record_batch <- function(stream){
-  UseMethod("read_record_batch")
-}
-
-#' @export
-read_record_batch.character <- function(stream){
-  assert_that(length(stream) == 1L)
-  read_record_batch(fs::path_abs(stream))
-}
-
-#' @export
-read_record_batch.fs_path <- function(stream){
-  stream <- file_open(stream); on.exit(stream$Close())
-  read_record_batch(stream)
-}
-
-#' @export
-`read_record_batch.arrow::io::RandomAccessFile` <- function(stream){
-  `arrow::RecordBatch`$new(read_record_batch_RandomAccessFile(stream))
-}
-
-#' @export
-`read_record_batch.arrow::io::BufferReader` <- function(stream){
-  `arrow::RecordBatch`$new(read_record_batch_BufferReader(stream))
-}
-
-#' @export
-read_record_batch.raw <- function(stream){
-  stream <- buffer_reader(stream); on.exit(stream$Close())
-  read_record_batch(stream)
-}
-