Skip to content

Commit

Permalink
Replace build_copy_queries() with dm_sql()
Browse files Browse the repository at this point in the history
  • Loading branch information
krlmlr committed Dec 21, 2023
1 parent 3c8d712 commit 600d7c0
Show file tree
Hide file tree
Showing 5 changed files with 151 additions and 157 deletions.
175 changes: 72 additions & 103 deletions R/db-interface.R
Original file line number Diff line number Diff line change
Expand Up @@ -5,14 +5,13 @@
#' and a [`dm`] object as its second argument.
#' The latter is copied to the former.
#' The default is to create temporary tables, set `temporary = FALSE` to create permanent tables.
#' Unless `set_key_constraints` is `FALSE`, primary key constraints are set on all databases,
#' and in addition foreign key constraints are set on MSSQL and Postgres databases.
#' Unless `set_key_constraints` is `FALSE`, primary key, foreign key, and unique constraints
#' are set on all databases.
#'
#' @inheritParams dm_examine_constraints
#'
#' @param dest An object of class `"src"` or `"DBIConnection"`.
#' @param dm A `dm` object.
#' @inheritParams rlang::args_dots_empty
#' @param set_key_constraints If `TRUE` will mirror `dm` primary and foreign key constraints on a database
#' and create indexes for foreign key constraints.
#' Set to `FALSE` if your data model currently does not satisfy primary or foreign key constraints.
Expand All @@ -23,38 +22,9 @@
#' `table_names` is not `NULL`.
#'
#' Not all DBMS are supported.
#' @param table_names Desired names for the tables on `dest`; the names within the `dm` remain unchanged.
#' Can be `NULL`, a named character vector, or a vector of [DBI::Id] objects.
#'
#' If left `NULL` (default), the names will be determined automatically depending on the `temporary` argument:
#'
#' 1. `temporary = TRUE` (default): unique table names based on the names of the tables in the `dm` are created.
#' 1. `temporary = FALSE`: the table names in the `dm` are used as names for the tables on `dest`.
#'
#' If a function or one-sided formula, `table_names` is converted to a function
#' using [rlang::as_function()].
#' This function is called with the unquoted table names of the `dm` object
#' as the only argument.
#' The output of this function is processed by [DBI::dbQuoteIdentifier()],
#' that result should be a vector of identifiers of the same length
#' as the original table names.
#'
#' Use a variant of
#' `table_names = ~ DBI::SQL(paste0("schema_name", ".", .x))`
#' to specify the same schema for all tables.
#' Use `table_names = identity` with `temporary = TRUE`
#' to avoid giving temporary tables unique names.
#'
#' If a named character vector,
#' the names of this vector need to correspond to the table names in the `dm`,
#' and its values are the desired names on `dest`.
#' The value is processed by [DBI::dbQuoteIdentifier()],
#' that result should be a vector of identifiers of the same length
#' as the original table names.
#'
#' Use qualified names corresponding to your database's syntax
#' to specify e.g. database and schema for your tables.
#' @param unique_table_names,copy_to Must be `NULL`.
#' @inheritParams dm_sql
#' @inheritParams rlang::args_dots_empty
#' @param unique_table_names,copy_to Deprecated.
#'
#' @family DB interaction functions
#'
Expand Down Expand Up @@ -100,14 +70,10 @@ copy_dm_to <- function(
# 3. implement the key situation within our `dm` on the DB

if (!is.null(unique_table_names)) {
deprecate_warn(
deprecate_stop(
"0.1.4", "dm::copy_dm_to(unique_table_names = )",
details = "Use `table_names = identity` to use unchanged names for temporary tables."
details = "Use `table_names = set_names(names(dm))` to use unchanged names for temporary tables."
)

if (is.null(table_names) && temporary && !unique_table_names) {
table_names <- identity
}
}

if (!is.null(copy_to)) {
Expand All @@ -124,113 +90,116 @@ copy_dm_to <- function(
check_suggested("dbplyr", "copy_dm_to")

dest <- src_from_src_or_con(dest)
src_names <- src_tbls_impl(dm)

if (is_db(dest)) {
dest_con <- con_from_src_or_con(dest)

# in case `table_names` was chosen by the user, check if the input makes sense:
# 1. is there one name per dm-table?
# 2. are there any duplicated table names?
# 3. is it a named character or ident_q vector with the correct names?
if (is.null(table_names)) {
table_names_out <- repair_table_names_for_db(src_names, temporary, dest_con, schema)
# https://github.com/tidyverse/dbplyr/issues/487
if (is_mssql(dest)) {
temporary <- FALSE
}
} else {
if (!is.null(schema)) abort_one_of_schema_table_names()
if (is_function(table_names) || is_bare_formula(table_names)) {
table_name_fun <- as_function(table_names)
table_names_out <- set_names(table_name_fun(src_names), src_names)
} else {
table_names_out <- table_names
}
check_naming(names(table_names_out), src_names)

if (anyDuplicated(table_names_out)) {
problem <- table_names_out[duplicated(table_names_out)][[1]]
abort_copy_dm_to_table_names_duplicated(problem)
}

names(table_names_out) <- src_names
}
} else {
# FIXME: Other data sources than local and database possible
deprecate_warn(
"0.1.6", "dm::copy_dm_to(dest = 'must refer to a remote data source')",
if (!is_db(dest)) {
deprecate_stop(
"0.1.6", "dm::copy_dm_to(dest = 'must refer to a DBI connection')",
"dm::collect.dm()"
)
table_names_out <- set_names(src_names)
}

# FIXME: if same_src(), can use compute() but need to set NOT NULL and other
# constraints
src_names <- src_tbls_impl(dm)
dest_con <- con_from_src_or_con(dest)

# in case `table_names` was chosen by the user, check if the input makes sense:
# 1. is there one name per dm-table?
# 2. are there any duplicated table names?
# 3. is it a named character or ident_q vector with the correct names?
if (is.null(table_names)) {
table_names_out <- repair_table_names_for_db(src_names, temporary, dest_con, schema)
# https://github.com/tidyverse/dbplyr/issues/487
if (is_mssql(dest)) {
temporary <- FALSE
}
} else {
if (!is.null(schema)) abort_one_of_schema_table_names()
if (is_function(table_names) || is_bare_formula(table_names)) {
table_name_fun <- as_function(table_names)
table_names_out <- set_names(table_name_fun(src_names), src_names)
} else {
table_names_out <- table_names
}
check_naming(names(table_names_out), src_names)

if (anyDuplicated(table_names_out)) {
problem <- table_names_out[duplicated(table_names_out)][[1]]
abort_copy_dm_to_table_names_duplicated(problem)
}

# Shortcut necessary to avoid copying into .GlobalEnv
if (!is_db(dest)) {
return(dm)
names(table_names_out) <- src_names
}

table_names_out <- ddl_check_table_names(table_names_out, dm)

# Must be done here because table types may depend on string length, #2066
dm <- collect(dm, progress = progress)

queries <- build_copy_queries(dest_con, dm, set_key_constraints, temporary, table_names_out)
if (isTRUE(set_key_constraints)) {
dm_for_sql <- dm
} else {
def_no_keys <- dm_get_def(dm)
def_no_keys$uks[] <- list(new_uk())
def_no_keys$fks[] <- list(new_fk())
# Must keep primary keys
dm_for_sql <- dm_from_def(def_no_keys)
}

sql <- dm_sql(dm_for_sql, dest_con, table_names_out, temporary)

# FIXME: Extract function
# FIXME: Make descriptions part of the dm_sql() output

ticker_create <- new_ticker(
pre <- unlist(sql$pre)
load <- unlist(sql$load)
post <- unlist(sql$post)

ticker_pre <- new_ticker(
"creating tables",
n = length(queries$sql_table),
n = length(pre),
progress = progress,
top_level_fun = "copy_dm_to"
)

# create tables
walk(queries$sql_table, ticker_create(~ {
walk(pre, ticker_pre(~ {
DBI::dbExecute(dest_con, .x, immediate = TRUE)
}))

ticker_populate <- new_ticker(
ticker_load <- new_ticker(
"populating tables",
n = length(queries$name),
n = length(load),
progress = progress,
top_level_fun = "copy_dm_to"
)

# populate tables
pwalk(
queries[c("name", "remote_name")],
ticker_populate(~ db_append_table(
con = dest_con,
remote_table = .y,
table = dm[[.x]],
progress = progress,
autoinc = dm_get_all_pks(dm, table = !!.x)$autoincrement
))
)
walk(load, ticker_load(~ {
DBI::dbExecute(dest_con, .x, immediate = TRUE)
}))

ticker_index <- new_ticker(
ticker_post <- new_ticker(
"creating indexes",
n = sum(lengths(queries$sql_index)),
n = length(post),
progress = progress,
top_level_fun = "copy_dm_to"
)

# create indexes
walk(unlist(queries$sql_index), ticker_index(~ {
walk(post, ticker_post(~ {
DBI::dbExecute(dest_con, .x, immediate = TRUE)
}))

# remote dm is same as source dm with replaced data
# FIXME: Extract function
def <- dm_get_def(dm)

remote_tables <- map2(
table_names_out,
map(def$data, colnames),
~ tbl(dest_con, ..1, vars = ..2)
~ tbl(dest_con, .x, vars = .y)
)

def$data <- unname(remote_tables[names(dm)])
def$data <- unname(remote_tables)
remote_dm <- dm_from_def(def)

invisible(debug_dm_validate(remote_dm))
Expand Down
41 changes: 6 additions & 35 deletions man/copy_dm_to.Rd

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

9 changes: 9 additions & 0 deletions tests/testthat/_snaps/db-interface.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,12 @@
# copy_dm_to() copies data frames from any source

Code
copy_dm_to(default_local_src(), dm_for_filter())
Condition
Error:
! The `dest` argument of `copy_dm_to()` must refer to a DBI connection as of dm 0.1.6.
i Please use `collect.dm()` instead.

# copy_dm_to() rejects overwrite and types arguments

Code
Expand Down
Loading

0 comments on commit 600d7c0

Please sign in to comment.