ARROW-16144: [R] Write compressed data streams (particularly over S3)

This PR enables reading/writing compressed data streams over s3 and locally and adds some tests to test some of those round trips. For the filesystem path I had to do a little regex on the string for compression detection but any feedback on alternative approaches is very welcome. Previously supplying a file with a compression extension wrote out an uncompressed file. Here is a reprex of the updated writing behaviour:

```r
library(arrow, warn.conflicts = FALSE)
## local
write_csv_arrow(mtcars, file = file)
write_csv_arrow(mtcars, file = comp_file)
file.size(file)
[1] 1303
file.size(comp_file)
[1] 567

## or with s3
dir <- tempfile()
dir.create(dir)
subdir <- file.path(dir, "bucket")
dir.create(subdir)

minio_server <- processx::process$new("minio", args = c("server", dir), supervise = TRUE)
Sys.sleep(2)
stopifnot(minio_server$is_alive())

s3_uri <- "s3://minioadmin:minioadmin@?scheme=http&endpoint_override=localhost%3A9000"
bucket <- s3_bucket(s3_uri)

write_csv_arrow(mtcars, bucket$path("bucket/data.csv.gz"))
write_csv_arrow(mtcars, bucket$path("bucket/data.csv"))

file.size(file.path(subdir, "data.csv.gz"))
[1] 567
file.size(file.path(subdir, "data.csv"))
[1] 1303
```

Closes #13183 from boshek/ARROW-16144

Lead-authored-by: Sam Albers <sam.albers@gmail.com>
Co-authored-by: Neal Richardson <neal.p.richardson@gmail.com>
Signed-off-by: Neal Richardson <neal.p.richardson@gmail.com>

Author: boshek

r/R/io.R

@@ -270,7 +270,7 @@ make_readable_file <- function(file, mmap = TRUE, compression = NULL, filesystem
       file <- ReadableFile$create(file)
     }
 
-    if (!identical(compression, "uncompressed")) {
+    if (is_compressed(compression)) {
       file <- CompressedInputStream$create(file, compression)
     }
   } else if (inherits(file, c("raw", "Buffer"))) {
@@ -292,7 +292,7 @@ make_readable_file <- function(file, mmap = TRUE, compression = NULL, filesystem
   file
 }
 
-make_output_stream <- function(x, filesystem = NULL) {
+make_output_stream <- function(x, filesystem = NULL, compression = NULL) {
   if (inherits(x, "connection")) {
     if (!isOpen(x)) {
       open(x, "wb")
@@ -309,11 +309,21 @@ make_output_stream <- function(x, filesystem = NULL) {
     filesystem <- fs_and_path$fs
     x <- fs_and_path$path
   }
+
+  if (is.null(compression)) {
+    # Infer compression from sink
+    compression <- detect_compression(x)
+  }
+
   assert_that(is.string(x))
-  if (is.null(filesystem)) {
-    FileOutputStream$create(x)
+  if (is.null(filesystem) && is_compressed(compression)) {
+    CompressedOutputStream$create(x) ##compressed local
+  } else if (is.null(filesystem) && !is_compressed(compression)) {
+    FileOutputStream$create(x) ## uncompressed local
+  } else if (!is.null(filesystem) && is_compressed(compression)) {
+    CompressedOutputStream$create(filesystem$OpenOutputStream(x)) ## compressed remote
   } else {
-    filesystem$OpenOutputStream(x)
+    filesystem$OpenOutputStream(x) ## uncompressed remote
   }
 }
 
@@ -322,6 +332,9 @@ detect_compression <- function(path) {
     return("uncompressed")
   }
 
+  # Remove any trailing slashes, which FileSystem$from_uri may add
+  path <- gsub("/$", "", path)
+
   switch(tools::file_ext(path),
     bz2 = "bz2",
     gz = "gzip",

r/R/util.R

@@ -211,3 +211,7 @@ handle_csv_read_error <- function(e, schema, call) {
   }
   abort(msg, call = call)
 }
+
+is_compressed <- function(compression) {
+  !identical(compression, "uncompressed")
+}

r/tests/testthat/test-csv.R

@@ -564,6 +564,23 @@ test_that("write_csv_arrow can write from RecordBatchReader objects", {
   expect_equal(nrow(tbl_in), 3)
 })
 
+test_that("read/write compressed file successfully", {
+  skip_if_not_available("gzip")
+  tfgz <- tempfile(fileext = ".csv.gz")
+  tf <- tempfile(fileext = ".csv")
+  on.exit(unlink(tf))
+  on.exit(unlink(tfgz))
+
+  write_csv_arrow(tbl, tf)
+  write_csv_arrow(tbl, tfgz)
+  expect_lt(file.size(tfgz), file.size(tf))
+
+  expect_identical(
+    read_csv_arrow(tfgz),
+    tbl
+  )
+})
+
 test_that("read_csv_arrow() can read sub-second timestamps with col_types T setting (ARROW-15599)", {
   tbl <- tibble::tibble(time = c("2018-10-07 19:04:05.000", "2018-10-07 19:04:05.001"))
   tf <- tempfile()

r/tests/testthat/test-s3-minio.R

@@ -54,6 +54,26 @@ if (arrow_with_s3() && process_is_running("minio server")) {
     )
   })
 
+  test_that("read/write compressed csv by filesystem", {
+    skip_if_not_available("gzip")
+    dat <- tibble(x = seq(1, 10, by = 0.2))
+    write_csv_arrow(dat, fs$path(minio_path("test.csv.gz")))
+    expect_identical(
+      read_csv_arrow(fs$path(minio_path("test.csv.gz"))),
+      dat
+    )
+  })
+
+  test_that("read/write csv by filesystem", {
+    skip_if_not_available("gzip")
+    dat <- tibble(x = seq(1, 10, by = 0.2))
+    write_csv_arrow(dat, fs$path(minio_path("test.csv")))
+    expect_identical(
+      read_csv_arrow(fs$path(minio_path("test.csv"))),
+      dat
+    )
+  })
+
   test_that("read/write stream", {
     write_ipc_stream(example_data, fs$path(minio_path("test3.ipc")))
     expect_identical(