Class name for dbplyr::in_schema() changed

Version now 0.2.0; see changelog

Author: russellpierce

DESCRIPTION

@@ -1,7 +1,7 @@
 Package: redshiftTools
 Type: Package
 Title: Redshift Tools
-Version: 0.1.2
+Version: 0.2.0
 Authors@R: c(person("Pablo", "Seibelt", email = "pabloseibelt@sicarul.com",
   role = c("aut", "cre")),person("Russell S.", "Pierce", role = c("ctb"), email = "russell.pierce@zapier.com"))
 Mantainers@R: person("Pablo", "Seibelt", email = "pabloseibelt@sicarul.com",
@@ -31,6 +31,6 @@ Imports:
 Description: Tools to upload data to an Amazon Redshift Database with good performance.
 License: MIT + file LICENSE
 LazyData: TRUE
-RoxygenNote: 7.1.0
+RoxygenNote: 7.1.1
 Suggests: testthat,
     lubridate

NAMESPACE

@@ -20,6 +20,7 @@ export(rs_table_exists)
 export(rs_unload)
 export(rs_upsert_table)
 export(sanitize_column_names_for_redshift)
+export(schema_to_character)
 export(spectrum_add_partition)
 export(spectrum_drop_partition)
 export(spectrum_list_partitions)
@@ -48,6 +49,7 @@ importFrom(dplyr,left_join)
 importFrom(dplyr,mutate_)
 importFrom(dplyr,pull)
 importFrom(dplyr,recode)
+importFrom(dplyr,rename)
 importFrom(dplyr,select)
 importFrom(dplyr,select_)
 importFrom(dplyr,tbl)

R/internal.R

@@ -1,29 +1,37 @@
+table_parts <- function(x) {
+  unlist(strsplit(as.character(x), split = ".", fixed = TRUE))
+}
+
 log_if_verbose <- function(...) {
   if (isTRUE(getOption("redshiftTools.verbose"))) {
     message("redshiftTools: ", ...)
   }
 }
 
+is_schema <- function(table_name) {
+  return(inherits(table_name, c("ident", "ident_q", "dbplyr_schema")))
+}
+
 stopifnoschema <- function(table_name) {
-  assertthat::assert_that("ident" %in% class(table_name), msg = "Table name must be result of dbplyr::in_schema()")
+  assertthat::assert_that(is_schema(table_name), msg = "Table name must be result of dbplyr::in_schema()")
 }
 
 warnifnoschema <- function(table_name) {
-  if (!"ident" %in% class(table_name)) {
+  if (!is_schema(table_name)) {
     warning(glue("No schema specified for {table_name}, will default to using public"))
     return(FALSE)
   } else {
     return(TRUE)
   }
 }
 
-#' Simple coalece
+#' Simple coalesce
 #'
-#' A simple non-type aware coalece
+#' A simple non-type aware coalesce
 #'
 #' @aliases coalesceifnull %||%
-#' @params x left position
-#' @params y right position
+#' @param x left position
+#' @param y right position
 coalesceifnull <- function(x, y) {
   return(x %||% y)
 }
@@ -36,7 +44,7 @@ boto <- reticulate::import("boto", delay_load = TRUE)
 read.text = function(pathname) {
   if (file.exists(pathname)) {
 	      return (paste(readLines(pathname), collapse="\n"))
-  } 
+  }
 }
 
 check_aws_credentials <- function() {
@@ -45,13 +53,13 @@ check_aws_credentials <- function() {
   renviron_path = "/etc/R/Renviron.site"
   rprofile_path = ".Rprofile"
   aws_key_env_set = Sys.getenv("AWS_ACCESS_KEY_ID") != ""
-  aws_configure_exists = file.exists(aws_configure_path) 
+  aws_configure_exists = file.exists(aws_configure_path)
   aws_credentials_exists = file.exists(aws_credentials_path)
   # Does not need checking as both would set env var, keeping it here for completness
   renviron_exists = file.exists(renviron_path)
   rprofile_exists = file.exists(rprofile_path)
   if (aws_key_env_set) {
-    log_if_verbose("AWS_ACCESS_KEY_ID is set, it will override other aws credentials") 	
+    log_if_verbose("AWS_ACCESS_KEY_ID is set, it will override other aws credentials")
   } else if (aws_credentials_exists && grepl("aws_access_key_id", read.text(aws_credentials_path))) {
       log_if_verbose("aws_access_key_id is set in ~/.aws/credentials, it will overrride other aws credentials")
   } else if (aws_configure_exists && grepl("aws_access_key_id", read.text(aws_configure_path))) {
@@ -168,20 +176,37 @@ queryDo <- function(dbcon, query) {
 #' @importFrom assertthat assert_that
 #' @importFrom DBI dbGetQuery
 #' @importFrom whisker whisker.render
+#' @importFrom dplyr rename
 get_table_schema <- function(dbcon, table) {
   assertthat::assert_that(length(table) <= 2)
-  assertthat::assert_that("character" %in% class(table))
   if (is.atomic(table)) {
     schema <- "public"
     target_table <- table
   } else {
-    schema <- table[1]
-    target_table <- table[2]
+    schema <- tolower(as.character(table[1]))
+    target_table <- tolower(as.character(table[2]))
   }
-  dbGetQuery(dbcon, whisker.render("SELECT *
-  FROM pg_table_def
-  WHERE tablename = '{{table_name}}'
-  AND schemaname = '{{schema}}'", list(table_name = target_table, schema = schema)))
+
+  ## pg_table_def only contains items that are in the search path
+
+  #original_search_path <- DBI::dbGetQuery(dbcon, "show search_path;")$search_path
+  ##dbGetQuery loses the single quotes around user
+  #original_search_path <- gsub("$user", "'$user'", original_search_path, fixed = TRUE)
+  #DBI::dbExecute(dbcon, whisker.render("set search_path to {{schema}}, '$user', public", list (schema = schema)))
+  ## really should be in a try block, otherwise we risk messing up the search path
+  #res <- dbGetQuery(dbcon, whisker.render("SELECT *
+  #FROM pg_table_def
+  #WHERE tablename = '{{table_name}}'
+  #AND schemaname = '{{schema}}'", list(table_name = target_table, schema = schema)))
+  #DBI::dbExecute(dbcon, whisker.render("set search_path to {{original}}", list (original = original_search_path)))
+  res <- dbGetQuery(dbcon,
+    glue::glue_sql(
+      .con = dbcon,
+      "select * from information_schema.columns where table_schema = {schema} and table_name = {target_table}"
+    )) %>%
+    rename(schemaname = table_schema, tablename = table_name, column = column_name, type = data_type) %>%
+    arrange(ordinal_position)
+  return(res)
 }
 
 #' Fix the order of columns in d to match the underlying Redshift table
@@ -238,7 +263,7 @@ fix_column_order <- function(d, dbcon, table_name, strict = TRUE) {
 #' and the result can't change quickly because Redshift resizes take quite some time.
 #'
 #' @importFrom memoise memoise timeout
-#' @params dbcon A database connection object
+#' @param dbcon A database connection object
 number_of_slices <- memoise::memoise(function(dbcon) {
   message("Getting number of slices from Redshift")
   slices <- queryDo(dbcon, "select count(1) from stv_slices")
@@ -283,22 +308,6 @@ make_dot <- function(table, schema = NULL) {
   paste0(schema, ".", table)
 }
 
-vec_to_dot <- function(x) {
-  if (length(x) == 1) {
-    schema <- "public"
-    table <- x
-  } else {
-    assertthat::assert_that(length(x) == 2)
-    schema <- x[1]
-    table <- x[2]
-  }
-  make_dot(table, schema)
-}
-
-dot_to_vec <- function(x) {
-  strsplit(x, split = ".", fixed = TRUE)[[1]]
-}
-
 
 #' Make Credentials
 #'

R/redshift-replace.R

@@ -5,7 +5,7 @@
 #'
 #' @param data a data frame
 #' @param dbcon an RPostgres connection to the redshift server
-#' @param tableName the name of the table to replace
+#' @param table_name the name of the table to replace
 #' @param split_files optional parameter to specify amount of files to split into. If not specified will look at amount of slices in Redshift to determine an optimal amount.
 #' @param bucket the name of the temporary bucket to load the data. Will look for AWS_BUCKET_NAME on environment if not specified.
 #' @param region the region of the bucket. Will look for AWS_DEFAULT_REGION on environment if not specified.
@@ -21,15 +21,15 @@
 #' host='my-redshift-url.amazon.com', port='5439',
 #' user='myuser', password='mypassword',sslmode='require')
 #'
-#' rs_replace_table(data=a, dbcon=con, tableName='testTable',
+#' rs_replace_table(data=a, dbcon=con, table_name='testTable',
 #' bucket="my-bucket", split_files=4)
 #'
 #' }
 #' @export
 #' @importFrom DBI dbExecute
 rs_replace_table <- function(data,
                              dbcon,
-                             tableName,
+                             table_name,
                              split_files,
                              bucket = Sys.getenv("AWS_BUCKET_NAME"),
                              region = Sys.getenv("AWS_DEFAULT_REGION"),
@@ -49,23 +49,23 @@ rs_replace_table <- function(data,
   # all the work. it's not a pure function!
   replace <- function(data, dbcon) {
     split_files <- min(split_files, nrow(data))
-    data <- fix_column_order(data, dbcon, table_name = tableName, strict = strict)
+    data <- fix_column_order(data, dbcon, table_name = table_name, strict = strict)
     prefix <- uploadToS3(data, bucket, split_files)
     raw_bucket <- paste0(bucket, if (Sys.getenv("ENVIRONMENT") == "production") "" else "-test")
     on.exit({
       message("Deleting temporary files from S3 bucket")
       deletePrefix(prefix, raw_bucket, split_files)
     })
     message("Truncating target table")
-    queryDo(dbcon, sprintf("truncate table %s", tableName))
+    queryDo(dbcon, sprintf("truncate table %s", table_name))
     if (remove_quotes) {
       query_string <- "copy %s from 's3://%s/%s.' region '%s' truncatecolumns acceptinvchars as '^' escape delimiter '|' removequotes gzip ignoreheader 1 emptyasnull STATUPDATE ON COMPUPDATE ON %s;"
     } else {
       query_string <- "copy %s from 's3://%s/%s.' region '%s' truncatecolumns acceptinvchars as '^' escape delimiter '|' gzip ignoreheader 1 emptyasnull STATUPDATE ON COMPUPDATE ON %s;"
     }
     DBI::dbExecute(dbcon, sprintf(
       query_string,
-      tableName,
+      table_name,
       raw_bucket,
       prefix,
       region,

R/redshift-spectrum.R

@@ -1,8 +1,5 @@
-table_parts <- function(table_name) {
-  return(strsplit(table_name, ".", fixed = TRUE)[[1]])
-}
-
 globalVariables(c("V1", "values")) # suppresses check note due to NSE
+
 #' List Spectrum partitions
 #'
 #' Provides character vector of partition names for a given table
@@ -75,7 +72,7 @@ spectrum_drop_partition <- function(dbcon, table_name, part_name, part_value) {
 #' @importFrom glue glue
 #' @importFrom DBI dbExecute
 spectrum_add_partition <- function(dbcon, table_name, part_name, part_value, base_location, quote_partition_value = TRUE) {
-  stopifnot("ident" %in% class(table_name))
+  stopifnoschema(table_name)
   if (quote_partition_value) {
     q <- "'"
   } else {

R/redshift-transaction.R

@@ -14,8 +14,8 @@
 #' .dbcon = rs$con,
 #' .function_sequence = list(
 #'  function(...) { rs_create_table(table_name = "mtcars", ...) },
-#'  function(...) { rs_upsert_table(tableName = "mtcars", ...) },
-#'  function(...) { rs_replace_table(tableName = "mtcars", ...) }
+#'  function(...) { rs_upsert_table(table_name = "mtcars", ...) },
+#'  function(...) { rs_replace_table(table_name = "mtcars", ...) }
 #'  )
 #' )
 #' }
@@ -40,14 +40,18 @@ transaction <- function(.data, .dbcon, .function_sequence) {
       TRUE
     },
     error = function(e) {
+      # send message now in case we hit another error during ROLLBACK - possible if we lost db connection
       message(e$message)
       DBI::dbExecute(.dbcon, "ROLLBACK;")
       message("Rollback complete")
-      FALSE
+      stop(glue("A redshift error occured: {e$message}"))
     }
   )
-  if (is.null(result) || !isTRUE(result)) {
-    stop("A redshift error occured")
+  if (is.null(result)) {
+    stop("A redshift error occured, the result of the transaction was NULL - which is unexpected")
+  }
+  if (!isTRUE(result)) {
+    stop("A redshift error occured, the result of the transaction was FALSE - which is unexpected")
   }
   return(result)
 }

R/redshift-upsert.R

@@ -6,7 +6,7 @@
 #'
 #' @param data a data frame
 #' @param dbcon an RPostgres connection to the redshift server
-#' @param tableName the name of the table to replace
+#' @param table_name the name of the table to replace
 #' @param split_files optional parameter to specify amount of files to split into. If not specified will look at amount of slices in Redshift to determine an optimal amount.
 #' @param keys athis optional vector contains the variables by which to upsert. If not defined, the upsert becomes an append.
 #' @param bucket the name of the temporary bucket to load the data. Will look for AWS_BUCKET_NAME on environment if not specified.
@@ -30,19 +30,19 @@
 #' host='my-redshift-url.amazon.com', port='5439',
 #' user='myuser', password='mypassword',sslmode='require')
 #'
-#' rs_upsert_table(data=nx, dbcon=con, tableName='testTable',
+#' rs_upsert_table(data=nx, dbcon=con, table_name='testTable',
 #' bucket="my-bucket", split_files=4, keys=c('a'))
 #'
 #' }
 #' @export
 rs_upsert_table <- function(
                             data,
                             dbcon,
-                            tableName,
+                            table_name,
                             keys = NULL,
                             split_files,
                             bucket=Sys.getenv("AWS_BUCKET_NAME"),
-                            region=Sys.getenv("AWS_DEFAULT_REGION"),
+                            region=Sys.getenv("AWS_DEFAULT_REGION", unset = "us-east-1"),
                             access_key=NULL,
                             secret_key=NULL,
                             strict = FALSE,
@@ -66,15 +66,19 @@ rs_upsert_table <- function(
         raw_bucket <- paste0(bucket, if (Sys.getenv("ENVIRONMENT") == "production") "" else "-test")
         split_files <- min(split_files, nrow(data))
 
-        data <- fix_column_order(data, dbcon, table_name = tableName, strict = strict)
+        data <- fix_column_order(data, dbcon, table_name = table_name, strict = strict)
         prefix <- uploadToS3(data, bucket, split_files)
         on.exit({
           message("Deleting temporary files from S3 bucket")
           deletePrefix(prefix, raw_bucket, split_files)
         })
         stageTable <- paste0(sample(letters, 32, replace = TRUE), collapse = "")
 
-        DBI::dbExecute(dbcon, sprintf("create temp table %s (like %s)", stageTable, tableName))
+        if (is_schema(table_name)) {
+          table_name <- schema_to_character(table_name)
+        }
+
+        DBI::dbExecute(dbcon, sprintf("create temp table %s (like %s)", stageTable, table_name))
 
         message("Copying data from S3 into Redshift")
         DBI::dbExecute(dbcon, sprintf(
@@ -88,18 +92,18 @@ rs_upsert_table <- function(
 
         if (!is.null(keys)) {
           message("Deleting rows with same keys")
-          keysCond <- paste(stageTable, ".", keys, "=", tableName, ".", keys, sep = "")
+          keysCond <- paste(stageTable, ".", keys, "=", table_name, ".", keys, sep = "")
           keysWhere <- sub(" and $", "", paste0(keysCond, collapse = "", sep = " and "))
           DBI::dbExecute(dbcon, sprintf(
             "delete from %s using %s where %s;",
-            tableName,
+            table_name,
             stageTable,
             keysWhere
           ))
         }
 
         message("Insert new rows")
-        DBI::dbExecute(dbcon, sprintf("insert into %s (select * from %s);", tableName, stageTable))
+        DBI::dbExecute(dbcon, sprintf("insert into %s (select * from %s);", table_name, stageTable))
         DBI::dbExecute(dbcon, sprintf("drop table %s;", stageTable))
       },
       error = function(e) {