ARROW-3814: [R] RecordBatch$from_arrays()

r/R/RecordBatch.R

@@ -91,10 +91,13 @@
 
 #' Create an [arrow::RecordBatch][arrow__RecordBatch] from a data frame
 #'
-#' @param .data a data frame
+#' @param ... A variable number of arrow::Array
+#' @param schema a arrow::Schema
 #'
 #' @return a [arrow::RecordBatch][arrow__RecordBatch]
 #' @export
-record_batch <- function(.data){
-  shared_ptr(`arrow::RecordBatch`, RecordBatch__from_dataframe(.data))
+record_batch <- function(..., schema = NULL){
+  arrays <- tibble::lst(...)
+  stopifnot(length(arrays) > 0)
+  shared_ptr(`arrow::RecordBatch`, RecordBatch__from_arrays(schema, arrays))
 }

r/R/Schema.R

@@ -61,7 +61,7 @@
 #'
 #' @export
 schema <- function(...){
-  shared_ptr(`arrow::Schema`, schema_(.fields(list(...))))
+  shared_ptr(`arrow::Schema`, schema_(.fields(list2(...))))
 }
 
 #' read a Schema from a stream

r/R/Table.R

@@ -53,11 +53,16 @@
 
 #' Create an arrow::Table from a data frame
 #'
-#' @param .data a data frame
+#' @param ... arrays, chunked arrays, or R vectors
+#' @param schema a schema. The default (`NULL`) infers the schema from the `...`
+#'
+#' @return an arrow::Table
 #'
 #' @export
-table <- function(.data){
-  shared_ptr(`arrow::Table`, Table__from_dataframe(.data))
+table <- function(..., schema = NULL){
+  dots <- tibble::lst(...)
+  stopifnot(length(dots) > 0)
+  shared_ptr(`arrow::Table`, Table__from_dots(dots, schema))
 }
 
 #' @export

r/R/feather.R

@@ -72,7 +72,12 @@ write_feather.default <- function(data, stream) {
 
 #' @export
 write_feather.data.frame <- function(data, stream) {
-  write_feather(record_batch(data), stream)
+  # splice the columns in the record_batch() call
+  # e.g. if we had data <- data.frame(x = <...>, y = <...>)
+  # then record_batch(!!!data) is the same as
+  # record_batch(x = data$x, y = data$y)
+  # see ?rlang::list2()
+  write_feather(record_batch(!!!data), stream)
 }
 
 #' @method write_feather arrow::RecordBatch

r/R/write_arrow.R

@@ -21,7 +21,10 @@ to_arrow <- function(x) {
 
 `to_arrow.arrow::RecordBatch` <- function(x) x
 `to_arrow.arrow::Table` <- function(x) x
-`to_arrow.data.frame` <- function(x) table(x)
+
+# splice the data frame as arguments of table()
+# see ?rlang::list2()
+`to_arrow.data.frame` <- function(x) table(!!!x)
 
 #' serialize an [arrow::Table][arrow__Table], an [arrow::RecordBatch][arrow__RecordBatch], or a
 #'   data frame to either the streaming format or the binary file format

r/lint.sh

@@ -33,4 +33,4 @@ CPPLINT=$CPP_BUILD_SUPPORT/cpplint.py
 $CPP_BUILD_SUPPORT/run_cpplint.py \
     --cpplint_binary=$CPPLINT \
     --exclude_glob=$CPP_BUILD_SUPPORT/lint_exclusions.txt \
-    --source_dir=$SOURCE_DIR/src --quiet $1
+    --source_dir=$SOURCE_DIR/src --quiet

r/src/array_from_vector.cpp

@@ -270,8 +270,7 @@ class VectorConverter {
   virtual Status Ingest(SEXP obj) = 0;
 
   virtual Status GetResult(std::shared_ptr<arrow::Array>* result) {
-    RETURN_NOT_OK(builder_->Finish(result));
-    return Status::OK();
+    return builder_->Finish(result);
   }
 
   ArrayBuilder* builder() const { return builder_; }
@@ -299,7 +298,7 @@ struct Unbox<Type, enable_if_integer<Type>> {
           return IngestRange<int64_t>(builder, reinterpret_cast<int64_t*>(REAL(obj)),
                                       XLENGTH(obj), NA_INT64);
         }
-      // TODO: handle aw and logical
+      // TODO: handle raw and logical
       default:
         break;
     }
@@ -881,7 +880,7 @@ std::shared_ptr<Array> MakeSimpleArray(SEXP x) {
   }
 
   auto data = ArrayData::Make(std::make_shared<Type>(), LENGTH(x), std::move(buffers),
-                              null_count, 0);
+                              null_count, 0 /*offset*/);
 
   // return the right Array class
   return std::make_shared<typename TypeTraits<Type>::ArrayType>(data);

r/src/arrow_types.h

@@ -19,6 +19,7 @@
 
 #include <limits>
 #include <memory>
+#include <vector>
 
 #include <RcppCommon.h>
 
@@ -35,11 +36,12 @@
 #include <arrow/type.h>
 #include <arrow/util/compression.h>
 
-#define STOP_IF_NOT(TEST, MSG)    \
-  do {                            \
-    if (!(TEST)) Rcpp::stop(MSG); \
+#define STOP_IF(TEST, MSG)     \
+  do {                         \
+    if (TEST) Rcpp::stop(MSG); \
   } while (0)
 
+#define STOP_IF_NOT(TEST, MSG) STOP_IF(!(TEST), MSG)
 #define STOP_IF_NOT_OK(s) STOP_IF_NOT(s.ok(), s.ToString())
 
 template <typename T>
@@ -178,8 +180,7 @@ inline constexpr Rbyte default_value<RAWSXP>() {
 SEXP ChunkedArray__as_vector(const std::shared_ptr<arrow::ChunkedArray>& chunked_array);
 SEXP Array__as_vector(const std::shared_ptr<arrow::Array>& array);
 std::shared_ptr<arrow::Array> Array__from_vector(SEXP x, SEXP type);
-std::shared_ptr<arrow::RecordBatch> RecordBatch__from_dataframe(Rcpp::DataFrame tbl);
-std::shared_ptr<arrow::DataType> Array__infer_type(SEXP x);
+std::shared_ptr<arrow::RecordBatch> RecordBatch__from_arrays(SEXP, SEXP);
 
 namespace arrow {
 namespace r {
@@ -207,5 +208,18 @@ inline std::shared_ptr<T> extract(SEXP x) {
   return Rcpp::ConstReferenceSmartPtrInputParameter<std::shared_ptr<T>>(x);
 }
 
+template <typename T>
+std::vector<std::shared_ptr<T>> list_to_shared_ptr_vector(SEXP lst) {
+  R_xlen_t n = XLENGTH(lst);
+  std::vector<std::shared_ptr<T>> res(n);
+  for (R_xlen_t i = 0; i < n; i++) {
+    res[i] = extract<T>(VECTOR_ELT(lst, i));
+  }
+  return res;
+}
+
+std::shared_ptr<arrow::Array> Array__from_vector(
+    SEXP x, const std::shared_ptr<arrow::DataType>& type, bool type_infered);
+
 }  // namespace r
 }  // namespace arrow