ARROW-9869: [R] Implement full S3FileSystem/S3Options constructor

Also adds tests that run on minio, which I've tested locally, added documentation for how to run them, and added them to the macOS ("autobrew") nightly test run (because the Linux builds still don't have aws-sdk-cpp support).

Closes apache#8197 from nealrichardson/r-s3-options

Authored-by: Neal Richardson <neal.p.richardson@gmail.com>
Signed-off-by: Neal Richardson <neal.p.richardson@gmail.com>

dev/tasks/homebrew-formulae/travis.osx.r.yml

@@ -45,9 +45,11 @@ before_install:
 - sed -i.bak -E -e 's@https://github.com/apache/arrow.git"$@{{ arrow.remote }}.git", :revision => "{{ arrow.head }}"@' tools/apache-arrow.rb && rm -f tools/apache-arrow.rb.bak
 # Sometimes crossbow gives a remote URL with .git and sometimes not. Make sure there's only one
 - sed -i.bak -E -e 's@.git.git@.git@' tools/apache-arrow.rb && rm -f tools/apache-arrow.rb.bak
+# Get minio for S3 testing
+- brew install minio/stable/minio
 script:
-- Rscript -e 'install.packages("rcmdcheck")'
+- Rscript -e 'install.packages(c("rcmdcheck", "sys"))'
 # Note that this is not --as-cran. CRAN doesn't do macOS checks --as-cran
-- travis_wait Rscript -e "rcmdcheck::rcmdcheck(build_args = '--no-build-vignettes', args = c('--no-manual', '--ignore-vignettes', '--run-donttest'), error_on = 'warning', check_dir = 'check')"
+- travis_wait Rscript -e "minio_dir <- tempfile(); dir.create(minio_dir); pid <- sys::exec_background('minio', c('server', minio_dir)); on.exit(tools::pskill(pid)); rcmdcheck::rcmdcheck(build_args = '--no-build-vignettes', args = c('--no-manual', '--ignore-vignettes', '--run-donttest'), error_on = 'warning', check_dir = 'check')"
 # If there's a build failure, it's probably in this log. Let's print it regardless though
 - cat check/arrow.Rcheck/00install.out

docs/source/python/filesystems.rst

@@ -43,7 +43,7 @@ and Amazon S3-compatible storage (:class:`S3FileSystem`).
 Usage
 -----
 
