You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by np...@apache.org on 2022/05/18 21:25:03 UTC
[arrow] branch master updated: ARROW-16144: [R] Write compressed data streams (particularly over S3)
This is an automated email from the ASF dual-hosted git repository.
npr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/arrow.git
The following commit(s) were added to refs/heads/master by this push:
new d2cbe9e0e2 ARROW-16144: [R] Write compressed data streams (particularly over S3)
d2cbe9e0e2 is described below
commit d2cbe9e0e2ce206fba71d3d171babe36bada1a9d
Author: Sam Albers <sa...@gmail.com>
AuthorDate: Wed May 18 14:24:55 2022 -0700
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 <sa...@gmail.com>
Co-authored-by: Neal Richardson <ne...@gmail.com>
Signed-off-by: Neal Richardson <ne...@gmail.com>
---
r/R/io.R | 23 ++++++++++++++++++-----
r/R/util.R | 4 ++++
r/tests/testthat/test-csv.R | 17 +++++++++++++++++
r/tests/testthat/test-s3-minio.R | 20 ++++++++++++++++++++
4 files changed, 59 insertions(+), 5 deletions(-)
diff --git a/r/R/io.R b/r/R/io.R
index 379dcf6f35..8e72187b43 100644
--- a/r/R/io.R
+++ b/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",
diff --git a/r/R/util.R b/r/R/util.R
index ff2bb070b8..4aff69e471 100644
--- a/r/R/util.R
+++ b/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")
+}
diff --git a/r/tests/testthat/test-csv.R b/r/tests/testthat/test-csv.R
index 631e75fd74..8e463d3abe 100644
--- a/r/tests/testthat/test-csv.R
+++ b/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()
diff --git a/r/tests/testthat/test-s3-minio.R b/r/tests/testthat/test-s3-minio.R
index 24f8495ae2..51db355783 100644
--- a/r/tests/testthat/test-s3-minio.R
+++ b/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(