feat(r): support target catalog/schema for ingestion

Author: eitsupi

r/adbcdrivermanager/R/helpers.R

@@ -30,6 +30,10 @@
 #' @param mode One of `"create"`, `"append"`, `"replace"`, `"create_append"` (error if the schema
 #'   is not compatible or append otherwise), or `"default"` (use the `adbc.ingest.mode`
 #'   argument of [adbc_statement_init()]). The default is `"default"`.
+#' @param catalog_name If not `NULL`, the catalog to create/locate the table in.
+#'   **This API is EXPERIMENTAL.**
+#' @param db_schema_name If not `NULL`, the schema to create/locate the table in.
+#'   **This API is EXPERIMENTAL.**
 #' @param query An SQL query
 #' @param bind A data.frame, nanoarrow_array, or nanoarrow_array_stream of
 #'   bind parameters or NULL to skip the bind/prepare step.
@@ -69,7 +73,9 @@ execute_adbc <- function(db_or_con, query, ..., bind = NULL) {
 #' @export
 write_adbc <- function(tbl, db_or_con, target_table, ...,
                        mode = c("default", "create", "append", "replace", "create_append"),
-                       temporary = FALSE) {
+                       temporary = FALSE,
+                       catalog_name = NULL,
+                       db_schema_name = NULL) {
   UseMethod("write_adbc", db_or_con)
 }
 
@@ -113,7 +119,9 @@ execute_adbc.default <- function(db_or_con, query, ..., bind = NULL, stream = NU
 #' @export
 write_adbc.default <- function(tbl, db_or_con, target_table, ...,
                                mode = c("default", "create", "append", "replace", "create_append"),
-                               temporary = FALSE) {
+                               temporary = FALSE,
+                               catalog_name = NULL,
+                               db_schema_name = NULL) {
   assert_adbc(db_or_con, c("adbc_database", "adbc_connection"))
   mode <- match.arg(mode)
 
@@ -127,6 +135,8 @@ write_adbc.default <- function(tbl, db_or_con, target_table, ...,
   stmt <- adbc_statement_init(
     con,
     adbc.ingest.target_table = target_table,
+    adbc.ingest.target_catalog = catalog_name,
+    adbc.ingest.target_db_schema = db_schema_name,
     adbc.ingest.mode = if (!identical(mode, "default")) paste0("adbc.ingest.mode.", mode),
     adbc.ingest.temporary = if (temporary) "true"
   )

r/adbcpostgresql/R/adbcpostgresql-package.R

@@ -34,6 +34,8 @@ NULL
 #' @param adbc.connection.autocommit Use FALSE to disable the default
 #'   autocommit behaviour.
 #' @param adbc.ingest.target_table The name of the target table for a bulk insert.
+#' @param adbc.ingest.target_db_schema The schema of the table for a bulk insert.
+#'   **This API is EXPERIMENTAL.**
 #' @param adbc.ingest.mode Whether to create (the default) or append.
 #'
 #' @return An [adbcdrivermanager::adbc_driver()]
@@ -78,10 +80,12 @@ adbc_connection_init.adbcpostgresql_database <- function(database, ...,
 #' @export
 adbc_statement_init.adbcpostgresql_connection <- function(connection, ...,
                                                       adbc.ingest.target_table = NULL,
+                                                      adbc.ingest.target_db_schema = NULL,
                                                       adbc.ingest.mode = NULL) {
   options <- list(
     ...,
     adbc.ingest.target_table = adbc.ingest.target_table,
+    adbc.ingest.target_db_schema = adbc.ingest.target_db_schema,
     adbc.ingest.mode = adbc.ingest.mode
   )
 

r/adbcpostgresql/tests/testthat/test-adbcpostgres-package.R

@@ -88,3 +88,28 @@ test_that("default options can open a database and execute a query", {
 
   adbcdrivermanager::adbc_statement_release(stmt)
 })
+
+test_that("write_adbc() supports db_schema_name", {
+  test_db_uri <- Sys.getenv("ADBC_POSTGRESQL_TEST_URI", "")
+  skip_if(identical(test_db_uri, ""))
+
+  db <- adbc_database_init(adbcpostgresql(), uri = test_db_uri)
+  con <- adbc_connection_init(db)
+  on.exit({
+    adbcdrivermanager::adbc_connection_release(con)
+    adbcdrivermanager::adbc_database_release(db)
+  })
+
+  adbcdrivermanager::execute_adbc(con, "CREATE SCHEMA IF NOT EXISTS testschema")
+  adbcdrivermanager::execute_adbc(con, "DROP TABLE IF EXISTS testschema.df_schema")
+
+  df <- data.frame(x = as.double(1:3))
+  expect_identical(
+    adbcdrivermanager::write_adbc(df, con, "df_schema", db_schema_name = "testschema", mode = "create"),
+    df
+  )
+
+  stream <- adbcdrivermanager::read_adbc(con, "SELECT * FROM testschema.df_schema ORDER BY x")
+  expect_identical(as.data.frame(stream), df)
+  stream$release()
+})

r/adbcsqlite/R/adbcsqlite-package.R

@@ -33,6 +33,8 @@ NULL
 #' @param adbc.connection.autocommit Use FALSE to disable the default
 #'   autocommit behaviour.
 #' @param adbc.ingest.target_table The name of the target table for a bulk insert.
+#' @param adbc.ingest.target_catalog The catalog of the table for a bulk insert.
+#'   **This API is EXPERIMENTAL.**
 #' @param adbc.ingest.mode Whether to create (the default) or append.
 #' @param adbc.sqlite.query.batch_rows The number of rows per batch to return.
 #'
@@ -78,11 +80,13 @@ adbc_connection_init.adbcsqlite_database <- function(database, ...,
 #' @export
 adbc_statement_init.adbcsqlite_connection <- function(connection, ...,
                                                       adbc.ingest.target_table = NULL,
+                                                      adbc.ingest.target_catalog = NULL,
                                                       adbc.ingest.mode = NULL,
                                                       adbc.sqlite.query.batch_rows = NULL) {
   options <- list(
     ...,
     adbc.ingest.target_table = adbc.ingest.target_table,
+    adbc.ingest.target_catalog = adbc.ingest.target_catalog,
     adbc.ingest.mode = adbc.ingest.mode,
     adbc.sqlite.query.batch_rows = adbc.sqlite.query.batch_rows
   )

r/adbcsqlite/tests/testthat/test-adbcsqlite-package.R

@@ -99,6 +99,25 @@ test_that("read/write/execute SQL work with sqlite connections", {
   adbcdrivermanager::adbc_database_release(db)
 })
 
+test_that("write_adbc() supports catalog_name", {
+  db <- adbc_database_init(adbcsqlite())
+  con <- adbc_connection_init(db)
+  on.exit({
+    adbcdrivermanager::adbc_connection_release(con)
+    adbcdrivermanager::adbc_database_release(db)
+  })
+
+  df <- data.frame(x = as.double(1:3))
+  expect_identical(
+    adbcdrivermanager::write_adbc(df, con, "df_catalog", catalog_name = "main", mode = "replace"),
+    df
+  )
+
+  stream <- adbcdrivermanager::read_adbc(con, "SELECT * from main.df_catalog")
+  expect_identical(as.data.frame(stream), df)
+  stream$release()
+})
+
 test_that("write_adbc() with temporary = TRUE works with sqlite databases", {
   skip_if_not(packageVersion("adbcdrivermanager") >= "0.6.0.9000")