-A FileSystem object can be created with one of the constuctors (and check the
+A FileSystem object can be created with one of the constructors (and check the
 respective constructor for its options)::
 
    >>> from pyarrow import fs

r/NAMESPACE

@@ -72,6 +72,7 @@ S3method(mean,Scalar)
 S3method(min,Array)
 S3method(min,ChunkedArray)
 S3method(names,Dataset)
+S3method(names,FeatherReader)
 S3method(names,RecordBatch)
 S3method(names,ScannerBuilder)
 S3method(names,Schema)
@@ -287,6 +288,7 @@ importFrom(rlang,enquos)
 importFrom(rlang,env)
 importFrom(rlang,env_bind)
 importFrom(rlang,eval_tidy)
+importFrom(rlang,exec)
 importFrom(rlang,is_false)
 importFrom(rlang,is_integerish)
 importFrom(rlang,list2)
@@ -309,6 +311,7 @@ importFrom(tidyselect,vars_rename)
 importFrom(tidyselect,vars_select)
 importFrom(utils,head)
 importFrom(utils,install.packages)
+importFrom(utils,modifyList)
 importFrom(utils,tail)
 importFrom(vctrs,s3_register)
 importFrom(vctrs,vec_cast)

r/NEWS.md

@@ -28,7 +28,7 @@
 ## AWS S3 support
 
 * S3 support is now enabled in binary macOS and Windows (Rtools40 only, i.e. R >= 4.0) packages. To enable it on Linux, you will need to build and install `aws-sdk-cpp` from source, then set the environment variable `EXTRA_CMAKE_FLAGS="-DARROW_S3=ON -DAWSSDK_SOURCE=SYSTEM"` prior to building the R package (with bundled C++ build, not with Arrow system libraries) from source.
-* File readers and writers (`read_parquet()`, `write_feather()`, et al.) now accept an `s3://` URI as the source or destination file, as do `open_dataset()` and `write_dataset()`. See `vignette("fs", package = "arrow")` for details.
+* File readers and writers (`read_parquet()`, `write_feather()`, et al.), as well as `open_dataset()` and `write_dataset()`, allow you to access resources on S3 (or on file systems that emulate S3) either by providing an `s3://` URI or by passing an additional `filesystem` argument. See `vignette("fs", package = "arrow")` for details.
 
 ## Computation
 

r/R/arrow-package.R

@@ -18,7 +18,7 @@
 #' @importFrom R6 R6Class
 #' @importFrom purrr as_mapper map map2 map_chr map_dfr map_int map_lgl
 #' @importFrom assertthat assert_that is.string
-#' @importFrom rlang list2 %||% is_false abort dots_n warn enquo quo_is_null enquos is_integerish quos eval_tidy new_data_mask syms env env_bind as_label set_names
+#' @importFrom rlang list2 %||% is_false abort dots_n warn enquo quo_is_null enquos is_integerish quos eval_tidy new_data_mask syms env env_bind as_label set_names exec
 #' @importFrom tidyselect vars_select
 #' @useDynLib arrow, .registration = TRUE
 #' @keywords internal

r/R/csv.R

@@ -68,6 +68,8 @@
 #' parsing options provided in other arguments (e.g. `delim`, `quote`, etc.).
 #' @param convert_options see [file reader options][CsvReadOptions]
 #' @param read_options see [file reader options][CsvReadOptions]
+#' @param filesystem A [FileSystem] where `file` can be found if it is a
+#' string file path; default is the local file system
 #' @param as_data_frame Should the function return a `data.frame` (default) or
 #' an Arrow [Table]?
 #'
@@ -98,6 +100,7 @@ read_delim_arrow <- function(file,
                              parse_options = NULL,
                              convert_options = NULL,
                              read_options = NULL,
+                             filesystem = NULL,
                              as_data_frame = TRUE) {
 
   if (is.null(parse_options)) {
@@ -119,7 +122,7 @@ read_delim_arrow <- function(file,
   }
 
   if (!inherits(file, "InputStream")) {
-    file <- make_readable_file(file)
+    file <- make_readable_file(file, filesystem = filesystem)
     on.exit(file$close())
   }
   reader <- CsvTableReader$create(
@@ -206,7 +209,7 @@ read_tsv_arrow <- function(file,
 #' The `CsvTableReader$create()` and `JsonTableReader$create()` factory methods
 #' take the following arguments:
 #'
-#' - `file` A character path to a local file, or an Arrow input stream
+#' - `file` An Arrow [InputStream]
 #' - `convert_options` (CSV only), `parse_options`, `read_options`: see
 #'    [CsvReadOptions]
 #' - `...` additional parameters.
@@ -227,7 +230,7 @@ CsvTableReader$create <- function(file,
                                   parse_options = CsvParseOptions$create(),
                                   convert_options = CsvConvertOptions$create(),
                                   ...) {
-  file <- make_readable_file(file)
+  assert_is(file, "InputStream")
   shared_ptr(
     CsvTableReader,
     csv___TableReader__Make(file, read_options, parse_options, convert_options)

r/R/feather.R

@@ -25,6 +25,8 @@
 #'
 #' @param x `data.frame`, [RecordBatch], or [Table]
 #' @param sink A string file path, URI, or [OutputStream]
+#' @param filesystem A [FileSystem] where `sink` should be written if it is a
+#' string file path; default is the local file system
 #' @param version integer Feather file version. Version 2 is the current.
 #' Version 1 is the more limited legacy format.
 #' @param chunk_size For V2 files, the number of rows that each chunk of data
@@ -52,6 +54,7 @@
 #' @include arrow-package.R
 write_feather <- function(x,
                           sink,
+                          filesystem = NULL,
                           version = 2,
                           chunk_size = 65536L,
                           compression = c("default", "lz4", "uncompressed", "zstd"),
@@ -106,7 +109,7 @@ write_feather <- function(x,
   assert_is(x, "Table")
 
   if (is.string(sink)) {
-    sink <- make_output_stream(sink)
+    sink <- make_output_stream(sink, filesystem)
     on.exit(sink$close())
   }
   assert_is(sink, "OutputStream")
@@ -141,17 +144,16 @@ write_feather <- function(x,
 #' # Can select columns
 #' df <- read_feather(tf, col_select = starts_with("d"))
 #' }
-read_feather <- function(file, col_select = NULL, as_data_frame = TRUE, ...) {
+read_feather <- function(file, col_select = NULL, as_data_frame = TRUE, filesystem = NULL, ...) {
   if (!inherits(file, "RandomAccessFile")) {
-    file <- make_readable_file(file)
+    file <- make_readable_file(file, filesystem = filesystem)
     on.exit(file$close())
   }
   reader <- FeatherReader$create(file, ...)
 
-  all_columns <- ipc___feather___Reader__column_names(reader)
   col_select <- enquo(col_select)
   columns <- if (!quo_is_null(col_select)) {
-    vars_select(all_columns, !!col_select)
+    vars_select(names(reader), !!col_select)
   }
 
   out <- reader$Read(columns)
@@ -198,10 +200,14 @@ FeatherReader <- R6Class("FeatherReader", inherit = ArrowObject,
   ),
   active = list(
     # versions are officially 2 for V1 and 3 for V2 :shrug:
-    version = function() ipc___feather___Reader__version(self) - 1L
+    version = function() ipc___feather___Reader__version(self) - 1L,
+    column_names = function() ipc___feather___Reader__column_names(self)
   )
 )
 
+#' @export
+names.FeatherReader <- function(x) x$column_names
+
 FeatherReader$create <- function(file, mmap = TRUE, ...) {
   assert_is(file, "RandomAccessFile")
   shared_ptr(FeatherReader, ipc___feather___Reader__Open(file))

r/R/filesystem.R

@@ -122,11 +122,39 @@ FileSelector$create <- function(base_dir, allow_not_found = FALSE, recursive = F
 #'
 #' @section Factory:
 #'
-#' The `$create()` factory methods instantiate the `FileSystem` object and
-#' take the following arguments, depending on the subclass:
+#' `LocalFileSystem$create()` returns the object and takes no arguments.
 #'
-#' - no argument is needed for instantiating a `LocalFileSystem`
-#' - `base_path` and `base_fs` for instantiating a `SubTreeFileSystem`
+#' `SubTreeFileSystem$create()` takes the following arguments:
+#'
+#' - `base_path`, a string path
+#' - `base_fs`, a `FileSystem` object
+#'
+#' `S3FileSystem$create()` optionally takes arguments:
+#'
+#' - `anonymous`: logical, default `FALSE`. If true, will not attempt to look up
+#'    credentials using standard AWS configuration methods.
+#' - `access_key`, `secret_key`: authentication credentials. If one is provided,
+#'    the other must be as well. If both are provided, they will override any
+#'    AWS configuration set at the environment level.
+#' - `session_token`: optional string for authentication along with
+#'    `access_key` and `secret_key`
+#' - `role_arn`: string AWS ARN of an AccessRole. If provided instead of `access_key` and
+#'    `secret_key`, temporary credentials will be fetched by assuming this role.
+#' - `session_name`: optional string identifier for the assumed role session.
+#' - `external_id`: optional unique string identifier that might be required
+#'    when you assume a role in another account.
+#' - `load_frequency`: integer, frequency (in seconds) with which temporary
+#'    credentials from an assumed role session will be refreshed. Default is
+#'    900 (i.e. 15 minutes)
+#' - `region`: AWS region to connect to. If omitted, the AWS library will
+#'    provide a sensible default based on client configuration, falling back
+#'    to "us-east-1" if no other alternatives are found.
+#' - `endpoint_override`: If non-empty, override region with a connect string
+#'    such as "localhost:9000". This is useful for connecting to file systems
+#'    that emulate S3.
+#' - `scheme`: S3 connection transport (default "https")
+#' - `background_writes`: logical, whether `OutputStream` writes will be issued
+#'    in the background, without blocking (default `TRUE`)
 #'
 #' @section Methods:
 #'
@@ -279,13 +307,56 @@ LocalFileSystem$create <- function() {
 #' @usage NULL
 #' @format NULL
 #' @rdname FileSystem
+#' @importFrom utils modifyList
 #' @export
 S3FileSystem <- R6Class("S3FileSystem", inherit = FileSystem)
-S3FileSystem$create <- function() {
-  fs___EnsureS3Initialized()
-  shared_ptr(S3FileSystem, fs___S3FileSystem__create())
+S3FileSystem$create <- function(anonymous = FALSE, ...) {
+  args <- list2(...)
+  if (anonymous) {
+    invalid_args <- intersect(c("access_key", "secret_key", "session_token", "role_arn", "session_name", "external_id", "load_frequency"), names(args))
+    if (length(invalid_args)) {
+      stop("Cannot specify ", oxford_paste(invalid_args), " when anonymous = TRUE", call. = FALSE)
+    }
+  } else {
+    keys_present <- length(intersect(c("access_key", "secret_key"), names(args)))
+    if (keys_present == 1) {
+      stop("Key authentication requires both access_key and secret_key", call. = FALSE)
+    }
+    if ("session_token" %in% names(args) && keys_present != 2) {
+      stop(
+        "In order to initialize a session with temporary credentials, ",
+        "both secret_key and access_key must be provided ",
+        "in addition to session_token.",
+        call. = FALSE
+      )
+    }
+    arn <- "role_arn" %in% names(args)
+    if (keys_present == 2 && arn) {
+      stop("Cannot provide both key authentication and role_arn", call. = FALSE)
+    }
+    arn_extras <- intersect(c("session_name", "external_id", "load_frequency"), names(args))
+    if (length(arn_extras) > 0 && !arn) {
+      stop("Cannot specify ", oxford_paste(arn_extras), " without providing a role_arn string", call. = FALSE)
+    }
+  }
+  args <- c(modifyList(default_s3_options, args), anonymous = anonymous)
+  shared_ptr(S3FileSystem, exec(fs___S3FileSystem__create, !!!args))
 }
 
+default_s3_options <- list(
+  access_key = "",
+  secret_key = "",
+  session_token = "",
+  role_arn = "",
+  session_name = "",
+  external_id = "",
+  load_frequency = 900L,
+  region = "",
+  endpoint_override = "",
+  scheme = "",
+  background_writes = TRUE
+)
+
 arrow_with_s3 <- function() {
   .Call(`_s3_available`)
 }
@@ -295,9 +366,12 @@ arrow_with_s3 <- function() {
 #' @rdname FileSystem
 #' @export
 SubTreeFileSystem <- R6Class("SubTreeFileSystem", inherit = FileSystem)
-SubTreeFileSystem$create <- function(base_path, base_fs) {
-  xp <- fs___SubTreeFileSystem__create(clean_path_rel(base_path), base_fs)
-  shared_ptr(SubTreeFileSystem, xp)
+SubTreeFileSystem$create <- function(base_path, base_fs = NULL) {
+  fs_and_path <- get_path_and_filesystem(base_path, base_fs)
+  shared_ptr(
+    SubTreeFileSystem,
+    fs___SubTreeFileSystem__create(fs_and_path$path, fs_and_path$fs)
+  )
 }
 
 #' Copy files, including between FileSystems

r/R/io.R

@@ -257,12 +257,16 @@ make_readable_file <- function(file, mmap = TRUE, compression = NULL, filesystem
   file
 }
 
-make_output_stream <- function(x) {
+make_output_stream <- function(x, filesystem = NULL) {
   if (is_url(x)) {
     fs_and_path <- FileSystem$from_uri(x)
-    fs_and_path$fs$OpenOutputStream(fs_and_path$path)
-  } else {
+    filesystem = fs_and_path$fs
+    x <- fs_and_path$path
+  }
+  if (is.null(filesystem)) {
     FileOutputStream$create(x)
+  } else {
+    filesystem$OpenOutputStream(x)
   }
 }
 

r/R/ipc_stream.R

@@ -35,13 +35,13 @@
 #' serialize data to a buffer.
 #' [RecordBatchWriter] for a lower-level interface.
 #' @export
-write_ipc_stream <- function(x, sink, ...) {
+write_ipc_stream <- function(x, sink, filesystem = NULL, ...) {
   x_out <- x # So we can return the data we got
   if (is.data.frame(x)) {
     x <- Table$create(x)
   }
   if (is.string(sink)) {
-    sink <- make_output_stream(sink)
+    sink <- make_output_stream(sink, filesystem)
     on.exit(sink$close())
   }
   assert_is(sink, "OutputStream")
@@ -90,16 +90,18 @@ write_to_raw <- function(x, format = c("stream", "file")) {
 #' open.
 #' @param as_data_frame Should the function return a `data.frame` (default) or
 #' an Arrow [Table]?
+#' @param filesystem A [FileSystem] where `file` can be found if it is a
+#' string file path; default is the local file system
 #' @param ... extra parameters passed to `read_feather()`.
 #'
 #' @return A `data.frame` if `as_data_frame` is `TRUE` (the default), or an
 #' Arrow [Table] otherwise
 #' @seealso [read_feather()] for writing IPC files. [RecordBatchReader] for a
 #' lower-level interface.
 #' @export
-read_ipc_stream <- function(file, as_data_frame = TRUE, ...) {
+read_ipc_stream <- function(file, as_data_frame = TRUE, filesystem = NULL, ...) {
   if (!inherits(file, "InputStream")) {
-    file <- make_readable_file(file)
+    file <- make_readable_file(file, filesystem = filesystem)
     on.exit(file$close())
   }
 

r/R/json.R

@@ -35,7 +35,15 @@
 #'   ', tf, useBytes=TRUE)
 #'   df <- read_json_arrow(tf)
 #' }
-read_json_arrow <- function(file, col_select = NULL, as_data_frame = TRUE, ...) {
+read_json_arrow <- function(file,
+                            col_select = NULL,
+                            as_data_frame = TRUE,
+                            filesystem = NULL,
+                            ...) {
+  if (!inherits(file, "InputStream")) {
+    file <- make_readable_file(file, filesystem = filesystem)
+    on.exit(file$close())
+  }
   tab <- JsonTableReader$create(file, ...)$Read()
 
   col_select <- enquo(col_select)
@@ -64,8 +72,7 @@ JsonTableReader$create <- function(file,
                                    read_options = JsonReadOptions$create(),
                                    parse_options = JsonParseOptions$create(),
                                    ...) {
-
-  file <- make_readable_file(file)
+  assert_is(file, "InputStream")
   shared_ptr(
     JsonTableReader,
     json___TableReader__Make(file, read_options, parse_options)