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

[boshek]

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:

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 s3dir<- 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

[github-actions]

Thanks for opening a pull request!

If this is not a minor PR. Could you open an issue for this pull request on JIRA? https://issues.apache.org/jira/browse/ARROW

Opening JIRAs ahead of time contributes to the Openness of the Apache Arrow project.

Then could you also rename pull request title in the following format?

ARROW-${JIRA_ID}: [${COMPONENT}] ${SUMMARY}

or

MINOR: [${COMPONENT}] ${SUMMARY}

See also:

• Other pull requests
• Contribution Guidelines - How to contribute patches

[github-actions]

https://issues.apache.org/jira/browse/ARROW-16144

[github-actions]

⚠️ Ticket has not been started in JIRA, please click 'Start Progress'.

[nealrichardson]

This is a nice addition

[nealrichardson]

One thing to watch out here: sometimes people name their parquet files something.parquet.snappy, but you wouldn't use a CompressedOutputStream for that, you'd pass the compression option to the parquet writer itself. I would guess that the make_readable_file() path handles this already, maybe that can be a model (or maybe it doesn't and needs to).

[boshek]

So for the parquet.snappy or even snappy.parquet I think it works because "snappy" isn't included here:

arrow/r/R/io.R

Lines 325 to 330 in 3df2e05

	switch(tools::file_ext(path),
	bz2="bz2",
	gz="gzip",
	lz4="lz4",
	zst="zstd",
	"uncompressed"

But if someone tried something like this we do get an error that isn't super informative. I think this is outside this PR so could the resolution here be to open another ticket for this specifically?

library(arrow, warn.conflicts=FALSE)
tf<- tempfile(fileext=".parquet.gz")
write_parquet(data.frame(x=1:5), tf, compression="gzip", compression_level=5)
read_parquet(tf)
#> Error: file must be a "RandomAccessFile"

[nealrichardson]

Sure, this fails on master too so ok to make a separate JIRA (please link to it here when you make it).

[ursabot]

Benchmark runs are scheduled for baseline = ce4dcbd and contender = d2cbe9e. d2cbe9e is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Finished ⬇️0.0% ⬆️0.0%] ec2-t3-xlarge-us-east-2
[Failed ⬇️1.29% ⬆️0.0%] test-mac-arm
[Failed ⬇️0.0% ⬆️0.0%] ursa-i9-9960x
[Finished ⬇️0.28% ⬆️0.08%] ursa-thinkcentre-m75q
Buildkite builds:
[Finished] d2cbe9e0 ec2-t3-xlarge-us-east-2
[Failed] d2cbe9e0 test-mac-arm
[Failed] d2cbe9e0 ursa-i9-9960x
[Finished] d2cbe9e0 ursa-thinkcentre-m75q
[Finished] ce4dcbdf ec2-t3-xlarge-us-east-2
[Failed] ce4dcbdf test-mac-arm
[Failed] ce4dcbdf ursa-i9-9960x
[Finished] ce4dcbdf ursa-thinkcentre-m75q
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java