ARROW-11787: [R] Implement write csv

cpp/src/arrow/csv/type_fwd.h

@@ -22,6 +22,7 @@ class TableReader;
 struct ConvertOptions;
 struct ReadOptions;
 struct ParseOptions;
+struct WriteOptions;
 
 }  // namespace csv
 }  // namespace arrow

r/NAMESPACE

@@ -122,6 +122,7 @@ export(CsvFragmentScanOptions)
 export(CsvParseOptions)
 export(CsvReadOptions)
 export(CsvTableReader)
+export(CsvWriteOptions)
 export(Dataset)
 export(DatasetFactory)
 export(DateUnit)
@@ -277,6 +278,7 @@ export(unify_schemas)
 export(utf8)
 export(value_counts)
 export(write_arrow)
+export(write_csv_arrow)
 export(write_dataset)
 export(write_feather)
 export(write_ipc_stream)

r/R/csv.R

@@ -381,6 +381,11 @@ CsvTableReader$create <- function(file,
 #' `TimestampParser$create()` takes an optional `format` string argument.
 #' See [`strptime()`][base::strptime()] for example syntax.
 #' The default is to use an ISO-8601 format parser.
+#' 
+#' The `CsvWriteOptions$create()` factory method takes the following arguments:
+#' - `include_header` Whether to write an initial header line with column names
+#' - `batch_size` Maximum number of rows processed at a time. Default is 1024.
+#' 
 #' @section Active bindings:
 #'
 #' - `column_names`: from `CsvReadOptions`
@@ -408,6 +413,19 @@ CsvReadOptions$create <- function(use_threads = option_use_threads(),
   )
 }
 
+#' @rdname CsvReadOptions
+#' @export
+CsvWriteOptions <- R6Class("CsvWriteOptions", inherit = ArrowObject)
+CsvWriteOptions$create <- function(include_header = TRUE, batch_size = 1024L){
+  assert_that(is_integerish(batch_size, n = 1, finite = TRUE), batch_size > 0)
+  csv___WriteOptions__initialize(
+    list(
+      include_header = include_header,
+      batch_size = as.integer(batch_size)
+    )
+  )
+}
+
 readr_to_csv_read_options <- function(skip, col_names, col_types) {
   if (isTRUE(col_names)) {
     # C++ default to parse is 0-length string array
@@ -585,3 +603,49 @@ readr_to_csv_convert_options <- function(na,
     include_columns = include_columns
   )
 }
+
+#' Write CSV file to disk
+#'
+#' @param x `data.frame`, [RecordBatch], or [Table]
+#' @param sink A string file path, URI, or [OutputStream], or path in a file
+#' system (`SubTreeFileSystem`)
+#' @param include_header Whether to write an initial header line with column names
+#' @param batch_size Maximum number of rows processed at a time. Default is 1024.
+#'
+#' @return The input `x`, invisibly. Note that if `sink` is an [OutputStream],
+#' the stream will be left open.
+#' @export
+#' @examples
+#' \donttest{
+#' tf <- tempfile()
+#' on.exit(unlink(tf))
+#' write_csv_arrow(mtcars, tf)
+#' }
+#' @include arrow-package.R
+write_csv_arrow <- function(x,
+                            sink,
+                            include_header = TRUE,
+                            batch_size = 1024L) {
+
+  write_options <- CsvWriteOptions$create(include_header, batch_size)
+
+  x_out <- x
+  if (is.data.frame(x)) {
+    x <- Table$create(x)
+  }
+
+  assert_is(x, "ArrowTabular")
+
+  if (!inherits(sink, "OutputStream")) {
+    sink <- make_output_stream(sink)
+    on.exit(sink$close())
+  }
+
+  if(inherits(x, "RecordBatch")){
+    csv___WriteCSV__RecordBatch(x, write_options, sink)
+  } else if(inherits(x, "Table")){
+    csv___WriteCSV__Table(x, write_options, sink)
+  }
+
+  invisible(x_out)
+}

r/_pkgdown.yml

@@ -98,6 +98,7 @@ reference:
       - write_ipc_stream
       - write_to_raw
       - write_parquet
+      - write_csv_arrow
   - title: C++ reader/writer interface
     contents:
       - ParquetFileReader
@@ -109,6 +110,7 @@ reference:
       - RecordBatchReader
       - RecordBatchWriter
       - CsvReadOptions
+      - CsvWriteOptions
   - title: Arrow data containers
     contents:
       - array

r/src/arrow_types.h

@@ -178,6 +178,7 @@ R6_CLASS_NAME(arrow::csv::ReadOptions, "CsvReadOptions");
 R6_CLASS_NAME(arrow::csv::ParseOptions, "CsvParseOptions");
 R6_CLASS_NAME(arrow::csv::ConvertOptions, "CsvConvertOptions");
 R6_CLASS_NAME(arrow::csv::TableReader, "CsvTableReader");
+R6_CLASS_NAME(arrow::csv::WriteOptions, "CsvWriteOptions");
 
 #if defined(ARROW_R_WITH_PARQUET)
 R6_CLASS_NAME(parquet::ArrowReaderProperties, "ParquetArrowReaderProperties");

r/src/csv.cpp

@@ -20,8 +20,21 @@
 #if defined(ARROW_R_WITH_ARROW)
 
 #include <arrow/csv/reader.h>
+#include <arrow/csv/writer.h>
+#include <arrow/memory_pool.h>
+
 #include <arrow/util/value_parsing.h>
 
+// [[arrow::export]]
+std::shared_ptr<arrow::csv::WriteOptions> csv___WriteOptions__initialize(
+    cpp11::list options) {
+  auto res =
+      std::make_shared<arrow::csv::WriteOptions>(arrow::csv::WriteOptions::Defaults());
+  res->include_header = cpp11::as_cpp<bool>(options["include_header"]);
+  res->batch_size = cpp11::as_cpp<int>(options["batch_size"]);
+  return res;
+}
+
 // [[arrow::export]]
 std::shared_ptr<arrow::csv::ReadOptions> csv___ReadOptions__initialize(
     cpp11::list options) {
@@ -174,4 +187,21 @@ std::shared_ptr<arrow::TimestampParser> TimestampParser__MakeISO8601() {
   return arrow::TimestampParser::MakeISO8601();
 }
 
+// [[arrow::export]]
+void csv___WriteCSV__Table(const std::shared_ptr<arrow::Table>& table,
+                           const std::shared_ptr<arrow::csv::WriteOptions>& write_options,
+                           const std::shared_ptr<arrow::io::OutputStream>& stream) {
+  StopIfNotOk(
+      arrow::csv::WriteCSV(*table, *write_options, gc_memory_pool(), stream.get()));
+}
+
+// [[arrow::export]]
+void csv___WriteCSV__RecordBatch(
+    const std::shared_ptr<arrow::RecordBatch>& record_batch,
+    const std::shared_ptr<arrow::csv::WriteOptions>& write_options,
+    const std::shared_ptr<arrow::io::OutputStream>& stream) {
+  StopIfNotOk(arrow::csv::WriteCSV(*record_batch, *write_options, gc_memory_pool(),
+                                   stream.get()));
+}
+
 #endif