Make RPresto compatible with dbplyr 2.4.0

DESCRIPTION

@@ -26,13 +26,14 @@ Imports:
     utils,
     purrr,
     dplyr (>= 0.7.0),
-    dbplyr (>= 2.0.0),
+    dbplyr (>= 2.3.3),
     tibble,
     bit64,
     rlang,
     lifecycle,
     lubridate,
-    progress
+    progress,
+    vctrs
 Suggests:
     testthat,
     hms,

NAMESPACE

@@ -22,6 +22,7 @@ S3method(get_tables_from_sql,lazy_multi_join_query)
 S3method(get_tables_from_sql,lazy_select_query)
 S3method(get_tables_from_sql,lazy_semi_join_query)
 S3method(get_tables_from_sql,lazy_set_op_query)
+S3method(get_tables_from_sql,lazy_union_query)
 S3method(sql_escape_date,PrestoConnection)
 S3method(sql_escape_datetime,PrestoConnection)
 S3method(sql_query_fields,PrestoConnection)

NEWS.md

@@ -1,5 +1,8 @@
 # RPresto 1.4.5.9000
 
+* `RPresto` is now compatible with `dbplyr` 2.4.0. We updated the minimum
+  version requirement of `dbplyr` to be 2.3.3.
+
 # RPresto 1.4.5
 
 * `compute()` uses simple table name rather than wrapped it in `as.sql()`.

R/cte.R

@@ -15,7 +15,11 @@ get_tables_from_sql.lazy_select_query <- function(query) {
 
 #' @export
 get_tables_from_sql.lazy_base_remote_query <- function(query) {
-  as.character(query$x)
+  if (inherits(query$x, "dbplyr_table_ident")) {
+    vctrs::field(query$x, "table")
+  } else {
+    as.character(query$x)
+  }
 }
 
 #' @export
@@ -35,10 +39,21 @@ get_tables_from_sql.lazy_semi_join_query <- function(query) {
 }
 
 #' @export
+# for compatibility with dbplyr < 2.4.0
 get_tables_from_sql.lazy_set_op_query <- function(query) {
   c(get_tables_from_sql(query$x), get_tables_from_sql(query$y))
 }
 
+#' @export
+# for compatibility with dbplyr >= 2.4.0
+get_tables_from_sql.lazy_union_query <- function(query) {
+  tables_from_x <- get_tables_from_sql(query$x)
+  tables_from_y <- purrr::map_chr(
+    query$unions$table, ~ get_tables_from_sql(.$lazy_query)
+  )
+  c(tables_from_x, tables_from_y)
+}
+
 is_cte_used <- function(sql) {
   startsWith(tolower(sql), "with")
 }

R/dbWriteTable.R

@@ -26,6 +26,9 @@ NULL
                           append = FALSE, field.types = NULL, temporary = FALSE,
                           row.names = FALSE, with = NULL, chunk.fields = NULL,
                           use.one.query = FALSE) {
+  force(with)
+  force(chunk.fields)
+  force(use.one.query)
   stopifnot(is.data.frame(value))
   if (!identical(temporary, FALSE)) {
     stop("Temporary tables not supported by RPresto", call. = FALSE)

R/dbplyr-src.R

@@ -186,7 +186,7 @@ copy_to.src_presto <- function(dest, df, name = deparse(substitute(df)), overwri
     )
   } else {
     df <- as.data.frame(dplyr::collect(df))
-    name <- dbplyr::db_copy_to(dest$con, name, df,
+    name <- dbplyr::db_copy_to(con = dest$con, table = name, values = df,
       overwrite = overwrite,
       types = NULL,
       temporary = FALSE,
@@ -196,7 +196,7 @@ copy_to.src_presto <- function(dest, df, name = deparse(substitute(df)), overwri
       ...
     )
     vars <- names(df)
-    out <- dplyr::tbl(dest, name, vars)
+    out <- dplyr::tbl(src = dest, from = name, vars = vars)
   }
   invisible(out)
 }

tests/testthat.R

@@ -6,6 +6,7 @@
 
 library("testthat")
 library("DBI")
+library("dplyr")
 library("RPresto")
 
 test_check("RPresto")

tests/testthat/test-cte.R

@@ -105,3 +105,51 @@ test_that("Nested CTEs work", {
     dplyr::collect(iris_presto.join_cte)
   )
 })
+
+test_that("CTEs using union work", {
+  parts <- setup_live_dplyr_connection()
+  db <- parts[["db"]]
+  tablename <- parts[["iris_table_name"]]
+  iris_presto <- dplyr::tbl(db, tablename)
+  iris_presto.virginica <- dplyr::compute(
+    dplyr::summarize(
+      dplyr::group_by(
+        dplyr::filter(
+          iris_presto,
+          species == "virginica"
+        ),
+        species
+      ),
+      mean_sepal_length = mean(sepal_length, na.rm = TRUE)
+    ),
+    name = "iris_width_virginica", cte = TRUE
+  )
+  expect_equal(db$con@session$getCTENames(), c("iris_width_virginica"))
+  iris_presto.setosa <- dplyr::compute(
+    dplyr::summarize(
+      dplyr::group_by(
+        dplyr::filter(
+          iris_presto,
+          species == "setosa"
+        ),
+        species
+      ),
+      mean_sepal_length = mean(sepal_length, na.rm = TRUE)
+    ),
+    name = "iris_width_setosa", cte = TRUE
+  )
+  expect_equal(
+    db$con@session$getCTENames(), c("iris_width_virginica", "iris_width_setosa")
+  )
+  iris_presto.union <- dplyr::compute(
+    dplyr::union_all(
+      iris_presto.virginica,
+      iris_presto.setosa
+    ),
+    name = "iris_presto_union", cte = TRUE
+  )
+  expect_equal(
+    db$con@session$getCTENames(),
+    c("iris_width_virginica", "iris_width_setosa", "iris_presto_union")
+  )
+})

tests/testthat/test-db_query_fields.R

@@ -147,6 +147,17 @@ test_that("db_query_fields works with mock", {
         ),
         next_uri = "http://localhost:8000/query_3/1",
       ),
+      mock_httr_response(
+        "http://localhost:8000/v1/statement",
+        status_code = 200,
+        state = "FINISHED",
+        # For dbplyr 2.4.0
+        request_body = paste0(
+          '^SELECT \\* FROM "empty_table" (AS )?"q[0-9]+"',
+          " WHERE 1 = 0$"
+        ),
+        next_uri = "http://localhost:8000/query_3/1",
+      ),
       mock_httr_response(
         "http://localhost:8000/v1/statement",
         status_code = 200,
@@ -175,6 +186,17 @@ test_that("db_query_fields works with mock", {
           " WHERE 1 = 0$"
         ),
         next_uri = "http://localhost:8000/query_4/1",
+      ),
+      mock_httr_response(
+        "http://localhost:8000/v1/statement",
+        status_code = 200,
+        state = "QUEUED",
+        # For dbplyr 2.4.0
+        request_body = paste0(
+          '^SELECT \\* FROM "two_columns" (AS )?"q[0-9]+"',
+          " WHERE 1 = 0$"
+        ),
+        next_uri = "http://localhost:8000/query_4/1",
       )
     ),
     `httr::GET` = mock_httr_replies(
@@ -241,8 +263,7 @@ test_that("db_query_fields works with mock", {
         dplyr::db_query_fields(
           s[["con"]],
           dplyr::ident("__non_existent_table__")
-        ),
-        "Query.*failed:.*Table .*__non_existent_table__ does not exist"
+        )
       )
     }
   )