# ------------------------------------------------------------------------------------------------------------------------------------------------------------ # # ---------------------------------------------------------------------------------------------------------------------------------------------------------- # # # # File Name : SF_Utils.R # # Last Updated Funclist : 14 Jul 2015, 8:00 PM (Tuesday) # # # # Author Name : Rick Saporta # # Author Email : RickSaporta@gmail.com # # Author URL : www.github.com/rsaporta # # # # Packages Called : RODBC # # Packages Used via NS : NA # # # # ---------------------------------------------------------------------------------------------------------------------------------------------------------- # # # # .q ( qry, ..., warehouse="SCIENCE" ) # # sfGetCon ( uid=getOption("snowflake_uid"), pwd=getOption("snowflake_pwd") # # , drv=getOption("snowflake_driver"), assign=TRUE # # , envir=getOption("con.environ", globalenv()), refresh=NULL # # , set_snowflake_inuse=TRUE ) # # dbschematbl ( dbname=if (!is.null(schema)) getSnowflakeDB() else NULL, schema=NULL, tbl=NULL ) # # sfShowCols ( tbl, schema=NULL, namelike=NULL, sort=TRUE, include.types=FALSE, returnRaw=FALSE # # , similar=TRUE, verbose=FALSE, simplify=TRUE, dbname=getSnowflakeDB() # # , warehouse = NULL, ...) # # sfDesc ( tbl=NULL, schema=NULL, dbname=getSnowflakeDB(), con=sfGetCon(), verbose=TRUE # # , ignore.case=TRUE, cleanBytes=TRUE # # , nomatch=c("all", "none"), details=c("auto", "all", "minimal"), comments_included=TRUE # # , showWarnings=TRUE, ... ) # # sfShowTables ( schema=NULL, namelike=NULL, dbname=getSnowflakeDB(), con=sfGetCon() # # , verbose=TRUE, ignore.case=TRUE, cleanBytes=TRUE # # , nomatch=c("all", "none"), details=c("auto", "all", "minimal"), comments_included=TRUE # # , showWarnings=TRUE, ... ) # # sfShowWarehouses ( namelike=NULL # # , state=c("ALL", "RUNNING", "STARTED", "SUSPENDED", "HIBERNATED"), details=c("auto", "minimal", "all"), connex=sfGetCon() # # , warehouse=NULL, justnames=FALSE, quote_name=FALSE, ..., showWarnings=TRUE # # , verbose=FALSE ) # # getNextHour ( time=now(), thresh_seconds=0 ) # # sf.make_list_name ( name, size, type ) # # sfGetWarehouseDetails ( name, ..., size=NULL, type=NULL, time_started=now() - 30, showWarnings=TRUE ) # # add_warehouse_details ( wh_opts ) # # getSnowflakeDB ( from_system=FALSE, showWarnings=TRUE ) # # sfSetWarehouseDetails ( name, ..., size=NULL, type=NULL, time_started=now() - 30, last_ran=NA # # , showWarnings=TRUE ) # # sfIsWarehouseOn ( name, connex=sfGetCon(), verbose="auto" ) # # sfWarehouseOn ( name=getOption("snowflake_defaultwh"), size=NULL, type=NULL, con=sfGetCon() # # , notify=TRUE, set_snowflake_inuse=TRUE, set_warehouse=TRUE, just_query=FALSE # # , quote_name=FALSE, showWarnings=TRUE, connex=con, verbose_qry=FALSE ) # # sfWarehouseOff ( name, connex=sfGetCon(), verbose=FALSE ) # # sfCreateWH ( name, size=NULL, type='STANDARD', auto_suspend_minutes=NULL, auto_resume=FALSE # # , initially_suspended=TRUE, comment=NULL, replace=FALSE, if_not_exists=TRUE # # , just_query=FALSE, quote_name=FALSE, showWarnings=TRUE, connex=sfGetCon() # # , verbose_qry=TRUE ) # # sfIncreaseWH ( name, size=NULL, type=NULL, auto_suspend_minutes=NULL, auto_resume=NULL # # , initially_suspended=NULL, comment=NULL, replace=FALSE, if_not_exists=TRUE # # , just_query=FALSE, quote_name=FALSE, showWarnings=TRUE, connex=sfGetCon() # # , verbose_qry=TRUE ) # # sf.make_wh_params ( size=NULL, type=NULL, auto_suspend_minutes=NULL, auto_resume=NULL # # , initially_suspended=NULL, comment=NULL, with_preface=FALSE, set_preface=FALSE # # , collapse="\n ", showWarnings=TRUE ) # # sf_clean_wh_input ( name # # , size, type, auto_suspend_minutes, auto_resume, initially_suspended, comment, quote_name=FALSE ) # # sf_clean_name ( name ) # # setSnowflakeDB ( dbname, schema=NULL, verbose=TRUE, verbose.qry=verbose ) # # sf_clean_size ( size ) # # sf_clean_type ( type ) # # sf_clean_auto_suspend_minutes ( auto_suspend_minutes ) # # sf_clean_auto_resume ( auto_resume ) # # sf_clean_initially_suspended ( initially_suspended ) # # sf_clean_comment ( comment ) # # sfQry ( qry, connex=sfGetCon() # # , warehouse=getOption("snowflake_currentwh", getOption("snowflake_defaultwh")), to.dt=exists("as.data.table") # # , refresh_con_if_expired = TRUE, fail_on_query_error = TRUE # # , update_last_ran = grepl("SELECT|UPDATE|INSERT", qry, ignore.case=TRUE) # # , verbose.max.width=getOption("width", 80) * 0.8, verbose.max.lines=22L # # , verbose.add.dots=nchar(qry) < 5000, verbose.shortCircuit=TRUE # # , verbose.indentAnd=grepl("\\bOR\\b", qry), verbose.indentOr=FALSE # # , verbose=TRUE, verbose.key=verbose, verbose.firstOfMonth=TRUE # # , size=NULL, type=NULL, ...) # # sfErrorParse ( qres, qry, return_error="auto", return_query="auto", fail_on_query_error=FALSE ) # # sfUpdateWH_QryLastRan ( name, size=NULL, type=NULL, last_ran=now() ) # # # # # # # # ---------------------------------------------------------------------------------------------------------------------------------------------------------- # # ------------------------------------------------------------------------------------------------------------------------------------------------------------ # # TODO: generic function for sfShowObject sfGetObject sfExistsObject ## Temporary workaround toUTCdate <- function(x) {as.Date(as.tz(x, "UTC"))} ## Quick function for personal use .q <- function(qry, ..., warehouse=if(exists("wh", envir=parent.frame())) get("wh", envir=parent.frame()) else getWH_by_interactive()) { sfQry(qry=qry, ..., warehouse=warehouse) } .warehouse_sizes <- c("X-Small", "Small", "Medium", "Large", "X-Large", "2X-Large", "3X-Large") .warehouse_sizes_old <- c("X-Small"="xS", "Small"="S", "Medium"="M", "Large"="L", "X-Large"="XL", "2X-Large"="XXL", "2X-Large"="2XL", "3X-Large"="3XL") sfBoolCase <- function(boolCol, nm=gsub("(.*_)(is_.*)", "\\2", boolCol), yes='Y', no='N', tbl=NULL, schema=NULL, dbname=NULL) { if (is.na(yes) || is.na(no)) stop ("yes/no cannot be NA. Quote it if you need it.") if (is.logical(yes) || is.logical(no)) { warning("snowflake does not have a column type for logical. (Thus the point to this function in the first place)\nResults will be returned as the strings '", yes, "' and '", no, "'") yes %<>% as.character() no %<>% as.character() } dstbl <- dbschematbl(dbname=dbname, schema=schema, tbl=tbl) ifelse(is.null(dstbl), boolCol, sprintf("%s.%s", dstbl, boolCol)) %>% sprintf("CASE %s WHEN 1 THEN '%s' ELSE '%s' END", ., yes, no) %>% setNames(nm=nm, obj=.) } sfSchemaExists <- function(schema, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), verbose=FALSE) { force(schema) qry <- sprintf("SHOW SCHEMAs in %s", dbname) DT.schemas <- sfQry(qry, wh=wh, dbname=dbname, verbose=verbose) toupper(schema) %in% DT.schemas$name } sfShowSchemas <- function(wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeSchema(), justnames=FALSE, verbose=FALSE) { qry <- sprintf("SHOW SCHEMAs in %s", dbname) ret <- sfQry(qry, wh=wh, dbname=dbname, verbose=verbose) if (!nrow(ret)) { warning("No schemas found") if (justnames) return (c()) else return(ret) } if (justnames) return(ret[["name"]]) return(ret) } sfUpdateWH_QryLastRan <- function(name, size=NULL, type=NULL, last_ran=now()) { is.char_of_length1(name, fail.if.not=TRUE) list_name <- .sf.make_list_name(name, size=size, type=type) current_options <- getOption("sf_warehouses", default = list()) ## Initialize if missing if (is.null(current_options[[list_name]][["last_ran"]])) current_options[[list_name]][["last_ran"]] <- list() if (is.null(current_options[[list_name]][["name"]])) current_options[[list_name]][["name"]] <- name current_options[[list_name]][["last_ran"]] <- now() options("sf_warehouses" = current_options) return(invisible(current_options)) } .rm_schema_preface <- function(schema, tbl, ignore.case=TRUE, sep=".", showWarnings=TRUE) { ## is the opposite of schemapaste if (is.null(tbl) || is.null(schema) || !length(tbl)) return(tbl) if (!is.character(tbl) || !is.character(schema)) stop ("schema and tbl must be characters") if (any(grepl("(\\s|\\b)JOIN(\\s|\\b)", tbl, ignore.case=TRUE))) { if (length(tbl) > 1) warning ("JOIN statement found in value of tbl, however length(tbl) is ", length(tbl), " and we do not know how to automatically deal with this") return(tbl) } L.tbl <- length(tbl) L.schema <- length(schema) if (L.tbl != L.schema && 1 %ni% c(L.tbl, L.schema)) stop ("schema and tbl must have the same length or one of them must have length 1") if (L.tbl > 1 || L.schema > 1) return(setNames(nm=names(tbl), obj=mapply(.rm_schema_preface, schema=schema, tbl=tbl, ignore.case=ignore.case, sep=sep, USE.NAMES=FALSE))) splat <- strsplit(tbl, escapeRegEx(sep)) %>% unlist if (length(splat) == 1) return(tbl) if (length(splat) > 2 && showWarnings) warning ("tbl '", tbl, "' has more than one separator. .rm_schema_preface() may not act correctly", call.=FALSE) if (toupper(splat[[1]]) == toupper(schema)) return(splat[-1] %>% pasteC(C=sep)) ## 2015-10-14 -- Not sure why using tail() here? ## Need to check if anywhere where this is needed? If not, just return(tbl) or return(pasteC(splat, C=sep)) return(tail(splat, 1)) } dbschematbl <- function(dbname=if (!is.null(schema)) getSnowflakeDB() else NULL, schema=NULL, tbl=NULL, clean_tbl=TRUE) { if (is.null(dbname) && is.null(schema) && is.null(tbl)) return(NULL) if (is.null(schema) && !is.null(dbname) && !is.null(tbl)) { if (!grepl("\\.", tbl)) stop ("schema cannot be NULL if both dbname and tbl are NOT null") } if (!is.null(dbname) && !is.null(schema) && grepl(sprintf("^%s\\.", dbname), schema, ignore.case=TRUE)) { dbname <- schema schema <- NULL } if (clean_tbl) tbl %<>% tableNameClean(schema=schema) ## .rm_schema_preface is sloppy and does not work well tbl %<>% ifelseNULL(no=removeText(paste0("^", dbname, "\\."), ., ignore.case=TRUE)) schema %<>% ifelseNULL(no=removeText(paste0("^", dbname, "\\."), ., ignore.case=TRUE)) tbl %<>% ifelseNULL(no=removeText(paste0("^", schema, "\\."), ., ignore.case=TRUE)) # old: schema %<>% .rm_schema_preface(schema=dbname, tbl=., showWarnings=FALSE) # old: tbl %<>% .rm_schema_preface(schema=schema, tbl=., showWarnings=FALSE) stbl <- { if (!is.null(schema) && !is.null(tbl)) paste(schema, tbl, sep=".") else if(!is.null(schema)) schema else if (!is.null(tbl)) tbl else NULL } dstbl <- { if (!is.null(dbname) && !is.null(stbl)) paste(dbname, stbl, sep=".") else if(!is.null(dbname)) dbname else if (!is.null(stbl)) stbl else NULL } return(dstbl) } sfShowCols <- function(tbl, schema=NULL, namelike=NULL, sort=TRUE, include.types=FALSE, returnRaw=FALSE, similar=TRUE, what=c("TABLE", "VIEW", "STAGE"), use.view=FALSE, verbose=FALSE, simplify=TRUE , dbname=getSnowflakeDB() , wh = getSnowflakeWH() , ... ) { if (isTRUE(use.view)) what <- "VIEW" what <- match.arg(toupper(what)[[1]], c("TABLE", "VIEW", "STAGE")) tbl %<>% tableNameClean(schema=schema) dst <- dbschematbl(dbname=dbname, schema=schema, tbl=tbl) DT.cols <- sfQry(sprintf("DESC %s %s", what, dst), wh=wh, verbose=FALSE) return(DT.cols$name) } .sf_drop_commentCol_ <- function(DT, comments_included="auto") { ## Many functions have a ret DT that if ("comment" %in% names(DT)) { if (identical(comments_included, "auto") && (all(is.na(DT$comment)) || all(DT$comment == ""))) comments_included <- FALSE if (!is.logical(comments_included) || is.na(comments_included)) { warning ("`comments_included` should be TRUE/FALSE or 'auto'") comments_included <- FALSE } if (!comments_included) DT[, comment := NULL] else setcolorderpt(DT, endCols="comment") } return(invisible(DT)) } sfDesc <- function(tbl=NULL, schema=NULL, dbname=getSnowflakeDB(), format_name=NULL, con=sfGetCon(), what=c("TABLE", "VIEW", "STAGE", "FILE FORMAT"), use.view=FALSE, verbose=TRUE, ignore.case=TRUE, cleanBytes=TRUE, nomatch=c("all", "none"), details=c("auto", "all", "minimal"), wh = getSnowflakeWH(), comments_included="auto", showWarnings=TRUE, quiet=FALSE, ...) { if (isTRUE(use.view)) what <- "VIEW" what <- match.arg(toupper(what)[[1]], c("TABLE", "VIEW", "STAGE", "FILE FORMAT")) if (!is.null(format_name)) { using_format <- TRUE what <- "FILE FORMAT" if (!is.null(tbl)) warning("if both tbl and format_name are used in sfDesc(), tbl is ignored") if (is.null(schema)) { if (grepl("\\.", format_name)) { splat <- splitOnLast(format_name, "\\.")[[1]] schema <- splat[[1]] format_name <- splat[[2]] } else if (missing(schema)) { message("using schema 'public' for file format") schema <- "PUBLIC" } else { warning("schema is NULL for format_name. Generally, most file formats are in the PUBLIC schema. However, this script will look in the default schema for the connection, which may be different") ## if schema is null and dbname is not, that will through an error for dbschematbl() below if (missing(dbname)) dbname <- NULL } } ## ELSE, not using_format } else using_format <- FALSE tbl %<>% tableNameClean(schema=schema) dst <- dbschematbl(dbname=dbname, schema=schema, tbl=ifelse(using_format, format_name, tbl)) qry <- sprintf("DESC %s %s", what, dst) %>% setQry ret <- sfQry(qry, wh=wh, verbose=FALSE) if ("kind" %in% names(ret) && all(ret[["kind"]] == "COLUMN")) ret[, kind := NULL] if ("default" %in% names(ret) && all(is.na(ret[["default"]]))) ret[, default := NULL] ## Optionally drop comments .sf_drop_commentCol_(ret, comments_included=comments_included) nr <- nrow(ret) blank_vec <- rep("", nr) if (nr) for (col in names(ret)) if (all(is.na(ret[[col]]))) ret[, (col) := blank_vec] if (!quiet) { if (using_format) print(ret) else print(ret[order(name)], nrow=1000) } return(invisible(ret)) } sfShowStages <- function(schema=NULL, namelike=NULL, what=c("TABLES", "VIEWS", "STAGES"), justnames=FALSE, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), connex=sfGetCon(), verbose=TRUE, ignore.case=TRUE, cleanBytes=TRUE, nomatch=c("all", "none"), details=c("auto", "all", "minimal"), use.view=FALSE, comments_included=TRUE, showWarnings=TRUE, ...) { nomatch <- match.arg(nomatch) details <- match.arg(details) ARGS <- collectArgs() ARGS$what <- "STAGES" do.call(sfShowTables, ARGS) } sfShowViews <- function(schema=NULL, namelike=NULL, what=c("TABLES", "VIEWS", "STAGES"), justnames=FALSE, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), connex=sfGetCon(), verbose=TRUE, ignore.case=TRUE, cleanBytes=TRUE, nomatch=c("all", "none"), details=c("auto", "all", "minimal"), use.view=FALSE, comments_included=TRUE, showWarnings=TRUE, ...) { nomatch <- match.arg(nomatch) details <- match.arg(details) ARGS <- collectArgs() ARGS$what <- "VIEWS" do.call(sfShowTables, ARGS) } sfShowTables <- function(schema=NULL, namelike=NULL, what=c("TABLES", "VIEWS", "STAGES"), justnames=FALSE, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), ignore.case=TRUE, cleanBytes=TRUE, nomatch=c("all", "none"), fail_if_empty=TRUE, details=c("auto", "all", "minimal"), use.view=FALSE, comments_included=TRUE, showWarnings=TRUE, blank_out_leons_yt_comments=TRUE, all_dbnames=FALSE, connex=sfGetCon(), verbose=TRUE, ...) { # if (length(schema) > 1) { # args <- collectArgs() # return(rbindlist(lapply())) # } ## be flexible as to the value of details if (isTRUE(details)) details <- "all" else if (identical(details, FALSE)) details <- "minimal" else details <- match.arg(details) nomatch <- match.arg(nomatch) if (isTRUE(use.view)) what <- "VIEWS" what <- match.arg(toupper(what)[[1]], c("TABLES", "VIEWS", "STAGES")) # dst <- dbschematbl(dbname=dbname, schema=schema, tbl=NULL) # # qry <- paste("SHOW TABLES", if (!is.null(dst)) " IN ", dst, if (!is.null(pattern)) ifelse(nchar(pattern), paste(" LIKE ", pasteQ(pattern, C=NULL, w="")), "")) %>% setQry # # if (length(qry) > 1) { # ret <- qry %>% lapply(sfQry, verbose=verbose, con=con) # try({ret %<>% rbindlist}, silent=TRUE) # } else { # ret <- qry %>% sfQry(con=con, verbose=verbose) # } ## if all_dbnames, then query to get the list of all databases, then iterate over the list of dbnames, and combine before cleaning ## otherwise, just query as normal. if (all_dbnames) { dbnames <- sfQry("SHOW DATABASES", verbose=FALSE)$name verboseMsg(verbose, func="message", "Will be checking", length(dbnames), "database(s): ", pasteC(dbnames, C=", "), time=FALSE) qrys <- sprintf("SHOW %s IN DATABASE %s", what, dbnames) if (use.view) qrys %<>% tableToView() ll_DT <- lapply(qry, function(x) sfQry(qry=x, con=connex, verbose=FALSE)) ## set empty data.tables to NULLs, then rbind the list ret <- lapply(ll_DT, function(x) if (nrow(x)) x else NULL) %>% removeNullsAndBlanksFromList %>% rbindlist(use.names=TRUE) } else { if (!is.null(dbname) && is.null(schema) && !grepl("\\.", dbname)) qry <- sprintf("SHOW %s IN DATABASE %s", what, dbname) %>% setQry else qry <- sprintf("SHOW %s IN %s", what, dbschematbl(dbname=dbname, schema=schema)) %>% setQry if (use.view) qry %<>% tableToView() ret <- qry %>% sfQry(qry=., con=connex, verbose=verbose) } ret_is_empty <- !nrow(ret) if (!missing(fail_if_empty)) warning ("fail_if_empty has been DEPRECATED from sfShowTables()", call.=FALSE) # if (fail_if_empty && !nrow(ret)) # stop ("Internal Error. Querying for SHOW TABLES did not return any results") ## IDENTIFY WHICH ROWS MATCH FOR SCHEMA if (!is.null(schema)) { matched_schema <- ## OLD: gsub(sprintf("^%s\\.", dbname), "", schema, ignore.case=TRUE) %>% removeText(paste0(dbname, "\\."), start=TRUE, x=schema) %>% regOr %>% grepl(pat=., x=ret[["schema_name"]], ignore.case=ignore.case) } else matched_schema <- TRUE ## IDENTIFY WHICH ROWS MATCH FOR TBL if (!is.null(namelike)) { matched_tbl <- gsub(sprintf("^(%s\\.)?%s\\.", dbname %>% valueIfNull(""), schema %>% valueIfNull("")), "", namelike, ignore.case=TRUE) %>% gsub("\\%", ".*", .) %>% sprintf("^%s$", .) %>% regOr %>% grepl(pat=., x=ret[["name"]], ignore.case=ignore.case) } else matched_tbl <- TRUE matched_rows <- matched_schema & matched_tbl if (!any(matched_rows)) { if (showWarnings) { ## Two possible reasons why no matches -- warn accordingly if (ret_is_empty) warning ("Schema '", schema, "' exists but has no ", tolower(what), ".") else if (showWarnings) warning ("No ", tolower(what), " found matching schema = '", schema, "' and namelike = '", namelike, "'", if (nomatch == 'all') " -- Returning all tables") } if (nomatch == "none") return(if (justnames) c() else ret[FALSE]) } else ret <- ret[matched_rows] if (justnames) return(ret[["name"]]) drop_column = function(vec, length_1_returns=FALSE, length_0_returns=TRUE) { ## The logical value indicates whether to drop the column or not if (length(vec) == 1) return(length_1_returns) if (length(vec) == 0) return(length_0_returns) if (all(is.na(vec))) return(TRUE) return (all(vec == vec[[1]])) } is_minimal <- identical(details, "minimal") if (details != "all" && !ret_is_empty) { if (is_minimal || all(is.na(ret$owner)) || all(ret$owner == "SYSADMIN")) ret[, owner := NULL] # account_name deprecated in latest release of snowflake # if (is_minimal || drop_column(ret$account_name)) # ret[, account_name := NULL] if ("kind" %in% names(ret) && all(ret$kind == "TABLE")) ret[, kind := NULL] if ("cluster_by" %in% names(ret) && (is_minimal || all(is.na(ret$cluster_by)))) ret[, cluster_by := NULL] if (drop_column(ret$schema_name)) ret[, schema_name := NULL] if (all(ret$database_name == "PROD") || (is_minimal && drop_column(ret$database_name))) ret[, database_name := NULL] # if (drop_column(ret$database_name)) # ret[, database_name := NULL] ## round off the bytes to human readable if (isTRUE(cleanBytes) && "bytes" %in% names(ret)) ret[, bytes := formatBytes(bytes)] ## Crop the timestamp to just date ret[, created_on := as.Date(created_on)] ## TEXT is too large and unwieldly. Wrapping it in a list, makes it display nice and neatly if ("text" %in% names(ret)) ret[, text := lapply(text, as.list)] } ## Whether to keep comments or not is a separate flag if (missing(comments_included) && is_minimal) comments_included <- FALSE .sf_drop_commentCol_(ret, comments_included=comments_included) if (blank_out_leons_yt_comments && all(c("schema_name", "comment") %in% names(ret))) ret[tolower(schema_name) == "youtube_analytics", comment := NA] # if ("comment" %in% names(ret)) { # if (!comments_included || (missing(comments_included) && is_minimal)) # ret[, comment := NULL] # else # setcolorderpt(ret, endCols="comment") # } return(ret[]) } sfWaitForWarehouse <- function(wh, turn_on=TRUE, N.seconds_to_wait = 60, max_iterations_to_wait = 100, verbose.waiting=verbose, verbose=TRUE) { iter <- 0 # while(!identical(100L, .avail <- sfShowWarehouses()[toupper(name) == toupper(wh), available]) && iter <= max_iterations_to_wait) { while(!identical(100L, .avail <- sfCheckWarehouseAmountAvailable(wh, turn_on=turn_on)) && iter <= max_iterations_to_wait) { verboseMsg(verbose.waiting, "Warehouse is not at 100% it is at ", fwp(.avail/100), " -- will wait ", N.seconds_to_wait, " seconds (iteration # ", iter, ")", sep="") Sys.sleep(N.seconds_to_wait) iter <- iter + 1 } if (!identical(100L, sfCheckWarehouseAmountAvailable(wh))) warning ("Warehouse did not reach 100%") else if (iter > 1) message ("Warehouse ", wh, " is now at 100%") return(invisible(NULL)) } # sfShowWarehouses() .sfKnownStates <- c("RUNNING", "STARTED", "SUSPENDED", "HIBERNATED", "CREATING", "RESIZING", "SUSPENDING") sfShowWarehouses <- function(namelike=NULL, state=c("ALL", .sfKnownStates), details=c("auto", "minimal", "all"), connex=sfGetCon(), warehouse=NULL, justnames=FALSE, quote_name=FALSE, ..., showWarnings=TRUE, verbose=FALSE) { ## Check that state is correct state <- toupper(state) if (length(state) == 1 && state %in% c("RUNNING", "RUN")) stop ("There is no state called 'running'\n\nHINT: Did you mean sate = 'started'") state <- match.arg(state) if (state == "SUSPENDED") { warning ("Snowflake uses inconsistent terminology regarding warehouse 'state'/'status' -- programatically, please use 'HIBERNATED' instead of 'SUSPENDED'", call.=FALSE) state <- "HIBERNATED" } details <- match.arg(details) qry <- "SHOW WAREHOUSES" if (!is.null(namelike)) qry <- sprintf("%s LIKE '%s'", qry, namelike) ## Execute query ret <- sfQry(qry, connex=connex, warehouse=warehouse, update_last_ran=FALSE, tz_blanks_to=getOption("default.tz", "America/New_York"), verbose=verbose) ## Add column for how many minutes remain if (all(c("state", "resumed_on") %in% names(ret)) && any(ret$state != "SUSPENDED")) { ret[, minutes_til_next_hour := NA_integer_] ret[state != "SUSPENDED", minutes_til_next_hour := as.POSIXlt(resumed_on)$min - as.POSIXlt(now())$min] ret[minutes_til_next_hour < 5, c("Shut Down Soon") := "<~~~~~ *** "] } ## Currently, we are only expecting c("STARTED", "HIBERNATED") for state. IF there are others, notify so we can modify script other_states <- setdiff(ret[["state"]], .sfKnownStates) if (length(other_states)) warning ("There are other 'state' values in sfShowWarehouses(). They are: \n ", paste(other_states, sep="\n\t", collapse="\n\t")) ## If 'ALL' is one of the states, then we are not filtering if (!("ALL" %in% state)) { total_rows_returned <- nrow(ret) e <- environment() ret <- ret[state %in% (get("state", envir=e))] if (showWarnings && nrow(ret) == 0) { if (total_rows_returned == 0) warning ("No warehouses found, regardless of state") else warning ("No warehouses found for state '", state, "' -- however ", total_rows_returned, " warehouse(s) found with other states") } } ## QUICK EXIT: If justnames, go ahead and return those to the user now if (justnames) return(ret[["name"]]) ## TODO: Separate these, check if they are other than the default values if (details != "all") { colsDropping <- c("created_on", "auto_resume", "owner", "comment", "uuid", "is_default", "is_current") # "suspended" ## colsAuto is the list of columns that only get dropped if details is set to "auto" ## The list contains the expected values. If the whole column is NA or the expected value, then the column is dropped colsAuto <- list(available="", running=0, queued=0, stranded=0, pendings=0, failed=0, provisioning="", quiescing="", other="") colsAuto %<>% {.[names(.) %in% names(ret)]} if (details == "minimal") ret[, names(colsAuto) := NULL] else if (details == "auto") { for (nm.col in names(colsAuto)) { if (all(is.na(ret[[nm.col]])) || all(ret[[nm.col]] == colsAuto[[nm.col]])) ret[, (nm.col) := NULL] } } ret[, (colsDropping) := NULL] } if (FALSE) { warehouse_details <- sfGetWarehouseDetails() if (!identical(warehouse_details, list())) { browser(text = "rbindlist") DT.warehouse_details <- rbindlist(lapply(warehouse_details, as.data.table), use.names=TRUE, fill=TRUE) # DT.warehouse_details[, state := factor("On ", levels=levels(ret$state))] ## For now, assume all warehouses are "On" DT.warehouse_details[, state := "STARTED"] ## For now, assume all warehouses are "On" ## setkey either to just name, state or to name, size, type, statea if (!anyDuplicated(DT.warehouse_details$name)) setkey(DT.warehouse_details, name, state) else { message("Internal Error: There are duplicate 'name' values in DT.warehouse_details. Details may not come over correctly") print(DT.warehouse_details) setkey(DT.warehouse_details, name, size, type, state) } colsToBring <- c("time_running", mins_left="minutes_left_in_hour") if (all(colsToBring %in% names(DT.warehouse_details))) addColsFrom_(ret, DT.warehouse_details, colsToBring=colsToBring, joinCols=key(DT.warehouse_details), showWarnings=FALSE) if ("mins_left" %in% names(ret)) ret[, mins_left := removeNA(gsub("mins .*", "mins", mins_left), "")] if ("time_running" %in% names(ret)) ret[, time_running := removeNA(gsub("mins .*", "mins", time_running), "")] } } ## Clean up part 2 -- after addColsFrom_ if (details != "all") { ret[, size := toFactorWithExpectedLevels(size, lev=.warehouse_sizes, lab=c("xS", "S", "M", "L", "XL", "XXL", "3XL"), cleanDuplicates=TRUE)] ret[, type := toFactorWithExpectedLevels(type, lev=c("STANDARD", "ENTERPRISE"), lab=c("STANDARD", "ENTERPRISE"), cleanDuplicates=TRUE)] ret[, state := toFactorWithExpectedLevels(state, lev=c("STARTED", "SUSPENDED", "CREATING", "RESIZING", "SUSPENDING"), lab=c("On ", "Off", "Booting Up", "Resizing", "Off'ing"), cleanDuplicates=TRUE)] ret[, auto_suspend := fwSecs(auto_suspend)] ret[is.na(auto_suspend), auto_suspend := " ----- "] ret[, nodes_active := sprintf("%2i / %2i", actives, actives + suspended)] ret[, c("actives", "suspended") := NULL] ## For character columns, clean up NAs and replace them with blanks for (col in names(ret)) if (is.character(ret[[col]])) ret[ , (col) := removeNA(x=get(col), replace="", modify_levels_when_x_is_factor=FALSE)] setkeyIfNot(ret, key=c("type", "size", "name", "state"), organize=TRUE, verbose=FALSE) } return(ret[]) } getNextHour <- function(time=now(), thresh_seconds=0) { time_pl <- as.POSIXlt(time) now_pl <- as.POSIXlt(now()) ## Check that the timezones match zone.t <- time_pl$zone zone.n <- now_pl$zone if (!identical(zone.t, zone.n)) warning (sprintf("Warning differing time zones for time_pl (%s) and now_pl (%s)", valueIfNull(zone.t, "NULL"), valueIfNull(zone.t, "NULL"))) ## Bring over time zone information attr(now_pl, "tzone") <- attr(time_pl, "tzone", exact=TRUE) ## Bring over these individual pieces pieces_to_preserve <- c("zone", "gmtoff", "isdst", "sec", "min") now_pl$zone <- time_pl$zone now_pl$gmtoff <- time_pl$gmtoff now_pl$isdst <- time_pl$isdst now_pl$sec <- time_pl$sec now_pl$min <- time_pl$min ## Add hours until threshold exceeded while (as.numeric(now_pl) - as.numeric(now()) <= thresh_seconds) now_pl$hour <- now_pl$hour + 1 return(as.POSIXct(now_pl)) } .sf.make_list_name <- function(name, size, type) { ## Note: size and type are NOT needed, since a warehouse name must be unique. ## Also, warehouse names are case INsensitive. Coercing all to upper case. is.char_of_length1(name, fail.if.not=TRUE) # use just name # if (!is.null(size)) # use just name # is.char_of_length1(size, fail.if.not=TRUE) # use just name # if (!is.null(type)) # use just name # is.char_of_length1(type, fail.if.not=TRUE) # use just name # # use just name # sprintf("%s_%s_%s", name, valueIfNull(size, "default"), valueIfNull(type, "default")) return(toupper(name)) } sfGetWarehouseDetails <- function(name, ..., size=NULL, type=NULL, time_started=now() - 30, showWarnings=TRUE) { add_warehouse_details <- function(wh_opts) { if (is.null(wh_opts$time_started)) return(wh_opts) wh_opts$as_of <- now() wh_opts$time_running <- fwTDiff(start=wh_opts$time_started, end=wh_opts$as_of) wh_opts$next_hour <- getNextHour(wh_opts$time_started, thresh_seconds=180) wh_opts$minutes_left_in_hour <- fwTDiff(start=wh_opts$as_of, end=wh_opts$next_hour) wh_opts } all_warehouse_details <- getOption("sf_warehouses", default=list()) if (missing(name) && missing(size) && missing(type)) return(lapply(all_warehouse_details, add_warehouse_details)) if (!is.character(name)) stop ("name should be a character") size %<>% valueIfNull("default") type %<>% valueIfNull("default") list_name <- .sf.make_list_name(name, size, type) opts <- all_warehouse_details[[list_name]] if (!is.null(opts)) opts <- add_warehouse_details(opts) return(opts) } if (FALSE) { sfSetWarehouseDetails("load_Wh") sfSetWarehouseDetails() ## ERROR } if (FALSE) { sfWarehouseOn("load_wh", just_query=TRUE, quote=TRUE) sfWarehouseOn("SCIENCE", just_query=TRUE, quote=TRUE) } sfWhicheverIsOn <- function(whs="a string-vector of warehouse names, checked in order", default_if_none_on=whs[[1]], verbose=TRUE) { ## Checks which warehouses are on, and selects the first in whs that is on; ## if none on, returns default_if_none_on DT.WHs <- sfShowWarehouses(verbose=FALSE) %>% setkeyIfNot(name, verbose=FALSE) whs_on <- DT.WHs[trim(state) == "On", toupper(name)] whs_found <- sapply(whs, extract, whs_on) %>% unlist if (!length(whs_found)) { ret <- default_if_none_on sprintf("None of the indicated warehouses is currently on; Using default '%s'", ret) %>% verboseMsg(verbose, ., func="message") } else { ret <- whs_found[[1]] sprintf("Using warehouse '%s' currently set to size '%s'", ret, DT.WHs[.(ret)]$size) %>% verboseMsg(verbose, ., func="message") } return(ret[]) } sfCheckWarehouseAmountAvailable <- function(wh, turn_on=TRUE, ignore.case=TRUE, showWarnings=TRUE, verbose=TRUE) { ## returns FALSE if cannot find warehouse ## returns 0 if warehouse is off, with warning ## Otherwsie returns the percentage of the warehouse that is available. ## 100 for fully-on, 0 for none-but-booting-up (no warning here for such 0) sfIsWarehouseOn(wh=wh, check_amount_avail=TRUE, turn_on=turn_on, ignore.case=ignore.case, showWarnings=showWarnings, verbose=verbose) } sfIsWarehouseOn <- function(wh, exact=TRUE, check_amount_avail=FALSE, turn_on=FALSE, ignore.case=FALSE, include_booting_up=TRUE, showWarnings=TRUE, verbose=TRUE) { DT.warehouses <- sfShowWarehouses(details=ifelse(check_amount_avail, "all", "minimal")) pat <- wh %>% {ifelse(exact, sprintf("^%s$", .), .)} DT.matches <- DT.warehouses[grepl(pat, name, ignore.case=ignore.case)] if (nrow(DT.matches) > 1) { if (length(m <- grep(sprintf("^%s$", wh), ignore.case=ignore.case)) == 1) DT.matches <- DT.matches[m] else { warning("\nThere is more than one matched warehouse.", ifelse(check_amount_avail, "\nWill Return FALSE for sfIsWarehouseAvailable()", "Will be using the first warehouse"),"\nHere are the details:") print(DT.matches) DT.matches[, state := ifelse(any(tolower(trim(state)) %in% c("on", "creating", "booting up")), "On", "Off")] if (check_amount_avail) return(FALSE) } } if (!nrow(DT.matches)) { verboseMsg(showWarnings, "Warehouse '", wh, "' could not be found") return(FALSE) } ## if checking amount available, first check for NA or Off if (check_amount_avail) { avail <- DT.matches[, available] if (is.na(avail)) { warning("wh '", wh, "' is not on. Will return 0 for available") if (turn_on) sfWarehouseOn(wh=wh, verbose=verbose) return(0) } return(avail) } ## The parameter turn_on is only for when checking if a warehouse is available if (turn_on) warning("the parameter 'turn_on' in sfIsWarehouseOn() only applies when check_amount_avail is TRUE. Otherwise it is ignored\n\nHint: use sfWarehouseOn(wh)") # state <- DT.matches[1, trim(as.character(state))] state <- DT.matches$state[[1]] %>% as.character %>% trim %>% tolower if (state == "off") { verboseMsg(verbose, "Warehouse '", wh, "' is currently Off", sep="") return(FALSE) } if (state %ni% c("on", "creating", "booting up")) { verboseMsg(verbose, "Warehouse '", wh, "' has a state other than On/Off/Creating/Booting Up. Namely: ", state, "\nReturning FALSE for sfIsWarehouseOn()", sep="") return(FALSE) } ## for messaging state.str <- if (state == "on") "On" else if (state == "creating") "being created" else state verboseMsg(verbose, "Warehouse '", wh, "' is ", state.str, "", sep="") return(TRUE) } ## WHOOPS --- I wrote the same function twice. ## This is the older one: ## ------------------------------------------------------------------------ # sfIsWarehouseOn <- function(name, connex=sfGetCon(), verbose="auto") { # ## auto verbose will only verbose when the warehouse is on # verbose_if_FALSE <- isTRUE(verbose) # verbose <- isTRUE(verbose) || verbose == "auto" # # if (verbose) # name %<>% toupper # # DT.warehouses <- sfShowWarehouses(justnames=FALSE) # if (name %ni% DT.warehouses$name) { # warning ("warehouse '", name, " does not exist") # return(FALSE) # } # setkeyIfNot(DT.warehouses, name, warnForColNameInEnv=FALSE) # details <- DT.warehouses[name] # # is_on <- DT.warehouses[name, trim(state) == "On"] # # if (!is_on) # verboseMsg(verbose_if_FALSE, sprintf("Warehouse '%s' is not currently On", name), time=FALSE) # else # verboseMsg(verbose, DT.warehouses[name, sprintf("Warehouse '%s' is On and of type %s with size %s", name, type, size)], time=FALSE) # # return(is_on) # } sfWarehouseOn <- function(wh=name, size=NULL, type=NULL, connex=sfGetCon(), notify=TRUE , set_snowflake_inuse=TRUE, set_warehouse=TRUE, just_query=FALSE, quote_name=FALSE, showWarnings=TRUE, verbose_qry=FALSE ## old argument instead of wh was name , name=getOption("snowflake_defaultwh") ) { ## Note that if the warehouse is already running, this will fail # name <- .sf_clean_wh_input(name=name) [[1L]] # size <- .sf_clean_wh_input(size=size) [[1L]] # type <- .sf_clean_wh_input(type=type) [[1L]] is.char_of_length1(wh, fail=TRUE) if (isTRUE(set_snowflake_inuse)) options("snowflake_inuse" = TRUE) start_time <- now() wh %<>% .sf_clean_name params_string <- .sf.make_wh_params(size=size, type=type) if (sfIsWarehouseOn(wh=wh, verbose="auto")) { if (set_warehouse) setSnowflake(wh=wh, connex=connex, verbose=FALSE) return(invisible(NULL)) ## debugging } else { ## debugging catn("Warehouse is NOT on ... have a browse") ## debugging browser(text="Warehouse is not on") } qry <- paste("ALTER WAREHOUSE" , wh , "RESUME" # , params_string , "\n" ) # size_param <- size %>% .sf_clean_size %>% {if (!is.null(.)) sprintf(" WAREHOUSE_SIZE = %s", .)} # type_param <- type %>% .sf_clean_type %>% {if (!is.null(.)) sprintf(" WAREHOUSE_TYPE = %s", .)} # size_param <- # if (!is.null(size)) { # size <- match.arg(toupper(size), choices=c("XSMALL", "SMALL", "MEDIUM", "LARGE", "XLARGE")) # # qry <- paste(qry, sprintf(" WAREHOUSE_SIZE = %s", toupper(size))) # sprintf(" WAREHOUSE_SIZE = %s", toupper(size)) # } # type_param <- # if (!is.null(type)) { # type <- match.arg(tolower(type), choices=c("standard", "enterprise")) # # qry <- paste(qry, sprintf(" WAREHOUSE_TYPE = %s", toupper(type))) # sprintf(" WAREHOUSE_TYPE = %s", toupper(type)) # } # ## Add in the parameters, assuming at least one is not NULL # params <- c(size_param, type_param) # if (!is.null(params)) { # qry <- paste(qry, "USING", paste(params, collapse=" ")) # } # ## Add a semicolon and set the class to be nice and neat # qry <- paste(qry, ";") setQry(qry) if (just_query) return(qry) ## info for user verboseMsg(verbose, "Provisioning warehouse '", wh, "'", if (!is.null(params_string)) paste(" with params", params_string), sep="", minw=88) ## Execute ret <- sfQry(qry, connex=connex, verbose=verbose_qry, wh=wh) ## Set details if success if (!isErr(ret) && set_warehouse) { setSnowflake(wh=wh, connex=connex, verbose=FALSE) } if (notify) notifyIfTimeConsuming(msg="", subj=sprintf("warehouse %s is initizlized with size %s", wh, valueIfNull(size, "default")), start_time=start_time, end_time=now(), minutes=1) return(invisible(ret)) } if (FALSE) sfWarehouseOn(size="L", type="stan") ## TODO for warehouse provisioning # , auto_suspend_minutes=15 # AUTO_SUSPEND # Specifies the period of inactivity (in seconds) after which a warehouse will be automatically suspended. A NULL value specifies that the warehouse is never automatically suspended due to inactivity. # AUTO_RESUME # Specifies whether to automatically resume a warehouse when it is accessed. Valid values are ‘TRUE’ and ‘FALSE’. # INITIALLY_SUSPENDED # Specifies whether the warehouse is created initially in suspended state. Valid values are ‘TRUE’ or ‘FALSE’. sfWarehouseOff <- function(name, connex=sfGetCon(), verbose=FALSE) { if (missing(name)) { current <- getSnowflakeWH(connex=connex, refresh=TRUE) stop ("warehouse name is missing from sfWarehouseOff(name). It must be set explicitly.\nHINT: The current warehouse is '", current, "'", call.=FALSE) } qry <- sprintf("ALTER WAREHOUSE %s SUSPEND", name) sfQry(qry, connex=connex, verbose=verbose, return_query=FALSE) } if (FALSE) { .sf_clean_wh_input("load_wh", size="xs") .sf_clean_wh_input(size="L", type="Enter") .sf_clean_wh_input(size="L", auto_resume="FALSE", auto_suspend_minutes=15) } ## TESTING sfCreateWH() if (FALSE) { sfCreateWH("BI", just_query=TRUE) sfCreateWH("BI", just_query=TRUE, showWarnings=FALSE) sfCreateWH("BI", size="M", just_query=TRUE) sfCreateWH("BI", size="M", type="std", just_query=TRUE) sfCreateWH("BI", size="M", type="ent", just_query=TRUE) ## ERROR sfCreateWH("BI", size="Md", just_query=TRUE) if (FALSE) { sfCreateWH("BI", size="M", type="std", just_query=TRUE) sfCreateWH("TesT", size="XS", type="std", quote_name=TRUE, just_query=TRUE) sfCreateWH("TesT", size="XS", type="std", quote_name=TRUE, just_query=FALSE) } } sfCreateWH <- function(name, size=NULL, type='STANDARD', auto_suspend_minutes=NULL, auto_resume=FALSE, initially_suspended=TRUE, comment=NULL , replace=FALSE, if_not_exists=TRUE, just_query=FALSE, quote_name=FALSE, showWarnings=TRUE, connex=sfGetCon(), verbose_qry=TRUE) { ## Reference https://documentation.snowflakecomputing.com/manuals/sql-reference/ddl-virtual-warehouse.html?#create-warehouse ## ## NOTE: No need to check name against existing warehouses. This is handled by the SQL ## ## quote_name :: setting to TRUE wraps name in 'single quotes'. This has the effect of making the warehouse name case SENSITIVE and must always be referred to exactly is.char_of_length1(name, fail.if.not=TRUE) if (showWarnings && is.null(size)) warning ("size is NULL. It is recommended that it be set explicitly to either XS, S, M, L, XL or XXL") message("NOTE TO RICK: I am not sure you have implemented 'If NOT EXISTS' correctly\n Perhaps it should be CREATE OR REPLACE?") args_not_params <- c("name", "replace", "if_not_exists", "just_query", "quote_name", "showWarnings", "connex", "verbose_qry", "args_not_params") args <- collectArgs(except=args_not_params) params_string <- do.call(.sf.make_wh_params, c(args, with=TRUE)) if (quote_name) name <- sprintf("'%s'", name) else name <- toupper(name) qry <- paste ( "CREATE" , if (isTRUE(replace)) "OR REPLACE" , "WAREHOUSE" , name , if (isTRUE(if_not_exists)) "IF NOT EXISTS" # , pasteC(PARAMS, C=",") , params_string ) setQry(qry) if (just_query) return(qry) sfQry(qry, connex=connex, verbose=verbose_qry, warehouse=NULL, size=size, type=type) } ## EXAMPLES if (FALSE) { sfIncreaseWH("SCIENCE", size = "XL", comment="Increased Warehouse for Spotify Streams", just_query=FALSE) sfIncreaseWH("SCIENCE", size = "XL", comment="Increased Warehouse for Spotify Streams", just_query=TRUE) } sfIncreaseWH <- function(name, size=NULL, type=NULL, auto_suspend_minutes=NULL, auto_resume=NULL, initially_suspended=NULL, comment=NULL , replace=FALSE, if_not_exists=TRUE, just_query=FALSE, quote_name=FALSE, showWarnings=TRUE, connex=sfGetCon(), verbose_qry=TRUE) { ## Reference https://documentation.snowflakecomputing.com/manuals/sql-reference/ddl-virtual-warehouse.html?#create-warehouse ## EXAMPLE # ALTER WAREHOUSE "LOAD_WH" SET # WAREHOUSE_SIZE = 'LARGE' # AUTO_SUSPEND = 1800 # AUTO_RESUME = TRUE; is.char_of_length1(name, fail.if.not=TRUE) if (showWarnings && is.null(size)) warning ("size is NULL. It is recommended that it be set explicitly to either XS, S, M, L or XL") if (!missing(comment)) { warning("cannot use 'comment' with sfIncreaseWH() -- only with sfCreateWH()\ncomment will be ignored", call.=FALSE) rm(comment) } args_not_params <- c("name", "replace", "if_not_exists", "just_query", "quote_name", "showWarnings", "connex", "verbose_qry", "args_not_params") args <- collectArgs(except=args_not_params) params_string <- do.call(.sf.make_wh_params, c(args, set=TRUE, showWarnings=!missing(auto_resume) || !missing(initially_suspended))) if (quote_name) name <- sprintf("'%s'", name) else name <- toupper(name) qry <- paste ( "ALTER WAREHOUSE" , name , params_string ) setQry(qry) if (just_query) return(qry) sfQry(qry, connex=connex, verbose=verbose_qry, warehouse=NULL, size=size, type=type) } .sf_ILIKE_to_lower <- function(x) { gsub("(\\s)([A-Za-z\\._]+?)(\\s+)ILIKE(\\s+)(.+?)(\\s|$)", "\\1lower(\\2)\\3LIKE\\4lower(\\5)\\6", x) } .sf.make_wh_params <- function(size=NULL, type=NULL, auto_suspend_minutes=NULL, auto_resume=NULL, initially_suspended=NULL, comment=NULL, with_preface=FALSE, set_preface=FALSE, collapse="\n ", showWarnings=TRUE) { ## no name argument here ## USE FALSE FOR AUTO-SUSPEND OFF ## USE FALSE FOR AUTO-RESUME OFF ## with_preface :: whether to preface the return string with 'WITH ' or NOT # instructions from snowflake: # The parameters do not need to be separated by any keyword or notation. # A blank space between a parameter value and the next parameter is all that is needed. For example: # ALTER WAREHOUSE myvw SET warehouse_size=MEDIUM warehouse_type=ENTERPRISE; ## name arg removed from this function # if (!missing(name)) # name_param <- name %>% .sf_clean_name if (set_preface && with_preface) stop ("Only one of set_preface or with_preface should be TRUE") size_param <- .sf_clean_size(size) %>% {if (!is.null(.)) sprintf("%22s = %s", "WAREHOUSE_SIZE", .) } type_param <- .sf_clean_type(type) %>% {if (!is.null(.)) sprintf("%22s = %s", "WAREHOUSE_TYPE", .) } auto_suspend_param <- .sf_clean_auto_suspend_minutes(auto_suspend_minutes) %>% {if (!is.null(.)) sprintf("%22s = %s", "AUTO_SUSPEND", if (is.numeric(.)) format(. * 60, scientific=FALSE) else .) } auto_resume_param <- .sf_clean_auto_resume(auto_resume) %>% {if (!is.null(.)) sprintf("%22s = %s", "AUTO_RESUME", toupper(.)) } initially_param <- .sf_clean_initially_suspended(initially_suspended) %>% {if (!is.null(.)) sprintf("%22s = %s", "INITIALLY_SUSPENDED", .) } comment_param <- .sf_clean_comment(comment) %>% {if (!is.null(.)) sprintf("%22s = %s", "COMMENT", .) } if (is.null(auto_resume)) { rm(auto_resume_param) if (showWarnings) warning ("for non auto-resume use auto-resume = FALSE") } if (is.null(initially_suspended)) { rm(initially_param) if (showWarnings) warning ("for non auto-resume use auto-resume = FALSE") } e <- environment() args <- ls(pattern="_param$", envir=e) names(args) <- args ret <- args %>% lapply(function(a) get(a, inherit=FALSE, envir=e)) %>% unlist(use.names=FALSE) if (length(ret) && all(sapply(ret, is.null))) { warning("ret in .sf.make_wh_params() is a list of length ", length(ret), " and is all NULL -- this is unexpected", call.=FALSE) ret <- NULL } if (!length(ret)) return(NULL) preface <- if (with_preface) "\nWITH" else if (set_preface) "\nSET" else "" # ret <- c(ifelse(with_preface, "\nWITH", ""), ret) ret <- c(preface, ret) return (paste(ret, collapse=collapse)) } .sf_clean_wh_input <- function(name, size, type, auto_suspend_minutes, auto_resume, initially_suspended, comment, quote_name=FALSE) { ## Checking user input on warehouse parameters ## returns a named list of the same parameters passed ## ## Snowflake Parameters ## AUTO_SUSPEND ## Specifies the period of inactivity (in seconds) after which a warehouse will be automatically suspended. A NULL value specifies that the warehouse is never automatically suspended due to inactivity. ## AUTO_RESUME ## Specifies whether to automatically resume a warehouse when it is accessed. Valid values are ‘TRUE’ and ‘FALSE’. ## INITIALLY_SUSPENDED ## Specifies whether the warehouse is created initially in suspended state. Valid values are ‘TRUE’ or ‘FALSE’. ## TODO: auto_suspend_minutes is the only oddball here. ## Specifically, the input is in minutes, but the output is in seconds. ## This isn't completely terrible, except that the return list has the name auto_suspend_minutes ## No arguments passed if (length(as.list(match.call())) == 1) { warning ("No arguments passed to .sf_clean_wh_input") return (list()) } ## Check valid input for name if (!missing(name) && !is.null(name)) { # if (is.null(name)) # stop ("warehouse name cannot be NULL\nHINT: This may be due to options('snowflake_defaultwh') not being set. Try running setSnowflakeOptions()") if (!is.char_of_length1(name, fail=FALSE, showWarnings=FALSE)) stop ("warehouse name should be a character of length 1") validNames <- sfShowWarehouses(justnames=TRUE) if (!any(grepl(name, validNames, ignore.case=!quote_name))) stop("'", name, "' is not a valid warehouse name\n\nHINT: valid names are: ", paste(validNames, collapse=", ")) if (quote_name) name <- sprintf("'%s'", name) else name <- toupper(name) } if (!missing(size) && !is.null(size)) { is.char_of_length1(size, fail=TRUE) if (tolower(size) %in% tolower(.warehouse_sizes_old)) size <- names(.warehouse_sizes_old)[tolower(size) == tolower(.warehouse_sizes_old)] size <- match.arg(toupper(size), choices=toupper(.warehouse_sizes)) } if (!missing(type) && !is.null(type)) { is.char_of_length1(type, fail=TRUE) if (substr(toupper(type), 1, 3) == "STD") type <- "standard" type <- match.arg(toupper(type), choices=c("STANDARD", "ENTERPRISE")) } ## Clean up auto_suspend_minutes if (!missing(auto_suspend_minutes)) { if (is.character(auto_suspend_minutes)) auto_suspend_minutes <- toupper(auto_suspend_minutes) # if (is.null(auto_suspend_minutes) || identical(auto_suspend_minutes, FALSE)) if (identical(auto_suspend_minutes, FALSE)) auto_suspend_minutes <- "NULL" else if (is.numeric(auto_suspend_minutes)) auto_suspend_minutes <- auto_suspend_minutes * 60 if (!is.null(auto_suspend_minutes)) { if (is.na(auto_suspend_minutes) || (!is.numeric(auto_suspend_minutes) && !identical(auto_suspend_minutes, "NULL"))) stop ("auto_suspend_minutes should be either a number or 'NULL'") } } ## Function for cleaning logical values. func_cleanTF <- . %>% ifelseNULL(., yes=FALSE, no=.) %>% {if (!is.logical(.) && !is.list(.)) as.logical(.) else . } ## Clean up auto_resume -- only valid values are TRUE and FALSE if (!missing(auto_resume)) { # if (is.null(auto_resume)) # auto_resume <- FALSE # if (!is.logical(auto_resume)) # auto_resume <- as.logical(auto_resume) # if (!auto_resume %in% c(TRUE, FALSE)) # stop ("auto_resume should be either TRUE or FALSE") auto_resume %<>% func_cleanTF if (!is_T_or_F(auto_resume, vectorize=FALSE, null_ok=FALSE, fail.if.not=FALSE, showWarnings=FALSE)) stop ("auto_resume must be either TRUE or FALSE") } ## Clean up initially_suspended -- only valid values are TRUE and FALSE if (!missing(initially_suspended)) { # if (is.null(initially_suspended)) # initially_suspended <- FALSE # if (!is.logical(initially_suspended)) # initially_suspended <- as.logical(initially_suspended) # if (!initially_suspended %in% c(TRUE, FALSE)) # stop ("initially_suspended should be either TRUE or FALSE") initially_suspended %<>% func_cleanTF if (!is_T_or_F(initially_suspended, vectorize=FALSE, null_ok=FALSE, fail.if.not=FALSE, showWarnings=FALSE)) stop ("initially_suspended must be either TRUE or FALSE") } ## Clean up comment -- should be string of length 1 if (!missing(comment) && !is.null(comment)) { is.char_of_length1(comment, fail=TRUE) if (!grepl("^\\s*'", comment) && !grepl("'\\s*$", comment)) comment <- sprintf("'%s'", comment) } args <- names(as.list(match.call())[-1L]) args <- setdiff(args, "quote_name") names(args) <- args ret <- lapply(args, function(a) get(a)) ## All inputs should be of length 1 lengths <- sapply(ret, length) if (any(lengths > 1)) { stop("All warehouse parameter inputs should be of length 1 or NULL.\nOffender(s) is/are: ", paste(names(ret)[lengths != 1], collapse=" ")) } return(ret[]) ## OLD: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ## ## collectArgs does not work with missing arguments ## ## Thus, manually grab the list of objects. use eval-parse to check if the arg is missing. ## ## If it is missing, return a dummy variable, .__drop_me__, which will then be filtered against ## args <- ls() ## names(args) <- args ## .__drop_me__ <- ".___argument_missing___" ## e <- environment() ## ret <- lapply(args, function(.a) eval(parse(text=sprintf("if (!missing(%s)) %1$s else .__drop_me__", .a)), envir=e)) ## ## ## return values that are not .__drop_me__ ## return(ret[!sapply(ret, identical, .__drop_me__)]) ## OLD: ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ } ## USED FOR CREATING the .sf_clean_* functions BELOW if (FALSE) { ## White space needs to be cleaned lapply(names(formals(.sf_clean_wh_input)), function(a) ".sf_clean_%s <- function(%1$s) { if (missing(%1$s)) return(NULL) .sf_clean_wh_input(%1$s = %1$s) [[1L]] }" %>% sprintf(a) %T>% cat(fill=TRUE, "\n") %>% {.} ) } .sf_clean_name <- function(name) { if (missing(name)) return(NULL) .sf_clean_wh_input(name = name) [[1L]] } .sf_clean_size <- function(size) { if (missing(size)) return(NULL) .sf_clean_wh_input(size = size) [[1L]] } .sf_clean_type <- function(type) { if (missing(type)) return(NULL) .sf_clean_wh_input(type = type) [[1L]] } .sf_clean_auto_suspend_minutes <- function(auto_suspend_minutes) { if (missing(auto_suspend_minutes)) return(NULL) .sf_clean_wh_input(auto_suspend_minutes = auto_suspend_minutes) [[1L]] } .sf_clean_auto_resume <- function(auto_resume) { if (missing(auto_resume)) return(NULL) .sf_clean_wh_input(auto_resume = auto_resume) [[1L]] } .sf_clean_initially_suspended <- function(initially_suspended) { if (missing(initially_suspended)) return(NULL) .sf_clean_wh_input(initially_suspended = initially_suspended) [[1L]] } .sf_clean_comment <- function(comment) { if (missing(comment)) return(NULL) .sf_clean_wh_input(comment = comment) [[1L]] } # ------------------------------------------ # sfQry <- function(qry , connex=sfGetCon(quiet_connex=!verbose.connex) , wh=warehouse # , warehouse=getOption("snowflake_currentwh", getOption("snowflake_defaultwh")) , warehouse=getSnowflakeWH() , to.dt=exists("as.data.table") , tz_blanks_to="UTC" , idcols_convert=FALSE , refresh_con_if_expired = TRUE , fail_on_query_error = TRUE , update_last_ran = grepl("SELECT|UPDATE|INSERT", qry, ignore.case=TRUE) , check_warehouse=FALSE , verbose.max.width=getOption("width", 80) * 0.8 , verbose.max.lines=22L , verbose.add.dots=nchar(qry) < 5000 , verbose.shortCircuit=TRUE , verbose.indentAnd=grepl("\\bOR\\b", qry) , verbose.indentOr=FALSE , verbose=TRUE , verbose.connex=!isTRUE(getOption("sfGetCon_quiet_connex")) , verbose.key=verbose , verbose.firstOfMonth=TRUE ## Used only for details updating , size=NULL , type=NULL , file=NULL , ... ) { if (!missing(file)) { if (!missing(qry)) stop("Cannot use both qry and file in sfQry()") ## Note that sqlFileToQry checks if file exists qry <- sqlFileToQry(file=file, clear=TRUE) } ## Make sure package is loaded if (!.is_RODBC_loaded() || .is_POSTGRES_loaded()) .DBlib("snowflake") ## force the connenction early force(connex) if (length(qry) != 1) stop ("qry has length ", length(qry), ". It should have length 1. Use sapply or similar") ## The odbc driver has a bug where front loading whitespace causes issues qry %<>% trim() %>% gsub("\\s*;\\s*$", ";", .) ## Certain Keywords cannot be column names on their own, without quotes keywords <- c("rows", "count") for (w in keywords) { pat <- sprintf("(AS\\s*)%s(\\b)", w) rep <- sprintf("\\1\"%s\"\\2", toupper(w)) qry %<>% gsub(pat, rep, .) } if (!inherits(connex, "RODBC")) stop ("connex is not of type 'RODBC'") ## Cleanup warehouse argument missing_wh <- missing(wh) && missing(warehouse) wh %<>% .sf_clean_name if (!missing_wh && !is.null(wh)) { if (!.sf_check_if_current("wh", name=wh)) setSnowflake(wh=wh, connex=if (!isConExpired(connex)) connex else sfGetCon(set_defaults=FALSE)) } ## TODO - wh - check, turn on, etc. ## ...... wh ...... ?? if (check_warehouse) { s.t({ current_wh <- getSnowflakeWH(connex=connex, refresh=TRUE, verbose=FALSE) if (!is.null(wh) && isTRUE(wh != current_wh)) { message (" %% reminder to RICK -- something about current_wh in sfQry() %% ") setSnowflake(wh=wh, connex=connex) current_wh <- getSnowflakeWH(connex=connex, refresh=TRUE, verbose=FALSE) if (current_wh != wh) stop ("warehouse did not take.\nCurrent is '", current_wh, "'\nExpected is '", wh, "'\n") } }, title = "Checking Warehouse") } else { current_wh <- getSnowflakeWH(refresh=FALSE, connex=connex, verbose=FALSE) } if (verbose) { hr <- paste0(pasteR(verbose.max.width+3), "\n") cat("\n\n\t Running query on snowflake using warehouse '", ifelseNULL(wh, current_wh, wh), "'. [Began at ", timeStamp(frmt="%R %p") ,"] \n ", hr , verboseQry(qry, max.lines=verbose.max.lines, max.width=verbose.max.width, shortCircuit.ifendl.detected=verbose.shortCircuit, indentAnd=verbose.indentAnd, indentOr=verbose.indentOr, add.dots=verbose.add.dots) , "\n ", hr, sep="") } ## DEBUGGING browser(expr=inDebugMode("sfQry"), text="in sfQry() before executing query") ## SCRATCH CODE. ## TODO 2015-05-18 Attempt to ship query without waiting for response, so we can increase WH size automagically if (FALSE) { sfShowTables() rr <- odbcQuery(connex, qry) rt <- sqlGetResults(connex) } ## EXECUTE QUERY -- OR TRY TO, AT LEAST ret <- try(sqlQuery(connex, qry, stringsAsFactors=FALSE)) last_ran <- now() ## force last_ran to evaluate force(last_ran) ## Update the last_ran timestamp if (update_last_ran) sfUpdateWH_QryLastRan(name=current_wh, size=size, type=type, last_ran=last_ran) ## IF con IS EXPIRED, REFRESH AND TRY AGAIN pat.expired <- c("first argument is not an open RODBC channel", "Authentication token has expired") %>% regOr if (refresh_con_if_expired) { if (isErr(ret) || (is.character(ret) && any(grepl(pat.expired, ret)))) { message("connex was expired, refreshing then trying the query again") connex <- sfGetCon(refresh=TRUE) ret <- sqlQuery(connex, qry, stringsAsFactors=FALSE) } } ## CHECK FOR ERROR if (is.character(ret) && length(ret) < 5 && any(grepl("error", ret, ignore.case=TRUE))) return(sfErrorParse(qres=ret, qry=qry, fail_on_query_error=fail_on_query_error)) ## ELSE, NO ERROR, continue. ## Convert to data.table; wrap in try() so as to not fail after having waited for the query to run if (to.dt) { try (ret <- as.data.table(ret)) try (setnamestolower(ret)) s.t(title="Check for and convert timezones", quiet.if.less.than=1, expr={ if (!is.null(tz_blanks_to)) try({ for (col in names(ret)) { ## POSIX cols that have no timezone are assigned tz_blanks_to, which by default is "UTC" if (isTRUE(is.POSIXct(ret[[col]])) && identical(attr(ret[[col]], "tzone"), "")) setattr(ret[[col]], "tzone", tz_blanks_to) } }) ## // closes try() }) ## // closes s.t() s.t(title="Convert ID Cols", quiet.if.less.than=1, expr={ if (isTRUE(idcols_convert)) try({ setIDCols(ret) }) ## // closes try() }) ## // closes s.t() } return(ret[]) } # ------------------------------------------ # sfErrorParse <- function(qres, qry, return_error="auto", return_query="auto", fail_on_query_error=FALSE) { ## ARGS: ## qres - the raw return from sqlQuery() ## qry - the original query (often included in qres. It will be gsubbed out) ## return_error -- if FALSE, returns NULL, if TRUE returns the parsed error ## Only returns the error if the erro is unknown. Otherwise returns NULL # Most errors are of the form # # [1] " SQL compilation error:\n" # # [2] "[RODBC] ERROR: Could not SQLExecDirect ''" # # warehouse Errors are of the form # # [1] "22000 90063 Invalid state. warehouse 'LOAD_WH' cannot be resumed since it is not suspended." # # [2] "[RODBC] ERROR: Could not SQLExecDirect 'ALTER WAREHOUSE LOAD_WH RESUME USING WAREHOUSE_SIZE = SMALL ;'" # # Other errors include: # # HY000 50110 [Snowflake][Support] (50110) Conversion from string to number failed due to overflow with value: nc_qr <- nchar(qry) if (mnchar(qres) < 700) { message("\n", pasteR("*", 60), "\nHere is the original error\n\n") dput(qres) message("\n", pasteR("*", 60), "\n") } ## Check for a simple "Warehouse is off error" pat_whoff <- "(.*Warehouse )('.+')( is suspended.*)" if (grepl(pat_whoff, qres[[1]])) { wh_name <- gsub(pat_whoff, "\\2", qres[[1]]) msg <- sprintf("Warehouse %s is off. To resume it, run\n sfWarehouseOn(%1$s)", wh_name) if (fail_on_query_error) stop(msg, call.=FALSE) warning(msg, "\n\n(returning original error)", call.=FALSE) returnin(invisible(qres)) } pat.known_errors <- c("SQL compilation error:\\n", "Invalid state. ", "Conversion from") pat.line1 <- regOr(pat.known_errors, escape=FALSE) pat.line2 <- "[RODBC] ERROR: " Error_header <- "ERROR SF SQL:: " ## TEMPORARY -- CONFIRM I HAVE THE REGEX CORRECT if (grepl("SQL compilation error", qres[[1]])) { if (grepl(pat.line1, qres[[1]], fixed=FALSE)) message ("REGEX IN sfErrorParse() is [OK]") else message ("REGEX IN sfErrorParse() is WRONG FOR : \n", pasteR(60), "\n", qres[[1]], "\n", pasteR(60), "\n") message("\n~~~~~~~ vvvv [ real error parsing below ] vvvv ~~~~~~~~~\n") } # CHECK FOR SERVERS ...... ## Not enough servers to start a warehouse # CHECK FOR SERVERS ...... if (grepl("Maximum number of servers for the account exceeded.")) { # CHECK FOR SERVERS ...... # CHECK FOR SERVERS ...... } ## We expect that an error be of length 2 and to be of a certain pattern. ## If it fails to match this criteria, inform the user. if (length(qres) != 2 || !(grepl(pat.line1, qres[[1]], fixed=FALSE) && grepl(pat.line2, qres[[2]], fixed=TRUE))) { message ("in sfErrorParse() -- received an error that was not as expected") ## replace the original query ret <- gsub(pattern=qry, repl="< ORIGINAL QRY WENT HERE >", x=qres, fixed=TRUE) msg <- paste(Error_header, paste(ret, collapse="\n ")) ## For 'auto' return_error, the criteria to NOT return the error is that the error is two strings and the second is the expected pattern ## In other words, we simply have not anticipated the specific details of the error if (identical(return_error, "auto")) return_error <- (length(qres) == 2 && grepl(pat.line2, qres[[2]])) if (identical(return_query, "auto")) return_query <- nc_qr < 500 ## PARSE THE RROR AS EXPECTED } else { ## If the second line is as expected, remove it full_line2 <- sprintf("%sCould not SQLExecDirect '%s'", pat.line2, qry) qres[[2]] <- gsub(pat=full_line2, "", x=qres[[2]], fixed=TRUE) qres <- trim(qres) if (qres[[2]] == "") qres <- qres[[1]] if (length(qres) == 1) { ret <- strsplit(qres, pat.line1) [[1]] %>% trim %>% setNames(nm=c("error_no", "error_msg")) %>% rev msg <- sprintf("%s %s (%s)", Error_header, ret[["error_msg"]], ret[["error_no"]]) } else { ret <- qres msg <- sprintf("%s ", Error_header, paste(qres, collapse="\n ")) } if (identical(return_query, "auto")) return_query <- nc_qr < 250 } if (isTRUE(return_query)) { msg <- sprintf("%s\n\n ---------- * ORIGINAL QUERY: * ---------- \n%s", msg, sub("^\n", "", x ={if (nc_qr < 1000) verboseQry(qry) else qry})) } ret[["original_qry"]] <- qry ## set error attribute setattr(ret, "isErr", TRUE) if (fail_on_query_error) { cat(msg, "\n (duplicated message in case error is caught)\n") stop(msg, call.=FALSE) } cat(msg, fill=TRUE, sep="") return(invisible(ret)) } getSFCommentFooter <- function(user=getOption("snowflake_uid", default="")) { if (user == "rsaporta") user <- "Rick Saporta" sprintf("[Created by %s via R]", user) } ## DEPRECATED # useSnowflake <- function(stop_use=FALSE, wh=NULL, dbname=NULL, schema=NULL) { # warning("DEPRECATED FUNCTION. setSnowflake()") # setSnowflake(wh=wh, dbname=dbname, schema=schema) # }