Arrow IPC serializers and parsers (#968)

Author: JosiahParry

DESCRIPTION

@@ -98,5 +98,5 @@ Collate:
     'validate_api_spec.R'
     'zzz.R'
 Language: en-US
-Config/Needs/check: Cairo
+Config/Needs/check: Cairo, r-quantities/units
 Config/Needs/website: tidyverse/tidytemplate

NAMESPACE

@@ -26,6 +26,7 @@ export(include_md)
 export(include_rmd)
 export(is_plumber)
 export(options_plumber)
+export(parser_arrow_ipc_stream)
 export(parser_csv)
 export(parser_excel)
 export(parser_feather)
@@ -76,6 +77,7 @@ export(registered_serializers)
 export(serializer_agg_jpeg)
 export(serializer_agg_png)
 export(serializer_agg_tiff)
+export(serializer_arrow_ipc_stream)
 export(serializer_bmp)
 export(serializer_cat)
 export(serializer_content_type)

NEWS.md

@@ -1,5 +1,7 @@
 # plumber (development version)
 
+* Added support for [Arrow IPC Streams](https://arrow.apache.org/docs/format/Columnar.html#serialization-and-interprocess-communication-ipc) (@josiahparry #968)
+
 # plumber 1.3.0
 
 * The port many now be specified as an environment variable. User-provided ports must be between 1024 and 49151 (following [IANA guidelines](https://www.iana.org/assignments/service-names-port-numbers/service-names-port-numbers.xhtml)) and may not be a known unsafe port. plumber will now throw an error if an invalid port is requested. (@shikokuchuo @gadenbuie #963)
@@ -20,6 +22,7 @@
 
 * Added `operationId` to each operation within the auto-generated OpenAPI output. The value is similar to the `PATH-VERB`, e.g. `/users/create-POST`. (#986)
 
+* Added support for graphic devices provided by ragg and svglite (@thomasp85 #964)
 
 # plumber 1.2.2
 

R/parse-body.R

@@ -496,6 +496,17 @@ parser_feather <- function(...) {
   }
 }
 
+#' @describeIn parsers Arrow IPC parser. See [arrow::read_ipc_stream()] for more details.
+#' @export
+parser_arrow_ipc_stream <- function(...) {
+  parser_read_file(function(tmpfile) {
+    if (!requireNamespace("arrow", quietly = TRUE)) {
+      stop("`arrow` must be installed for `parser_arrow_ipc_stream` to work")
+    }
+    arrow::read_ipc_stream(tmpfile, ...)
+  })
+}
+
 #' @describeIn parsers parquet parser. See [arrow::read_parquet()] for more details.
 #' @export
 parser_parquet <- function(...) {
@@ -623,6 +634,7 @@ register_parsers_onLoad <- function() {
   register_parser("rds",     parser_rds,     fixed = "application/rds")
   register_parser("feather", parser_feather, fixed = c("application/vnd.apache.arrow.file", "application/feather"))
   register_parser("parquet", parser_parquet, fixed = "application/vnd.apache.parquet")
+  register_parser("arrow_ipc_stream", parser_arrow_ipc_stream, fixed = "application/vnd.apache.arrow.stream")
   register_parser("excel",   parser_excel,   fixed = c("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", "application/vnd.ms-excel"))
   register_parser("text",    parser_text,    fixed = "text/plain", regex = "^text/")
   register_parser("tsv",     parser_tsv,     fixed = c("application/tab-separated-values", "text/tab-separated-values"))

R/serializer.R

@@ -291,6 +291,21 @@ serializer_feather <- function(type = "application/vnd.apache.arrow.file") {
   )
 }
 
+#' @describeIn serializers Arrow IPC serializer. See also: [arrow::write_ipc_stream()]
+#' @export
+serializer_arrow_ipc_stream <- function(type = "application/vnd.apache.arrow.stream") {
+  if (!requireNamespace("arrow", quietly = TRUE)) {
+    stop("`arrow` must be installed for `serializer_arrow_ipc_stream` to work")
+  }
+  serializer_write_file(
+    fileext = "",
+    type = type,
+    write_fn = function(val, tmpfile) {
+      arrow::write_ipc_stream(val, tmpfile)
+    }
+  )
+}
+
 #' @describeIn serializers parquet serializer. See also: [arrow::write_parquet()]
 #' @export
 serializer_parquet <- function(type = "application/vnd.apache.parquet") {
@@ -708,6 +723,7 @@ add_serializers_onLoad <- function() {
   register_serializer("csv",         serializer_csv)
   register_serializer("tsv",         serializer_tsv)
   register_serializer("feather",     serializer_feather)
+  register_serializer("arrow_ipc_stream",   serializer_arrow_ipc_stream)
   register_serializer("parquet",     serializer_parquet)
   register_serializer("excel",       serializer_excel)
   register_serializer("yaml",        serializer_yaml)

inst/WORDLIST

@@ -21,6 +21,7 @@ HTTPOnly
 HTTPS
 Hookable
 IANA
+IPC
 IPs
 IPv
 JSON

man/parsers.Rd

@@ -109,6 +109,26 @@ test_that("Test feather parser", {
   expect_equal(parsed, r_object)
 })
 
+test_that("Test Arrow IPC parser", {
+  skip_if_not_installed("arrow")
+
+  tmp <- tempfile()
+  on.exit({
+    file.remove(tmp)
+  }, add = TRUE)
+
+  r_object <- iris
+  arrow::write_ipc_stream(r_object, tmp)
+  val <- readBin(tmp, "raw", 10000)
+
+  parsed <- parse_body(val, "application/vnd.apache.arrow.stream", make_parser("arrow_ipc_stream"))
+  # convert from feather tibble to data.frame
+  parsed <- as.data.frame(parsed, stringsAsFactors = FALSE)
+  attr(parsed, "spec") <- NULL
+
+  expect_equal(parsed, r_object)
+})
+
 test_that("Test parquet parser", {
   skip_if_not_installed("arrow")
 

man/serializers.Rd

@@ -0,0 +1,41 @@
+context("Arrow IPC serializer")
+
+test_that("Arrow IPC serializes properly", {
+  skip_if_not_installed("arrow")
+
+  d <- data.frame(a=1, b=2, c="hi")
+  val <- serializer_arrow_ipc_stream()(d, data.frame(), PlumberResponse$new(), stop)
+  expect_equal(val$status, 200L)
+  expect_equal(val$headers$`Content-Type`, "application/vnd.apache.arrow.stream")
+
+  # can test  by doing a full round trip if we believe the parser works via `test-parse-body.R`
+  parsed <- parse_body(val$body, "application/vnd.apache.arrow.stream", make_parser("arrow_ipc_stream"))
+  # convert from feather tibble to data.frame
+  parsed <- as.data.frame(parsed, stringsAsFactors = FALSE)
+  attr(parsed, "spec") <- NULL
+
+  expect_equal(parsed, d)
+})
+
+test_that("Errors call error handler", {
+  skip_if_not_installed("arrow")
+
+  errors <- 0
+  errHandler <- function(req, res, err){
+    errors <<- errors + 1
+  }
+
+  expect_equal(errors, 0)
+  serializer_feather()(parse(text="hi"), data.frame(), PlumberResponse$new("csv"), errorHandler = errHandler)
+  expect_equal(errors, 1)
+})
+
+test_that("Errors are rendered correctly with debug TRUE", {
+  skip_if_not_installed("arrow")
+
+  pr <- pr() %>% pr_get("/", function() stop("myerror"), serializer = serializer_feather()) %>% pr_set_debug(TRUE)
+  capture.output(res <- pr$serve(make_req(pr = pr), PlumberResponse$new("csv")))
+
+  expect_match(res$body, "Error in (function () : myerror", fixed = TRUE)
+})
+

tests/testthat/test-parse-body.R

@@ -38,3 +38,4 @@ test_that("Errors are rendered correctly with debug TRUE", {
 
   expect_match(res$body, "Error in (function () : myerror", fixed = TRUE)
 })
+