# -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- # # ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ # # # # File Name : DB_Utils.r # # Last Updated Funclist : 06 Apr 2015, 3:01 PM (Monday) # # # # Author Name : Rick Saporta # # Author Email : RickSaporta@gmail.com # # Author URL : www.github.com/rsaporta # # # # Packages Called : RMySQL # # Packages Used via NS : NA # # # # ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ # # # # verbose.dbcon.on ( ) # # giveMeACon ( drv.str=getDBdriver(default="PostgreSQL"), pos=2, assignIfMissing=TRUE # # , cluster=getOption("DB.cluster") # # , verbose=getOption("verbose.DBcon", FALSE), envir=getOption("con.environ", default=parent.frame(pos)) ) # # getDB ( drv.str=getDBdriver(), verbose=getOption("verbose.DBcon", FALSE) ) # # qDesc ( tbl, schema, cluster=NULL ) # # qDescTable ( ... ) # # qUpdatePerms ( tbl, schema, user, group, permissions=if (!missing(tbl)) "SELECT" else "USAGE" # # , showWarnings=TRUE, failOnError=FALSE, cluster=NULL, verbose=TRUE # # , showStatusUpdate="-- DEPRECATED --", verbose.qry=FALSE ) # # qTableDetails ( tbl, minDate=NULL, cluster=NULL, verbose=TRUE ) # # getRevenueAndUnitsColsForTbl ( tbl, nms=c("revenue", "units") ) # # getDateColForTBL ( tbl ) # # qMaxDate ( tbl, dateCols=getDateColForTBL(tbl_like) # # , schema=ifelse(tbl_like %in% c("analytics", "accounting", "mbworkbook", "GPU"), "bi", "production"), tbl_like=gsub("analytics_.+", "analytics", gsub("accounting_.+", "accounting", tbl)), min=FALSE # # , verbose=TRUE, cluster=NULL, refresh.DT.periods=FALSE, simplify=TRUE ) # # qRowCount ( tbl, schema=NULL, dbname=NULL, cluster=NULL, force.numeric=FALSE, verbose=FALSE ) # # verbose.dbcon.off ( ) # # qExtractCols ( pat.regex, tbl, schema=NULL, dbname=NULL, verbose=FALSE, cluster=NULL, showAllOnNoMatch=TRUE ) # # qShowCols ( tbl, schema=NULL, dbname=NULL, namelike=NULL, sort=TRUE, include.types=FALSE, returnRaw=FALSE # # , similar=TRUE, verbose=FALSE, simplify=TRUE, cluster=NULL ) # # tableNameClean ( tbl, schema=NULL ) # # headDB ( tbl, n=limit, connex=giveMeACon(verbose=inVerboseMode.dbcon() || verbose) # # , justHeader=FALSE, whereIn=NULL, dateCol, minDate=NULL, maxDate=NULL # # , colsUsing="*", schema=NULL, limit=5L, dont.drop.anything=FALSE # # , check.table.perms=TRUE, cluster=NULL, verbose=FALSE ) # # nrowDB ( tbl, connex=giveMeACon(verbose.dblib=FALSE), whereClause=NULL ) # # funcOfCols ( funcName, cols ) # # minDB ( tbl, cols, connex=giveMeACon(verbose.dblib=FALSE) ) # # maxDB ( tbl, cols, connex=giveMeACon(verbose.dblib=FALSE) ) # # getKeyVals ( DT ) # # inVerboseMode.dbcon ( ) # # not ( vals, col, keyVals ) # # extractTablenameFromQry ( qry, debug=FALSE ) # # sqlFileToQry ( file, clearComments=clear, clearDoubleBlankLines=clearComments, clear=FALSE # # , replaceRStyleComments=TRUE ) # # print.query ( x, max.width=getOption("width", 80) - 5L, max.lines=45L, ... ) # # sql1stOfMonth ( dateCol, as=NULL, tbl=NULL, schema=NULL ) # # DBlib ( pkgs=c("PostgreSQL", "MySQL"), unload=FALSE, verbose=TRUE ) # # schemaSplit ( stbl, sep=".", nms=c("schema", "table"), schema.default="production" ) # # schemaPaste ( schema, tbl, sep="." ) # # qCurrUser ( ) # # qMyQs ( user, currentUser=isTRUE(missing(user)) ) # # dbListAll ( drv=MySQL() ) # # qShowQueries ( user, currentUser=isTRUE(missing(user)), rdsdb_ignore=TRUE, showWarnings=TRUE ) # # qWhatsRunning ( user, currentUser=isTRUE(missing(user)), rdsdb_ignore=TRUE, showWarnings=TRUE ) # # qWhatsRunning_OLD ( user=NULL, currentUser=mine, mine=FALSE ) # # qKill ( id, force=FALSE, verbose=TRUE, user=NULL, currentUser=mine, mine=FALSE, cluster=NULL ) # # qCanRead ( tbl, user=NULL, schema=NULL, connex=giveMeACon(verbose.dblib=FALSE), checkTableExists=FALSE # # , showWarnings=TRUE, cluster=NULL, verbose=FALSE ) # # tableToView ( string, ignore.case=TRUE ) # # qShowViews ( where=NULL, schema="bi", ..., what="views", cluster=NULL, verbose=FALSE ) # # qShowTables ( where=NULL, schema=c("production", "bi", "ds_scratch"), namelike=NULL # # , rowCount=("bi" %in% schema) && (!justnames), showDetails=FALSE, justnames=FALSE # # , filterByPerm=FALSE, includePerm=FALSE, connex=giveMeACon(verbose=verbose) # # , verbose=FALSE, what=c("tables", "views"), showWarnings=TRUE, cluster=NULL # # , order=c("schema", "cluster", "name") ) # # TtoV ( string ) # # qFindView ( ..., use.view=TRUE ) # # dbDisconnectAll ( drv, showWarnings=TRUE ) # # qFindTable ( pattern, schema=NULL, ignoreCase=ignore.case, ignore.case=TRUE # # , exact=!any(grepl("\\%", pattern)), cluster=NULL, use.view=FALSE ) # # qViewExists ( ..., cluster=NULL, use.view=TRUE ) # # qTableExists ( tbl, schema=NULL, dbname=NULL, exact=TRUE, ignoreCase=TRUE, cluster=NULL, use.view=FALSE ) # # makeProperColName ( tbl, col, AS.prefix=tbl, AS.suffix=NULL, sep="_", verbose=FALSE ) # # getGenresForAllupcs ( id.col.AS="upc", dontExecute.justReturnQry=FALSE, verbose=TRUE # # , verbose.max.width=98L + (200 * dontExecute.justReturnQry), cluster=NULL ) # # getReleaseDates ( values, id.col="releaseId", id.col.AS=id.col, numericValues=FALSE, sep="__" # # , origin="1970-01-01", genre=TRUE, dontExecute.justReturnQry=FALSE, verbose=TRUE # # , verbose.max.width=98L + (200 * dontExecute.justReturnQry) # # , minimal=is.null(values), cluster=NULL ) # # cleanReleaseDatesDT_ ( DT.releaseDates # # , id.col.AS=ifelse("upc" %in% names(DT.releaseDates), "upc", "releaseId"), minimal=FALSE # # , origin="1970-01-01", failOnMissingCols=!minimal, showWarnings=!minimal # # , verbose=FALSE ) # # infinitDBNotWorking.metas ( DT.nm ) # # qShowErrs ( all=FALSE, verbose=TRUE, userid=NULL, lastN=1, cluster=NULL ) # # qShowDist ( cluster=NULL ) # # getDB.mysql ( dbname="dw", host=getOption("db.defaulthost") ) # # qGetAll ( tbl, filter=list(), limit=NULL, dont.check.table=FALSE # # , confirmed.single.quotes.are.escaped=FALSE, verbose=TRUE, ... ) # # QuoteAndFlatten ( x ) # # createTblFromQry ( qry, tbl, schema, cluster=NULL, verbose=TRUE ) # # qRenameTable ( tbl.old, tbl.new=tbl.old, schema=NULL, schema.old=schema, schema.new=schema.old # # , cluster=NULL, verbose=TRUE, verbose.qry=FALSE, msg.pre=NULL ) # # makeDropQryFromCreate ( qry.create, instruction=c("DROP", "TRUNCATE") ) # # qDiskUsage ( limit=NULL, verbose=TRUE ) # # getDB.old ( drv.str=c("PostgreSQL", "MySQL"), dbname="dw", host=getOption("db.defaulthost") # # , drv.explicit=FALSE, cluster=getOption("DB.cluster"), verbose=FALSE ) # # isConExpired ( con ) # # isConOfTypeX ( con, X, strict.match=FALSE, ignore.case=!strict.match ) # # # # # # # # ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ # # -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- # # example: # -- #' The length of a string (in characters). # example: # -- #' # example: # -- #' @param string input character vector # example: # -- #' @return numeric vector giving number of characters in # example: # -- #' each element of the character vector. Missing strings have # example: # -- #' missing length. # example: # -- #' @keywords character # example: # -- #' @seealso \code{\link{nchar}} which this function wraps # example: # -- #' @export # example: # -- #' @examples # example: # -- #' str_length(letters) # example: # -- #' str_length(c("i", "like", "programming", NA)) # example: # -- str_length <- function(string) { # example: # -- string <- check_string(string) # example: # -- nc <- nchar(string, allowNA = TRUE) # example: # -- is.na(nc) <- is.na(string) # example: # -- nc # example: # -- } #---------------------------------------------------------------------------------# # #---------------------------------------------------------------------------------# ## PERSONAL ## .utilsDir.tmp <- getOption("utilsDir", paste(getOption("baseDir", "~/git/misc/rscripts"), "utils", sep="/") ) if (is.null(.utilsDir.tmp)) { warning("options('utilsDir') is NOT set. It should have been set in .First().\nCannot source DB setters & getters.\n") } else { try( source (paste(.utilsDir.tmp, "DB_Utils", "DB_Utils_Setters_and_Getters.r", sep="/")) ) # try( source (as.path(.utilsDir.tmp, "DB_Utils_Setters_and_Getters.r")) ) } #' The default host address used by \code{getDB()} ## TODO: This might be outdated. not Sure ## in the DB_utils_settings file,there is an options ## setting 'db.defaulthosts' which is a named list ## eg: options("db.defaulthosts"=list(PostgreSQL="xx.xxx.xxx", MySQL="xx.xxx.xxx")) ## TODO: deprecate this options("db.defaulthost" = getOption("db.defaulthosts")[[1]] ) options("db.defaulthost.MySQL" = getOption("db.defaulthosts")[["MySQL"]]) ## Current driver if (is.null(getOption("db.driver"))) options("db.driver"="PostgreSQL") if (is.null(getOption("DB.cluster"))) options("DB.cluster"=NULL) #' We use giveMeACon often. We don't want its verbose output all of the time. Thus an additional option just for this function options(verbose.givemeacon=FALSE) #' Envrionment where to store the DB connection. Default is globalenv() options(con.environ=globalenv()) verbose.dbcon.on <- function() { options("verbose.DBcon" = TRUE) } verbose.dbcon.off <- function() { options("verbose.DBcon" = FALSE) } inVerboseMode.dbcon <- function() { getOption("verbose.DBcon", FALSE) } #' Collection of wrapper functions for DBI functions #' List all available connections for a given driver (default is MySQL) #' #' @param drv a driver object, same as would be passed used in \code{dbListConnections(drv)} #' @return a list of all currently open connections on driver \code{drv}. #' @keywords con, connection, driver, DBI #' @seealso \code{\link{dbListConnections}} which this function wraps #' @export #' @examples #' dbListAll() #' dbListAll(dbDriver("ODBC")) dbListAll <- function(drv=MySQL()) dbListConnections(drv) #' Disconnects all available connections for a given driver (default is MySQL) #' #' @param drv a driver object, same as would be passed used in \code{dbListConnections(drv)} #' @return a list of disconnect results for each connection. #' @keywords con, connection, driver, DBI, disconnect #' @seealso \code{\link{dbListConnections}} and \code{\link{dbDisconnect}} which this function invokes #' @export #' @examples #' dbDisconnectAll() #' dbDisconnectAll(dbDriver("ODBC")) dbDisconnectAll <- function(drv, showWarnings=TRUE, snowflake_inuse=getOption("snowflake_inuse", default=FALSE)) { if (snowflake_inuse) return(invisible(NULL)) if (missing(drv) || is.null(drv)) { if (exists("PostgreSQL")) drv <- PostgreSQL() else if (exists("MySQL")) drv <- MySQL() ## a NULL value means that we're not necessarilly expecting a drv to exist else if (is.null(drv)) return(invisible(NULL)) else stop("No known driver found. Please specify one") } ## TRY allConnections <- dbListConnections(drv) if (!length(allConnections) && exists("con")) { message("There are no opened connections to close") return(invisible(NULL)) } ## ELSE: Look for uncleared results, then close connections ## Check for uncleared results before breaking the connections unclearedResults <- lapply(allConnections, dbListResults) if (any(wh.unclear <- 0 != sapply(unclearedResults, length))) { warning("There were some pending results which will be cleared") cleared <- sapply(unclearedResults[wh.unclear], function(cc) any(sapply(cc, dbClearResult))) ## Confirm that cleared if (!all(cleared)) warning ("Internal Error inside dbDisconnectAll(): Something went wrong with clearning the results") } ## Disconnect closed <- sapply(allConnections, dbDisconnect) ## Confirm if (!all(closed)) warning ("Internal Error inside dbDisconnectAll(): Something went wrong with closing the connectinos") return(invisible(NULL)) } #' The length of a string (in characters). #' #' @param string input character vector #' @return numeric vector giving number of characters in #' each element of the character vector. Missing strings have #' missing length. #' @keywords character #' @seealso \code{\link{nchar}} which this function wraps #' @export #' @examples #' str_length(letters) #' str_length(c("i", "like", "programming", NA)) # -- old --- # # for infinitDB getDB.mysql <- function(dbname="dw", host=getOption("db.defaulthost")) { stop("EXECUTING THE WRONG FUNCTION !!! ") dbConnect(RMySQL::MySQL(), user="rsaporta", password="0rchrs01", dbname=dbname, host=host) } getDB.old <- function(drv.str=c("PostgreSQL", "MySQL"), dbname="dw", host=getOption("db.defaulthost"), drv.explicit=FALSE, cluster=getOption("DB.cluster"), verbose=FALSE) { # cluster is just used for redshift # check if explicitly given if (missing(drv.explicit)) drv.explicit <- (!missing(drv.str)) ## The accounting DB is on MySQL if (identical(cluster, "accounting") && missing(drv.str)) drv.str <- "MySQL" drv.str <- match.arg(drv.str) if (missing(host) && drv.str=="MySQL") host <- getOption("db.defaulthost.MySQL") ## Loads the required R packages for the DB connection we are using, also unloads the others .DBlib(pkgs=as.character(drv.str)) if (drv.str == "MySQL") { # if (missing(host)) warning("\n\nexecuting MySQL with default host: ", host, "\n\n") return(dbConnect(drv=RMySQL::MySQL(), user="rsaporta", password="0rchrs01", dbname=dbname, host=host)) } else if (drv.str == "PostgreSQL") { verboseMsg(verbose, "in `getDB()`.. inside of 'PostgreSQL' branch.", time=FALSE) ## get the settings according to the cluster db.settings <- getDBSettings(cluster=cluster) if (verbose) showDBsettings(cluster=cluster) ## dbhost <- db.settings[["dbhost"]] UID <- db.settings[["UID"]] PW <- db.settings[["PW"]] PT <- db.settings[["PT"]] DB <- db.settings[["DB"]] return(dbConnect(drv=PostgreSQL(), host=dbhost, port=PT, dbname=DB, user=UID, password=PW)) } else stop("invalid drv.str. Dont know how to handle that one.") } isConExpired <- function(connex) { ## TRUE :: means connex IS expired ## If there is an error with the connection, assume it expired # if (isErr(force(connex)) || isErr(capture.output(print(connex)))) if (isErr(force(connex))) return(TRUE) if (identical(connex, -1L)) return (TRUE) if (inherits(connex, "DBIConnection")) return(any(grepl("Expired", capture.output(connex)))) if (inherits(connex, "RODBC")) { is_valid <- RODBC:::odbcValidChannel(connex) is_expired <- !is_valid return(is_expired) } warning("Cannot determine if connex is expired or not -- returning FALSE") return(FALSE) } isConOfTypeX <- function(con, X, strict.match=FALSE, ignore.case=!strict.match) { # Wrapper function to check if con inherits class X. # However, it allows for X to be a partial class name, such as # 'PostgreSQL' instead of 'PostgreSQLConnection' # # ARGS: # strict.match : if TRUE, then con must inherit X. # if FALSE, then X must be a string in the class of con # ignore.case : only used when strict.match is FALSE. ignore.case is passed to grepl() if (length(con) != 1) stop ("'con' must be exactly length 1, it has length ", length(con), ".") if (strict.match) return(inherits(con, X)) return(any(grepl(X, is(con), ignore.case=ignore.case))) } giveMeACon <- function(drv.str=getDBdriver(default="PostgreSQL"), pos=2 , refresh=FALSE , assignIfMissing=TRUE , cluster=getOption("DB.cluster") # , envir=if(drv.str == "MySQL") globalenv() else getOption("con.environ", default=parent.frame(pos)) , envir=getOption("con.environ", default=parent.frame(pos)) , wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeSchema(), snowflake_inuse=getOption("snowflake_inuse", default=FALSE) , verbose.dblib=verbose , verbose=getOption("verbose.DBcon", FALSE) ) { ## Snowflake if (isTRUE(snowflake_inuse)) return(sfGetCon(refresh=refresh, assign=assignIfMissing, wh=wh, dbname=dbname, schema=schema)) # check if explicitly given drv.explicit <- (!missing(drv.str)) drv.str <- matchDriver(drv.str) ## Check: if pos was given explicit and it would have resulted in a different environment, warn user if (!missing(pos) && !identical(envir, parent.frame(pos))) warning("Argument 'pos' is ignored when getOption('con.environ') is not NULL.\nThe environ set in option differs from parent.frame(pos)") ## We use giveMeACon often. We don't want its verbose output all of the time. Thus an additional option just for this function verbose <- isTRUE(verbose) && getOption("verbose.givemeacon", default=TRUE) # if no option set, then presume not FALSE verboseMsg(verbose, "\n\t-- Getting a connection for drv.str == ", drv.str, " -- ", time=FALSE) ## Loads the required R packages for the DB connection we are using, also unloads the others .DBlib(pkgs=as.character(drv.str), verbose=verbose.dblib) browser(expr=inDebugMode(flag="con"), text="Checking giveMeACon() at top") ## TODO: Make sure con is of the right type if ( !refresh && exists("con", envir=envir) && ## There is an object `con` inherits(con, "DBIConnection") && ## It is a DBIConnection object !isConExpired(con) ## Its not expired ){ verboseMsg(verbose, "Existing 'con' found. Current value is ", capture.output(con), time=FALSE) ## Check that the connection is of the correct type ## TODO: I do not remember what the purpose of 'drv.explicit' was, so i removed it from the ifclause ## ' if (drv.explicit && isConOfTypeX(con, drv.str)) ' if (isConOfTypeX(con, drv.str)) { verboseMsg(verbose, "using existing 'con' (& exiting giveMeACon() )... \n", time=FALSE) return (con) } else{ verboseMsg(verbose, "'con' exists, but *NOT* using it. Closing it out and opening a new one", time=FALSE) dbDisconnect(get("con", envir=parent.frame())) } } # Else open a new connection, assing if flagged, the return it. verboseMsg(verbose, "Not using existing con. Creating new con....", time=FALSE) ret <- getDB(drv.str=drv.str, verbose=verbose) if (assignIfMissing) assign ("con", ret, envir=envir) return(ret) } # ---------------- # getDB <- function(drv.str=getDBdriver(), verbose=getOption("verbose.DBcon", FALSE)) { ## USER SHOULD NOT CALL THIS DIRECTLY ## User should use giveMeACon() ## ## This is the function that actually establishes the connection to the DB ## giveMeACon() uses this function. ## giveMeACon() first checks the status of any existing conenection relative to user needs ## and if a fresh connection is needed, giveMeACon() calls getDB() ## if (is.null(getDBsettings())) setDBsettings(drv.str) ## If it was set explicitly, match ittsar. if (!missing(drv.str)) { # OLD: drv.str <- matchDBdriver(drv.str) drv.str <- matchDriver(drv.str) } ## if drv.str is different from the current settings, then set ALL of the settings ## However, sometimes drv.str is a string and drv is a driver, thus we need to check those as well ## For now, this second part is done in a hacky fashion, simply for RMySQL current_drv <- getDBsettings()$drv if (!identical(drv.str, current_drv) && !(identical(current_drv, RMySQL::MySQL()) && drv.str %in% c("RMySQL", "MySQL")) ) { setDBall(drv.str) } else setDBdriver(drv.str) browser(expr=inDebugMode(flag="con"), text="in getDB() in DB_Utils.r.\n Check that the right settings are in getDBsettings()\n") if (verbose) showDBsettings() args <- getDBsettings() try({ if (!inherits(args$drv, "DBIDriver")) { if (args$drv == "MySQL") args$drv <- RMySQL::MySQL() else ## NOTE: 2016-Jan, Changes I made at start of month broke something; Possibly introduced the "R" preface. Removing it args$drv <- match.fun(removeText("^R", args$drv))() } }) # do.call(dbConnect, args) return(do.call(dbConnect, args)) } # -------------------------------- # qDesc <- function(tbl, schema, cluster=NULL) { tbl <- tableNameClean(tbl, schema=schema) Qry <- sprintf("SELECT * FROM pg_table_def WHERE tablename = '%s' AND schemaname = '%s'; ", tbl, schema) runQry(Qry, verbose=FALSE, cluster=cluster) } qDescTable <- function(...) { ## synonym function warning ("qDescTable has been deprecated.\n Use qDesc() instead") qDesc(...) } qUpdatePerms <- function(tbl, schema, user, group, permissions=if (!missing(tbl)) "SELECT" else "USAGE", showWarnings=TRUE, failOnError=FALSE, cluster=NULL, verbose.qry=FALSE, msg_sf=getOption("snowflake_msg_sf", TRUE), wh=getSnowflakeWH(), dbname=NULL, snowflake_inuse=getOption("snowflake_inuse", FALSE), verbose=TRUE) { if (snowflake_inuse) { verboseMsg(msg_sf, "Do not know how to update perms for snowflake. Are perms needed on the table level?", time=FALSE, func="message") return(NULL) } force(permissions) # if (!missing(showStatusUpdate)) { # warning ("showStatusUpdate has been deprecated. Use verbose instead") # verbose <- showStatusUpdate # } ## one of either 'user' or 'group' must be explicitly set if (missing(group) == missing(user)) stop("exactly one of either 'user' or 'group' must be explicitly set") if (missing(schema) && missing(tbl)) stop ("At least one of 'schema' or 'table' must be explicitly set") ug.using <- { if (!missing(group)) paste("group", group) else paste(user) } ts.using <- { if (missing(tbl)) sprintf("SCHEMA %s", schema) else sprintf("TABLE %s", schemaPaste(schema=schema, tbl=tbl) ) } len.u <- length(ug.using) len.t <- length(ts.using) if ((1 %ni% c(len.u, len.t)) && (len.u != len.t)) stop(sprintf("length of ug.using [ie, group or user] (%s) is not compatible with length of tbl (%s).\nThey should be the same length or one of them should be of length 1", len.u, len.t)) ## Create the query ### REMOVED 20141118 Qrys <- sprintf("GRANT %s on TABLE %s to %s;", permissions, schemaPaste(schema=schema, tbl=tbl), ug.using) Qrys <- sprintf("GRANT %s on %s to %s;", permissions, ts.using, ug.using) ## The names will be used in the messaging names(Qrys) <- sprintf("%s and %s", ts.using, ug.using) browser(expr=inDebugMode(c("qUpdatePerms")), text="in qUpdatePerms") for (nm in names(Qrys)) { verboseMsg(verbose, "updating permissions for", nm, time=FALSE) ## EXECUTE the query ret <- try(runQry(Qrys[[nm]], cluster=cluster, verbose=verbose.qry ), silent=TRUE) if (isErr(ret)) { msg <- paste("Permission update failed for", nm) if (failOnError) stop(msg) if (showWarnings) warning(msg) } } ## // end for loop } qTableDetails <- function(tbl, minDate=NULL, cluster=NULL, verbose=TRUE, period_convert=TRUE) { # if (any(wh <- tbl %ni% c("analytics", "accounting", "fact_analytics", "fact_sales"))) # stop ("Do not know which columns are needed for tbl '", commaSep(tbl[wh]), "'") if (!length(tbl)) return(tbl) if (verbose && length(tbl)>1 && length(cluster)>1) { cat(sprintf("running %i queries\n", length(tbl) * length(cluster))) } browser(expr=inDebugMode(c("qTableDetails")), text="in qTableDetails before qry selection") if (length(tbl) > 1) { verbose <- !missing(verbose) && verbose ret.tbl <- lapply(tbl, function(tt) qTableDetails(tbl=tt, minDate=minDate, cluster=cluster, verbose=verbose)) return(ret.tbl) } if (length(cluster) > 1) { verbose <- !missing(verbose) && verbose ret.cl <- lapply(cluster, function(cc) qTableDetails(tbl, minDate=minDate, cluster=cc, verbose=verbose)) ret.cl <- rbindlist(ret.cl) ret.cl[, SAME := all(sapply(ret.cl[, !"cluster", with=FALSE], areEqual))] return(ret.cl) } WHERE <- {if (!is.null(minDate)) sprintf(" WHERE download_activity_date >= '%s'" ,minDate) else ""} tbl <- tail(strsplit(tbl, "\\.")[[1]], 1) if (grepl("fact_accounting", tbl)) { warning("fact_accounting is not a table. Using fact_sales") tbl <- sub("_accounting", "_sales", tbl) } { if (FALSE) { "blank to allow moving arround / commenting out" } else if (grepl("^fact_analytics", tbl)) { cols <- c("Max Activity Date" = "download_activity_date") tbl <- paste0("production.", tbl) } else if (grepl("^fact_sales", tbl)) { cols <- c("max accounting period"="accountingperiodid", "max activity period"="activityperiodid") tbl <- paste0("production.", tbl) } else if (grepl("^analytics", tbl)) { cols <- c("Max Activity Date" = "download_activity_date", "download_accounting_month") tbl <- paste0("bi.", tbl) } else if (grepl("^accounting", tbl)) { cols <- c("accounting_month", "activity_month") tbl <- paste0("bi.", tbl) } else stop ("Do not know which columns are needed for tbl '", tbl, "'") } names(cols)[names(cols) == ""] <- cols[names(cols)==""] if (is.null(names(cols))) names(cols) <- cols colSelected <- commaSep(sprintf("MAX(%s) AS \"%s\"", cols, names(cols))) qry <- sprintf("SELECT %s, count(*) as rows FROM %s%s", colSelected, tbl, WHERE) if (verbose) cat(verboseQry(qry), "\n ", pasteR(52), "\n") ret <- runQry(qry, cluster=cluster, verbose=FALSE) if (period_convert) { periodCols <- names(cols)[grepl("periodid", cols)] if (length(periodCols)) addDateCols.periodid_(ret, periodCols=periodCols, showWarnings=TRUE) } ## put rows at end, right before tbl and cluster setcolorderpt(ret, endcols="rows") ret[, table := tbl][, cluster := valueIfNull(cluster, NA_integer_)] return(ret) } getRevenueAndUnitsColsForTbl <- function(tbl, nms=c("revenue", "units"), schema=NULL) { if (!is.null(schema)) warning("schema is not used in getRevenueAndUnitsColsForTbl()", call.=FALSE) if (length(tbl) > 1) return (lapply(tbl, getRevenueAndUnitsColsForTbl, nms=nms, schema=schema)) tbl %<>% tolower SalesAndUnitsCols.ll <- list( analytics = c(revenue="gest", units="paidunits" ) , analytics_track_level = c(revenue="gest", units="paidunits" ) , analytics_release_level = c(revenue="NULL", units="paidunits" ) , fact_analytics = c(revenue="royaltydollar", units="paidunits" ) , accounting = c(revenue="gross", units="units" ) , fact_sales = c(revenue="gross", units="sales" ) , staging_raw_itunes = c(revenue="royalty_price", units="units" ) # , staging_raw_spotify_v2 = c(revenue='NULL', units="COUNT(*)" ) ## this attempt at using NULL and count(*) is not a good idea; makeQry did not like. ) ## Warning message for analytics if (any(grepl("analytics", tbl))) message("You have requested Revenue and Units for an analytics table.\nRemember that analytics uses 'gest' (not royaltydollar) and this field is not necessarilly accurate.\nAdditionally, analytics_release_level does not have a revenue column") if (tbl %ni% names(SalesAndUnitsCols.ll)) stop ("Do not know the revenue & units columns for tbl='", tbl, "'\n\nHINT: edit the list in getRevenueAndUnitsColsForTbl() in the DB_Utils.r file", call.=FALSE) return(setNames(nm=nms, obj=SalesAndUnitsCols.ll[[tbl]])) } qDateInfo <- function(tbl, schema=NULL, dbname=NULL, wh=getSnowflakeWH(), snowflake_inuse=getOption("snowflake_inuse", default=FALSE), cluster=NULL, verbose=TRUE, whereIn=list(), ...) { ## Catch names like bak_staging_raw_spotify_v2 dateCol <- getDateColForTBL(tbl=gsub("(_bak|bak_)", "", tbl), schema=schema) nms.dateCol <- colNamesFromVector(gsub("::date", "_to_date", dateCol)) dst <- dbschematbl(dbname=dbname, schema=schema, tbl=tbl) whereClause <- makeWhereClause(whereIn=whereIn) if (!is.null(whereClause) && nchar(whereClause)) whereClause <- paste(" WHERE", whereClause) runQry(sprintf("SELECT MIN(%1$s) as min_%2$s, MAX(%1$s) as max_%2$s FROM %3$s %4$s", dateCol, nms.dateCol, dst, whereClause), cluster=cluster, snowflake_inuse=snowflake_inuse, wh=wh, verbose=verbose, ...) } getDateColForTBL <- function(tbl, schema, clean_tbl=TRUE, connex=giveMeACon(verbose.dblib=FALSE) , msg_sf = getOption("snowflake_msg_sf", TRUE) , snowflake_inuse = getOption("snowflake_inuse", default=FALSE) , verbose="auto" ) { force(connex) if (length(tbl) > 1) { if (identical(verbose, "auto")) verbose <- TRUE ARGS <- collectArgs(except="tbl") return (lapply(tbl, function(.t) do.call(getDateColForTBL, args=c(list(tbl=.t), ARGS)))) } if (identical(verbose, "auto")) verbose <- FALSE verboseMsg(verbose, "Processing Table ", schemaPaste(schema=if (missing(schema)) NULL else schema, tbl=tbl), minw=90) dateCol.ll <- list( analytics = c("activity_date") , fact_analytics = c("download_activity_date") , fact_analytics_error = c("downloaddate") , accounting = c("accounting_month") , fact_sales = c("accountingperiodid") , mbworkbook = c("accounting_month") , gpu = c("activity_month544") , staging_raw_itunes = c("download_activity_date") , staging_raw_spotify = c("download_datetime::date") , staging_raw_spotify_v2 = c("tmstamp::date") , sos_from_raw_view = c("tmstamp::date") , bak_staging_raw_spotify_v2 = c("tmstamp::date") , streams_from_raw_analytics = c("activity_date") , staging_raw_apple_music = "download_date" # c("datestamp") , staging_raw_apple_music_streams = "download_date" # c("datestamp") # rick's tables for applemusic quick comparisons in snowflake , spotifyaggd = c("download_date") , applemusicraw = c("datestamp") ) if (clean_tbl) tbl %<>% tableNameClean(schema=schema) %>% tolower ## Missing schema is not a big deal if the tbl is one of the expected values in dateCol.ll ## Otherwise, it might be a problem, and we thus throw a warning. ## Either way, we set it to NULL if (missing(schema)) { if (tbl %ni% names(dateCol.ll)) warning ("schema is missing in getDateColForTBL -- it will default to NULL. Results may be inaccurate") schema <- NULL } if (snowflake_inuse && !is.null(schema)) schema %<>% tolower if (!is.null(schema) && schema == "youtube_analytics") return("date") if (grepl("^analytics", tbl) && tbl %ni% names(dateCol.ll)) tbl <- "analytics" tbl_is_staging <- grepl("^staging", tolower(tbl)) tbl_is_dim <- grepl("^dim_", tolower(tbl)) ## Fail if cannot find a dateCol AND tbl is not staing if (tbl %ni% names(dateCol.ll)) { if (!tbl_is_staging && !tbl_is_dim) stop ("Do not know the dateCol for tbl='", tbl, "'") if (tbl_is_dim) return("date_created") ## tbl is of type staging. Check the columns, and look for download_date or download_activity_date cols <- try({qShowCols(schema=schema, tbl=tbl, verbose=FALSE, showWarnings=FALSE, connex=connex, snowflake_inuse=snowflake_inuse) %>% tolower}) if (isErr(cols)) cols <- c() cols <- cols[grep("download.*date", cols)] if (length(cols) > 1 && length(grep("activity", cols)) == 1) cols <- cols[grep("activity", cols)] if (length(cols) == 1) return(cols) if (length(cols) > 1) stop ("Do not know the dateCol for tbl='", tbl, "';\nAttempting to guess returned more than one column: ", pasteQand(cols), if (is.null(schema)) "\nHINT: Try including a schema") if (!length(cols)) { if ("weekof" %in% cols) return("weekof") stop ("Do not know the dateCol for tbl='", tbl, "';\nAttempting to guess returned zero columns matching download.*date pattern", if (is.null(schema)) "\nHINT: Try including a schema") } } return(dateCol.ll[[tbl]]) } ## EXAMPLES & TEST CASES if (FALSE) { print( qMaxDate("fact_analytics") ) print( qMaxDate(schema="production", tbl="fact_analytics", dateCols=c("download_activity_date", "processeddaytime")) ) tbls <- c("fact_analytics", "staging_raw_itunes") print( qMaxDate(tbls) ) print( qMaxDate(tbls, dateCols=list(c("download_activity_date", "processeddaytime"), c("download_activity_date", "download_date"))) ) print( qMaxDate(tbls, dateCols=list(c( "processeddaytime"), c("download_activity_date"))) ) } qMaxDate <- function(tbl, dateCols=getDateColForTBL(tbl_like, schema=schema), schema=ifelse(tbl_like %in% c("analytics", "analytics_track_level", "analytics_release_level", "analytics_label_level", "accounting", "mbworkbook", "GPU"), "bi", "production"), byCols=NULL, wh=getSnowflakeWH(), dbname=NULL, tbl_like=gsub("analytics_.+", "analytics", gsub("accounting_.+", "accounting", tbl)), min=FALSE, whereIn=NULL, verbose=TRUE, cluster=NULL, connex=giveMeACon(verbose.dblib=FALSE), snowflake_inuse=getOption("snowflake_inuse", FALSE), refresh.DT.periods=FALSE, simplify=TRUE, period_convert=TRUE, ...) { ## TODO: This is an older function. Incorporate qTableDetails ## TODO: what about processdaytime ## ITERATE OVER MULTIPLE TABLES if (length(tbl) > 1) { ARGS <- collectArgs(except=c("tbl", "schema", "dateCols", "tbl_like")) force (schema); force(dateCols) ret.list <- mapply(qMaxDate, tbl=tbl, dateCols=dateCols, schema=schema, dbname=dbname, MoreArgs=ARGS, SIMPLIFY=FALSE) ## Only simplify if all have length 1. And then check if there are any POSIX if (simplify && alllength1(ret.list)) ret.list <- unlistDateTime(ret.list) return(ret.list) } ## Clean args and create Qry tbl %<>% tableNameClean MaxMinFunc <- if (isTRUE(min)) "min" else "max" Qry <- makeQry(colsToPull=byCols, colsToAgg=dateCols, aggFunc=MaxMinFunc, tbl=tbl, schema=schema, dbname=dbname, whereIn=whereIn, key=NULL, snowflake_inuse=snowflake_inuse) ## Execute query ret <- runQry(Qry, verbose=verbose, cluster=cluster, verbose.key=FALSE, dont.setkey=TRUE, snowflake_inuse=snowflake_inuse, dbname=dbname, wh=wh, connex=connex, ...) setkeyIfNot(ret, colNamesFromVector(c(dateCols, byCols)), verbose=FALSE, organize=TRUE) ret <- ret[orderch(colNamesFromVector(dateCols)[[1]], decreasing=TRUE)] if (period_convert) { try(addDateCols.periodid_(ret, drop=TRUE, showWarnings=FALSE, cluster=cluster, refresh.DT.periods=refresh.DT.periods, snowflake_inuse=snowflake_inuse)) } if (simplify && nrow(ret) == 1) { if (!period_convert) ret <- unlist(ret) else ret <- unlistDateTime(ret) } return(ret) } qMinDate <- function(tbl, ..., min=TRUE, note="This function calls qMaxDate()") { if (!missing(min)) warning ("argument 'min' is ignored in qMinDate(). Use qMaxDate() if necessary.") qMaxDate(tbl=tbl, ..., min=TRUE) } qCountNullsInCol <- function(tbl="fact_analytics", schema="production", colsToCheck=c("date"="download_activity_date"), round=TRUE, verbose.qry=verbose, verbose=TRUE, ...) { case_statements <- colsToCheck %>% colNamesFromVector %>% paste0("_is_null") %>% setNames(obj=sprintf("CASE WHEN %s IS NULL THEN 'YES' ELSE 'NO' END", colsToCheck)) qry <- makeQry(tbl=tbl, schema=schema, colsToPull=case_statements, colsToAgg=c(rows="*"), limit=NULL, key="colsToPull") DT.ret <- runQry(qry, ..., verbose=verbose.qry) ## Add Percentage DT.ret[, percent := percOfTotal(rows, na.rm=FALSE)] ## Add Total try({ if (nrow(DT.ret) > 1) DT.out <- cbind(t(rep("TOTAL", length(colsToCheck))), DT.ret[, lapply(.SD, sumn), .SDcols=c("rows", "percent")]) %>% rbind(DT.ret, ., use.names=FALSE) %>% formnumb(round=round) else DT.out <- copy(DT.ret) %>% formnumb(round=round) }) if (exists("DT.out", inherits=FALSE)) print (DT.out) else print(DT.ret) return(invisible(DT.ret)) } qRowCount <- function(tbl, cluster=NULL, force.numeric=FALSE, whereIn=list(), dateCol=NULL, minDate=NULL, maxDate=NULL, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=NULL, msg_sf=getOption("snowflake_msg_sf", TRUE), snowflake_inuse=getOption("snowflake_inuse", FALSE), verbose=FALSE) { if (!length(tbl)) return (integer()) tbl %<>% tableNameClean(schema=schema) stbl <- schemaPaste(schema=schema, tbl=tbl) if (is.null(dateCol) && (!is.null(minDate) || !is.null(maxDate))) { message ("dateCol was missing from qRowCount() -- will attempt to auto find") dateCol <- getDateColForTBL(tbl=tbl, schema=schema) } ## 'ordering' column is necessary if union of more than one query and the results are expected in the ## order of tbl (as is the case with qShowTables), since otherwise row order can come in any random order whereClause <- makeWhereClause(whereIn=whereIn, dateCol=dateCol, minDate=minDate, maxDate=maxDate, tbl=tbl, schema=schema) whereClause %<>% {if(!is.null(.) && nchar(.)) paste(" WHERE", .) else ""} Qry <- pasteC(sprintf("(SELECT count(*) as \"rows\", %s AS \"ordering\" FROM %s%s)", seq(stbl), stbl, whereClause), C="\nUNION ALL\n") classAppend_(Qry, "query") ret <- runQry(Qry, cluster=cluster, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse, msg_sf=msg_sf, verbose=verbose) if (!nrow(ret)) { warning ("No results returned from qRowCount(tbl = ", if (length(tbl) > 1) capture.output(dput(tbl)) else tbl, ")") return(NA_integer_) } ## column names are not preserved in the union statement. I do not undersand why setnames(ret, c("rows", "ordering")) ret[order(ordering), as.int_or_num(rows, force.numeric=force.numeric)] } qExtractCols <- function(pat.regex, tbl, schema=NULL, dbname=NULL, verbose=FALSE, cluster=NULL, showAllOnNoMatch=TRUE) { if (missing(tbl)) stop ("tbl must be specified.\nHINT: Did you mean to use qShowCols()") cols <- qShowCols(tbl=tbl, schema=schema, dbname=dbname, verbose=verbose, cluster=cluster) ret <- extract(pat.regex, cols) if (!length(ret)) { warning ("No columns of ", schemaPaste(tbl, schema=schema), " matched pattern '", pat.regex, "'") if (showAllOnNoMatch) ret <- cols } return(ret) } qShowCols <- function(tbl, namelike=NULL, sort=TRUE, include.types=FALSE, returnRaw=FALSE, similar=TRUE, simplify=TRUE, cluster=NULL , connex=giveMeACon(verbose.dblib=FALSE) , wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=NULL, snowflake_inuse=getOption("snowflake_inuse", FALSE) , showWarnings=TRUE, verbose=FALSE ) { ## returnRaw: The raw results are a table with tablename, column, type, etc. Return this information ## simplify: if not TRUE, will return a list ## similar : if TRUE, uses 'ILIKE' to match, if FALSE, uses '=' to match ## TODO: Check schema against list of schemas if (length(tbl) > 1) { selfname_(tbl) if (length(schema) > 1 && length(schema) == length(tbl)) return(mapply(qShowCols, tbl, schema, MoreArgs=list(namelike=namelike, include.types=include.types, returnRaw=returnRaw, simplify=simplify), verbose=verbose, SIMPLIFY=simplify, connex=connex, snowflake_inuse=snowflake_inuse, wh=wh, dbname=dbname)) else return(sapply(tbl, qShowCols, namelike=namelike, include.types=include.types, returnRaw=returnRaw, verbose=verbose, simplify=simplify, snowflake_inuse=snowflake_inuse, connex=connex, wh=wh, dbname=dbname, schema=schema)) } tbl %<>% tableNameClean(schema=schema) if (snowflake_inuse) { ARGS <- collectArgs(except="snowflake_inuse") return(do.call(sfShowCols, ARGS)) } if (!is.logical(similar)) stop ("'similar' should be logical -- ie, should 'ILIKE' or '=' be used for matching in SQL\nHINT: did you mean to use the 'namelike' argument?") ## TODO - use schemaSplit() ## If user passed a value like tbl=schema.tbl it needs to be spliced if (is.null(schema) && grepl("\\.", tbl) && length(splat <- strsplit(tbl, "\\.")[[1]])==2) { schema <- splat[[1]] tbl <- splat[[2]] } if (is.null(schema) && tolower(tbl) %in% c("analytics", "accounting", "gpu", "mgmtreport", "mgmtreport_months_reported")) schema <- "bi" tbl.clean <- tableNameClean(tbl, schema=schema) Qry <- sprintf("SELECT * FROM pg_table_def WHERE tablename %s %s", ifelse(similar, "ILIKE", "="), pasteQ(tbl.clean)) if (!is.null(namelike)) Qry <- sprintf("%s AND \"column\" %s %s ", Qry, ifelse(similar, "ILIKE", "in"), pasteQ(namelike)) if (!is.null(schema)) { ## Append to end of schema Qry <- sprintf("%s AND schemaname %s %s ", Qry, ifelse(similar, "ILIKE", "in"), pasteQ(schema)) ## Add search_path criteria, necessary for pg_table_def ## Check the current search path, if the schema set is not in the current path, append the path .sp.current <- runQry("show search_path", cluster=cluster, verbose=FALSE)[, strsplit(search_path, ", ")[[1]]] if (schema %ni% .sp.current) Qry <- sprintf("SET search_path to %s;\n %s", pasteQ(c(.sp.current, schema), w=""), Qry) } Qres <- runQry(Qry, cluster=cluster, connex=connex, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse, verbose=FALSE) ## put the search_path back if (exists(".sp.current") && schema %ni% .sp.current) runQry(sprintf("SET search_path to %s; ", pasteQ(.sp.current, w="")), cluster=cluster, connex=connex, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse, verbose=FALSE) if (returnRaw) return(Qres) if (!length(dim(Qres))) Qres <- data.table(column=NA_character_, type=NA_character_)[0L] ## Check if empty (indicative of some error, or non-exists) if (!is.data.table(Qres) || !ncol(Qres)) { verboseMsg(showWarnings, "No results returned for Qry \n\t", Qry, "\n") return(invisible()) } if (sort) setkey(Qres, column) if (include.types) return(Qres[, list(column, type)]) ## ELSE if (similar) return(Qres[tolower(tablename) %in% tolower(tbl.clean), column]) else return(Qres[tablename %in% tbl.clean, column]) } tableNameClean <- function(tbl, schema=NULL, dontClean=getDBsettings()$dbname %in% c("art_relations", "direct_delivery"), verbose=TRUE) { ## TODO: use schema to distinguish between bi.analytics and production.fact_analytics if (isTRUE(dontClean) || getOption("tbl.dontclean", default=FALSE)) return(tbl) if (is.null(tbl)) return(NULL) if (!is.character(tbl)) stop("the tbl passed to tableNameClean is NOT a character -- it is a ", class(tbl)[[1]], "\n") if (is.null(schema) && tbl == "amazon") stop("amazon!!!") ## Take a copy for notifying the user of any changes .tbl_original <- copy(tbl) tbl[tolower(tbl) %in% c("mb", "mbworkbook", "gl")] <- "mbworkbook" tbl[tolower(tbl) %in% c("anal", "ana", "analytics")] <- "analytics" tbl[tolower(tbl) %in% c("acco", "acc", "accounting")] <- "accounting" ## This section is much older, when our team was not creating many tables ## It should be deprecated, but then again so should this function ## It's purpose was really only interactive, not for programming if (is.null(schema) || schema %in% c("production")) { ## 2015-08-04 - Spotify is now always V2 ## To use the original spotify, name it explicitly tbl[tolower(tbl) %in% c("spotify")] <- "spotify_v2" ## Some specificly allowed values tbl[tbl %in% c("transac", "transactype", "transaction", "transaction_type", "transac_type")] <- "transactiontype" tbl[tbl == "fanal"] <- "fact_analytics" tbl[tbl == "facc"] <- "fact_sales" tbl[tolower(tbl) %in% c("spotify2", "spotifyv2", "spotify_v2", "spot2") ] <- "spotify_v2" tbl[tolower(tbl) %in% c("apple_music", "apple", "music", "applemusic") ] <- "apple_music" dim.tbls <- c("day", "transactiontype", "store", "release", "artist", "genre", "label", "track", "isrc" , "product", "country", "owner", "period", "catalog", "imprint") fact.tbls <- c("sales") # "analytics" can no longer be here staging.tbls <- c("itunes", "spotify", "rdio", "deezer", "amazon", "spotify_v2", "apple_music") tbl.lower <- tolower(tbl) dims <- tbl.lower %in% dim.tbls facts <- tbl.lower %in% fact.tbls staging <- tbl.lower %in% staging.tbls tbl[dims] <- paste0("dim_", tbl.lower[dims]) tbl[facts] <- paste0("fact_", tbl.lower[facts]) tbl[staging] <- paste0("staging_raw_", tbl.lower[staging]) } ## Notify the users of which table names have been changed if (verbose && any(wh.changed <- tbl != .tbl_original)) { catnn( sprintf("\tCleaning table name from '%s' to '%s'", .tbl_original[wh.changed], tbl[wh.changed]) , sprintf("\tschema is %s", ifelseNULL(schema, "NULL", sprintf('%s', schema))) ) } return(tbl) } headDB <- function(tbl, n=limit, connex=giveMeACon(verbose=inVerboseMode.dbcon() || verbose, snowflake_inuse=snowflake_inuse), justHeader=FALSE, whereIn=NULL, dateCol, minDate=NULL, maxDate=NULL, colsUsing="*", schema=NULL, dbname=NULL, limit=5L, dont.drop.anything=FALSE, check.table.perms=TRUE, cluster=NULL, msg_sf=getOption("snowflake_msg_sf", TRUE), snowflake_inuse=getOption("snowflake_inuse"), verbose.qry=FALSE, verbose=TRUE) { ## Allow for an unquoted argument if (!exists(as.character(substitute(tbl))) || !is.character(tbl)) tbl <- as.character(substitute(tbl)) ## If n is negative, returns all rows. Just a basic query. if (length(n) != 1) stop ("n must be a single number, not a vector, nor list, etc") if (length(tbl) > 1) { aa <- as.list(environment()) aa <- aa[! names(aa) %in% c("tbl")] return(lapply(tbl, function(t) do.call(headDB, c(tbl=t, aa)))) } whereClause <- makeWhereClause(whereIn=whereIn, dateCol=dateCol, minDate=minDate, maxDate=maxDate) # if (missing(connex) || isConExpired(connex)) # connex <- giveMeACon() ## if (!missing(connex)) ## warning("connex no longer used") tbl <- tableNameClean(tbl, schema=schema) if (is.null(schema) && !grepl("\\.", tbl)) { if (tolower(tbl) %in% c("analytics", "accounting")) schema <- "bi" else if (snowflake_inuse) schema <- "production" } stbl <- schemaPaste(schema=schema, tbl=tbl) query <- paste("SELECT", paste(colsUsing, collapse=", "), "\nFROM ", stbl) # Append the WHERE clause if (length(whereClause) && nchar(whereClause)) { ## Error-check: Confirm that the user did NOT include the keyword WHERE if (grepl("^WHERE ", whereClause, ignore.case=TRUE)) stop("`whereClause` should NOT start with the word 'WHERE'") query <- paste(query, "\nWHERE", whereClause) } ## append the limit, if n is non-negative if (n >= 0L) query <- paste(query, "\nLIMIT", ifelse(n==0, 5, n)) qres <- runQry(query, connex=connex, dont.drop.anything=dont.drop.anything, check.table.perms=check.table.perms, cluster=cluster, msg_sf=msg_sf, snowflake_inuse=snowflake_inuse, verbose=verbose.qry) ## Convert IDcols setIDCols(qres, verbose=FALSE) verboseMsg(verbose, " ~~~~~~ \t", dbschematbl(dbname=dbname, schema=schema, tbl=tbl), "\t ~~~~~~ ", time=FALSE) if (justHeader || n==0) return(names(qres)) return(qres) } nrowDB <- function(tbl, connex=giveMeACon(verbose.dblib=FALSE) , whereClause=NULL) { # Where clause should be a string that follows "WHERE" in a SQL Query. # eg, whereClause="id=7" # eg: nrowDB(tbl) # nrowDB(tbl, whereClause="dateCreated > '2013-08-01'") # nrowDB(tbl, whereClause=paste0("dateCreated > '", minDate, "'")) # if (missing(connex) || isConExpired(connex)) # connex <- giveMeACon() query <- paste("select count(*) from", tbl) if (length(whereClause)) { ## Error check for user input if (grepl("^WHERE ", whereClause, ignore.case=TRUE)) stop("`whereClause` should NOT start with the word 'WHERE'") query <- paste(query, "WHERE", whereClause) } return(dbGetQuery(connex, query)) } funcOfCols <- function(funcName, cols) { ## Wrapper function to create SQL Queries ## eg: funcOfCols(c("max", "min"), cols) if (length(funcName) > 1) return(paste( sapply(funcName, funcOfCols, cols=cols), collapse=", ") ) paste0(funcName, "(", cols, ") AS ", funcName, "_", cols, collapse=", ") } minDB <- function(tbl, cols, connex=giveMeACon(verbose.dblib=FALSE)) { # if (missing(connex) || isConExpired(connex)) # connex <- giveMeACon() query <- paste("select ", funcOfCols("min", cols), " from", tbl) return(dbGetQuery(connex, query)) } maxDB <- function(tbl, cols, connex=giveMeACon(verbose.dblib=FALSE)) { # if (missing(connex) || isConExpired(connex)) # connex <- giveMeACon() query <- paste("select ", funcOfCols("max", cols), " from", tbl) return(dbGetQuery(connex, query)) } ## EG: # minDB(tbl, cols) # minDate <- "2013-08-15" # nrowDB(tbl, whereClause=paste0("dateCreated > '", minDate, "'")) getKeyVals <- function(DT) { keyCols <- key(DT) keyVals <- list() for (col in keyCols) { # for factor cols, grab the levels if (is.factor (DT[[col]]) ) keyVals[[col]] <- levels(DT[[col]]) # otherwise grab the unique values else keyVals[[col]] <- unique(DT[[col]]) } return(keyVals) } not <- function(vals, col, keyVals) { #' used for DT key indexing. #' col can be a character name of a column or an index reference to the key order. #' example kv <- getKeyVals(DT) #' DT[.(c("a", "bb", "ac"), not(c("xx", "yy"), 2, kv))] setdiff(keyVals[[col]], vals) } extractTablenameFromQry <- function(qry, debug=FALSE) { ## This is a sloppy regex attempt to extract the table name from a query. ## It is prone to errors. Do not depened heavily on this ## For some reason, my regex does not work if there is a line break directly after the table name. qry <- gsub("\\s+\\s*", " ", qry) pat <- "\\s*(FROM|JOIN)\\s+(.+?\\s)" matches <- regexec(pat, qry, ignore.case=TRUE) extracted <- regmatches(qry, matches) browser(expr=isTRUE(debug), text="in extractTablenameFromQry().\nCheck matches & extracted.") ret <- unique(as.vector(sapply(extracted, gsub, pat="^\\s*(FROM|JOIN)\\s*|\\s*$", repl=""))) ret[ret != ""] } sqlFileToQry <- function(file, clearComments=clear, clearDoubleBlankLines=clearComments, clear=FALSE, replaceRStyleComments=TRUE, verbose=TRUE) { ## clearRStyleComments :: I sometimes, incorrectly, use hashes to add comments in SQL files. ## If the sql file gets read in the middle of a long script, the error recovery might be tedious. Instead, I prefer to catch it in here. nms <- if (is.null(names(file))) file else names(file) if (length(file) > 1) { force(clear); force(clearComments); force(clearDoubleBlankLines) ret <- sapply(file, sqlFileToQry, clearComments=clearComments, clearDoubleBlankLines=clearDoubleBlankLines, replaceRStyleComments=replaceRStyleComments, verbose=verbose) setQry(ret) setattr(ret, "names", nms) return(ret) } if (!length(file)) stop ("'file' value sent to sqlFileToQry() has no length") if (any(is.na(file))) stop ("There are NAs in the 'file' value sent to sqlFileToQry()") if (any(!file.exists(file))) stop("The following file(s) do(es) not exist: ", file[!file.exists(file)]) raw <- readLines(file) if (replaceRStyleComments) { pat.rstyle <- "( *##|^ *#)" pat.commented <- "--.*#" if (any(wh.rstyle <- grepl(pat.rstyle, raw) & !grepl(pat.commented, raw))) warning(pasteC(c("R-style comments found in file ", pasteQ(path.unexpand(file), w=""), "These will be converted to SQL comments. Below are the first few offending lines\n", sprintf("%4i : %s", head(which(wh.rstyle), 6), substr(head(raw[wh.rstyle], 6), 1, 65) )), C="\n ")) # warning("There appears to be R-style comments in file '", path.unexpand(file), "'\nThese will be converted to SQL comments. Example lines (head 6) :\n ", pasteC(sprintf("%4i : %s", head(which(wh.rstyle), 6), substr(head(raw[wh.rstyle], 6), 1, 65) ), C="\n ")) raw[wh.rstyle] <- gsub(pat.rstyle, "-- \\1", raw[wh.rstyle]) } if (clearComments) { ## if a line is nothing but comments, remove the whole line raw <- raw[!grepl("^\\s*--", raw)] ## Otherwise, remove just the comment raw <- gsub("\\s*--.*", "",raw) } if (clearDoubleBlankLines && length(blanklines <- which(trim(raw) == ""))) { toDrop <- which(diffNA(blanklines) == 1) if (length(toDrop)) raw <- raw[-blanklines[toDrop]] } ret <- pasteC(raw, C="\n") setQry(ret) setattr(ret, "names", nms) return(ret) } print.query <- function(x, linenumbers=FALSE, max.width=getOption("width", 80) - 5L, max.lines=45L, ...) { if (!length(x)) { warning ("\n", capture.output(substitute(x)), " sent to print.query() has no length") return(invisible(NULL)) } if (is.list(x)) { if (length(x) == 1 && inherits(x[[1]], "query")) return(print(x[[1]])) else return(print(unclass(x))) } ret <- sapply(x, verboseQry, max.width=max.width, max.lines=max.lines, ..., USE.NAMES=!is.null(names(x))) ## If there is more than one query, or if x is named, append a line with the names and a long hr bar nms <- names(ret) if (length(ret) > 1 || !is.null(nms) ) { if (is.null(nms)) nms <- sprintf("[[%i]]", seq(ret)) mc <- mnchar(nms) frmt <- paste0("\n~%s %-", mc+2, "s ~%1$s\n") hr <- pasteR("-~", max(1, (max.width-mc-2)/4) ) header <- sprintf(frmt, hr, nms) cat(paste0(header, sub("(\\s*^(\\s*\\n)*|(\\s*\\n)*\\s*$)", "", ret), "\n")) } else if (linenumbers) { strsplit(ret, "\\n")[[1]] %>% {capture.output(cbind(.))} %>% gsub(",\\] \"", "] ", .) %>% gsub("\"\\s*$", "", .) %>% catn(sep="\n") } else { cat(ret, "\n") } return(invisible(ret)) } sql_count_is_like_integer <- function(colName, nms=colNamesFromVector(colName)) { ## This function needs a better name if (!length(colName)) return(NULL) setNames(sprintf("sum(CASE WHEN trunc(%1$s) != %1$s THEN 1 ELSE 0 END)", colName), nm=nms) } sql_yr_wk <- function(dateCol, name="yr_wk") { ret <- sprintf("CASE WHEN (MONTH(%1$s) = 12 AND WEEK(%1$s) = 1) THEN YEAR(%1$s) + 1 ELSE YEAR(%1$s) END + WEEK(%1$s)/100", dateCol) return(setNames(ret, name)) } sql1stOfMonth <- function(dateCol, as=NULL, tbl=NULL, schema=NULL, snowflake_inuse=getOption("snowflake_inuse", FALSE)) { ## use schemaPaste to append the tbl name to the column name stbl <- schemaPaste(schema=schema, tbl=tbl) col <- schemaPaste(schema=stbl, tbl=dateCol) qryStructure <- ifelse(snowflake_inuse, "date_trunc('month', %s)::date", "CAST(TO_CHAR (%s, 'YYYY-MM-01') AS date)") if (!is.null(as)) qryStructure <- paste0(qryStructure, " AS \"", as, "\"") return(sprintf(qryStructure, col)) } ## The package RPostgreSQL breaks when using the rstats-db version of DBI ## This patch fixes it try({ if (exists("dbIsValid")) setMethod("dbIsValid", "PostgreSQLConnection", function(dbObj, ...) { isValid <- tryCatch({dbGetInfo(dbObj)}, error = function(e) NULL) !is.null(isValid) }) }) .DBlib <- function(pkgs=names(getOption("db.defaulthosts")), unload=FALSE, envir_con=getOption("con.environ", default=parent.frame(pos)), verbose=TRUE, use_rstats_db_postgres_instead=getOption("use_rstats_db_postgres", default=FALSE)) { ## remembering the propper capitalization is a pain in the ass. ## This is a wrapper function that will load the required package # gc() if (!length(pkgs)) { warning ("pkgs has no length. Did not set package and Returning NULL") return(NULL) } loaded_packages <- search() %>% setdiff(c(".GlobalEnv", "Autoloads")) %>% gsub("^package:", "", .) if (missing(pkgs)) pkgs <- pkgs[[1]] packages <- c("RPostgreSQL", "RMySQL", "snowflake", "RODBC") matches <- grep(tolower(pkgs), tolower(packages), ignore.case=TRUE) if (!length(matches) || is.na(matches)) matches <- pmatch(tolower(pkgs), tolower(packages)) browser(expr=inDebugMode("DBlib"), text="in .DBlib() from DB_Utils.r.\nAbout to unload the packages. Check matches and packages[-matches].") if (length(matches) > 1 || is.na(matches)) { stop("Matched more than one or no matches at all. \n Options are: \n\n\ ", paste(packages, collapse=" \t ")) } ## USE RPostgres if (use_rstats_db_postgres_instead) packages[packages == "RPostgreSQL"] <- "RPostgres" pkgToLoad <- packages[matches] ## DELETE ANY EXISITING CONNECTIONS if (pkgToLoad != "snowflake") { tmp_was_loaded <- .is_RODBC_loaded() || getOption("snowflake_inuse", default=FALSE) sf_envir <- .get_sfEnvir() if (is.environment(sf_envir)) rm(list=ls(envir=sf_envir, all=TRUE), envir=sf_envir) if (tmp_was_loaded) gcQuietly() } if (pkgToLoad == "snowflake") pkgToLoad <- "RODBC" pkgsToRemove <- packages[packages != pkgToLoad] %>% intersect(loaded_packages) ## Add in DBI if different packge to load # OLD: if (length(pkgsToRemove) && !(pkgs %in% pkgsToRemove)) # OLD: packages <- c(packages, "DBI") # new 2015-08-15: # if (length(pkgsToRemove) && !(pkgToLoad %in% pkgsToRemove)) # pkgsToRemove <- c(pkgsToRemove, "DBI") %>% unique if (pkgToLoad == "RODBC" && "DBI" %in% loaded_packages) pkgsToRemove <- c(pkgsToRemove, "DBI") %>% unique if (all(pkgToLoad %in% loaded_packages) && !any(pkgsToRemove %in% loaded_packages)) { verboseMsg(verbose, "No need to change R Packages in .DBlib") return(invisible(TRUE)) } ## Otherwise, start clearing if (!"RODBC" %in% pkgToLoad) try(RODBC::odbcCloseAll(), silent=TRUE) try(dbDisconnectAll(), silent=TRUE) suppressWarnings(rm(sfCon, .sfcon, con, connex, envir=envir_con)) for (pkg.rm in pkgsToRemove) if (any(grepl(pkg.rm, search()))) { verboseMsg(verbose, "Detaching package", pkg.rm, time=FALSE) detach(paste0("package:", pkg.rm), character.only=TRUE, force=TRUE, unload=unload) # , unload=TRUE suppressWarnings(unloadNamespace(pkg.rm)) } gc() ## This is redundant # if (!any(grepl(pkgToLoad, search()))) # verboseMsg(verbose, "Loading Package", pkgToLoad, time=FALSE) if ("RPostgreSQL" %in% pkgToLoad) { source("~rsaporta/git/misc/rscripts/utils/DB_Utils/dbIsValid for RPostgreSQL.r") } require(pkgToLoad, character.only=TRUE) } schemaSplit <- function(stbl, sep=".", nms=c("schema", "table"), schema.default="production") { unknown <- NULL sep.escaped <- escapeRegEx(sep) splat <- strsplit(stbl, sep.escaped)[[1]] l.nms <- length(nms) l.splat <- length(splat) # browser() if (l.splat > l.nms) { diff <- l.splat - l.nms warning(sprintf("Expected %i splits, but '%s' yielded %i. Using only the last %1$i %s", l.nms, stbl, l.splat, plrl("parts", l.nms))) unknown <- pasteC(splat[1:diff], C=sep) splat <- tail(splat, l.nms) l.splat <- length(splat) } if (l.nms > l.splat) { if (l.nms != 2 && l.nms - l.splat != 1 ) stop ("Not enough values in splat to match up to nms") if (l.nms != "schema") stop ("Not enough values in splat to match up to nms and 'schema' is not first in nms, so cannot add default") splat <- c(schema.default, splat) } names(splat) <- nms[seq(length(splat))] # dont use l.splat here in case changed return(splat) } schemaPaste <- function(schema, tbl, sep=".") { # This is a wrapper function to paste0(schema, sep, tbl) # that ensures that there is exactly one sep between schema and tbl # and also allows for NULL or "" values of schema to be intermixed # Does not allow for multiple values of tbl l.t <- length(tbl) l.s <- length(schema) if (!l.s) return(tbl) if (!l.t) { warning ("\n No value for `tbl` passed. Returning tbl.") return(tbl) } if (l.t != l.s) { if (!(l.s == 1 || l.t == 1)) stop (sprintf("schema (length %i) and tbl (length %i) have differing lengths", l.s, l.t)) else schema <- rep(schema, l.t) } # if (length(tbl) > 1) # stop ("\n tbl must be exactly length 1. Use `sapply` or `mapply`") # if schema is exactly sep, we keep those blanks <- nchar(schema) == 0 # ensure each schema has exactly one sep by remove any seps then pasting them back sep.pat <- gsub("\\^", "", glob2rx(sep)) schema <- gsub(sep.pat, "", schema) schema <- paste0(schema, sep) schema[blanks] <- "" has_schema <- mapply(grepl, pattern=escapeRegEx(schema), x=tbl) ## Check that tbl does not alredy have the schema prepended tbl[!has_schema] <- paste0(schema[!has_schema], tbl[!has_schema]) return(tbl) } # --------------- # qCurrUser <- function() { ## Returns the current user runQry("SELECT current_user;", verbose=FALSE, to.dt=FALSE)[[1]] } qMyQs <- function(user, currentUser=isTRUE(missing(user))) { ## Show what queries are running under the current user qWhatsRunning(currentUser=TRUE) } qShowQueries <- function(user, currentUser=isTRUE(missing(user)), rdsdb_ignore=TRUE, showWarnings=TRUE) { colsToPull <- c("pid", "starttime", "duration") colsToTrim <- c("user_name", "db_name", qry="substring(query,1,30)", "status") tbl <- "stv_recents" schema <- NULL whereIn <- if (rdsdb_ignore) ("NOT (user_name = 'rdsdb')") ret <- runQry(makeQry(tbl=tbl, schema=schema, dbname=dbname, colsToPull=colsToPull, colsToAgg=colsToTrim, aggFunc="trim", whereIn=whereIn, limit=NULL), verbose=FALSE) if (!nrow(ret)) { verboseMsg(showWarnings, "No queries were found in stv_recents") return(invisible(NULL)) } if (!is.logical(currentUser)) warning ("'currentUser' should be TRUE or FALSE. \nHINT: Did you mean to set the 'user' arg? ") if (isTRUE(currentUser)) { if (!missing(user)) verboseMsg(showWarnings, "'currentUser' is TRUE and 'user' has been manually set. 'user' value will be ignored and replaced with result of qCurrUser") user <- qCurrUser() } if (!any(user %in% ret[["user_name"]]) ) verboseMsg(showWarnings, "No queries found for user name(s) ", pasteQ(user, w=""), " -- returning results for all users") else ret <- ret[user_name %in% user] ## Clean up date/time if ("starttime" %in% names(ret)) { ret[, startdate := as.Date(starttime, tz = getOption("default.tz", "America/New_York"))] ret[, starttime := readableTime(starttime, TRUE)] ret <- ret[order.readableTime(starttime)][order(startdate)] } setcolorderpt(ret, c("pid", "status", "startdate", "poop", "starttime"), showWarnings=FALSE) if ("status" %in% names(ret)) { ret[, status := setFactorOrder(status, c("Running", "Done"), showWarnings=FALSE)] setkey(ret, status, startdate) } ## Clean duration if ("duration" %in% names(ret)) try(ret[, duration := fwSecs(duration, units="microseconds", fraction=TRUE)]) ## sort by status ret return(ret) } qWhatsRunning <- function(user, currentUser=isTRUE(missing(user)), rdsdb_ignore=TRUE, showWarnings=TRUE) { ret <- qShowQueries(user=user, currentUser=currentUser, rdsdb_ignore=rdsdb_ignore, showWarnings=FALSE) if (is.null(ret) || !any(ret$status == "Running")) { verboseMsg(showWarnings, "No queries were found") return (NULL) } ## ELSE return(ret[status=="Running"]) } qWhatsRunning_OLD <- function(user=NULL, currentUser=mine, mine=FALSE) { ## This queries the pg_stat_activity table instead of stv_blocklist ret <- runQry("select * from pg_stat_activity", verbose=FALSE) ## Error check for input if (!is.logical(currentUser) && !is.null(currentUser)) { stop (paste("The argument currentUser should be T or F." , if (is.character(currentUser)) "\nIf you want to pass a username, use argument `user=____`\n")) } if (currentUser) user <- qCurrUser() if (!is.null(user) && !all(is.na(user))) { ret <- ret[ret$usename %in% user, ] } return(ret) } qKill <- function(id, force=FALSE, verbose=TRUE, user=NULL, currentUser=mine, mine=FALSE, cluster=NULL) { # mine : synonym for currentUser # currentUser : Logical value, indicating if we should restrict search to current user whoIsCalling <- qCurrUser() allQs <- qWhatsRunning(user=user, currentUser=currentUser) ## ALLOW MANUAL SELECTION OF WHICH id TO KILL: if (missing(id) || is.null(id)) { print(allQs) id <- readline("Which to kill? > ") # Manually quit with eXit/qUit/Cancel if (substr(tolower(id), 1,1) %in% c("x", "q", "c", "e")) { cat("\nNo action taken\n") return(invisible(NULL)) } id <- strsplit(id, ",| |;")[[1]] id <- suppressWarnings(unique(removeNA(as.numeric(id)))) invalid <- FALSE # No id's remain after strsplit and removals if (!length(id)) invalid <- TRUE # id can be a procpid or a row number. (It might accidentally be a usesysid or datid) # if it's a procpid, then all good, move on. Otherwise, try to match or throw an erro if (!invalid && !any(id %in% allQs[["procpid"]])) { if (all(id %in% seq(nrow(allQs)))) id <- allQs[id, procpid] else invalid <- TRUE } if (invalid) { cat("\nYou must enter a valid procpid or a row number corresponding to a procpid\nNo action taken\n") return(invisible(NULL)) } } # // end manual selection of id toKill <- allQs[allQs$procpid %in% id, ] cat("\n\n") cat("You are currently accessing the DB as user: \"", whoIsCalling, "\" \n", sep="") cat("The following processes will be killed\n\n") print(toKill) cat("\n\n") if (any(!(toKill$usename == whoIsCalling))) { cat("\n ", pasteR("!", 65), "\n\t\t\t\tWARNING:\n\tYou are killing processes for which you are not the user\n ", pasteR("!", 65), "\n\n", sep="") } areYouSure <- readline("Are you sure you want to proceed? (type \'yes\') > ") # give two tries if (areYouSure != "yes" && tolower(substr(areYouSure,1,1))=="y") { areYouSure <- readline("If you'd like to proceed, you must type exactly \'yes\' > ") } if (areYouSure != "yes") { cat("\nNo action taken\n") return(invisible(NULL)) } # else cat("Killing procpids ", paste(id, collapse=", "), ". . . \n" ) ## allow a second to frantically ctrl+esc even after the confirmation Sys.sleep(1.3) Q <- paste0("select pg_cancel_backend(", paste(id, collapse=", "), ")") # runQry(Q) cat("\t\tRICK'S NOTE:\nTEMPORARILLY NOT RUNNING THE QUERY UNTIL I AM CERTAIN THERE ARE NO BUGS.\nRUN THIS MANUALLY: \n\n ", paste0("runQry('", Q ,"', cluster=", if(is.null(cluster)) "NULL" else cluster, ")\n\n") ) return(invisible(NULL)) } qCanRead <- function(tbl, user=NULL, connex=giveMeACon(verbose.dblib=FALSE), checkTableExists=FALSE, showWarnings=TRUE, cluster=NULL, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=NULL, snowflake_inuse=getOption("snowflake_inuse", FALSE), verbose=FALSE) { ## checkTableExists : if TRUE will first see if the table name is an actual table. ## qShowTables calls qCanRead . if checkTableExists is TRUE, then qCanRead also calls qShowTables if (snowflake_inuse) { message("qCanRead() was called when snowflake_inuse. Not sure how to check for snowflake. Returning TRUE") rep(TRUE, length(tbl)) } ## backup for debugging tbl.bak <- tbl if (any(tbl=="")) { warning ("Some values of tbl sent to qCanRead() are blank. These will not be checked") tbl <- tbl[tbl!=""] } if (!length(tbl)) stop ("tbl has no length") ## For verbose messaging LL <- length(tbl) if (!LL || !is.character(tbl)) stop("Invalid tbl argument") # we will be re-using the connection. Force it early. force(connex) tbl <- tableNameClean(tbl, schema=schema) if (!is.null(schema)){ if (length(schema) == length(tbl)) tbl <- mapply(schemaPaste, schema=schema, tbl=tbl.bak) else tbl <- sapply(schema, schemaPaste, tbl=tbl) } if (length(tbl) > 250) { warning ("Two many tables to check (", length(tbl), "). Sheepishly exiting qCanRead() and returning NA") return(NA) } ## Check that Table Exists if (checkTableExists) { allTables <- qShowTables(connex=connex, includePerm=FALSE)[, unique(c(tbl, paste0(schema, ".", tbl)))] if (any(wh <- {!(tbl %in% allTables)})) { if (all(wh)) stop ("Table ", if(!is.null(schema)) "& schema combination ", "not found in the list of tables.\nWe searched for:\n ", paste(tbl, collapse=" "),"\n\n") if (showWarnings) warning("The following tables do not exist in the database: \n ", paste(tbl[wh], collapse=" "), "\n" ) tbl <- tbl[!wh] } } if (length(user) > 1) return(sapply(user, function(u) qCanRead(tbl=tbl, user=u, schema=NULL, connex=connex, verbose=verbose))) verboseMsg(verbose, "Checking read perrmissions for", if (is.null(user)) "default user" else user, "on", if (LL>5) paste(LL, "tables") else paste_l(tbl, 4, ","), time=FALSE) Q <- paste("SELECT has_table_privilege(" , {if (!is.null(user)) paste0("'", user, "', ")} , paste0("'", tbl, "', ") , "'select'" , ");") browser(expr=inDebugMode(or='qCanRead'), text="In qCanRead(). Right before running qry. Run, then check for empty results") ret <- sapply(Q, runQry, verbose=FALSE, to.dt=FALSE, connex=connex, check.table.perms=FALSE, emailWhenDone=FALSE, dont.drop.anything=TRUE, cluster=cluster) ret <- setNames(unlist(ret, use.names=FALSE), tbl) ## Verbose messagig verboseMsg(verbose, "User has permission on", if (sum(ret)==LL) "all" else paste(sum(ret), "of the"), LL, "tables", time=FALSE) return(ret) } tableToView <- function(string, ignore.case=TRUE) { gsub("table", "view", gsub("Table", "View", gsub("TABLE", "VIEW", string)), ignore.case=ignore.case) } qShowViews <- function(where=NULL, schema="bi", what="views", cluster=NULL, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), snowflake_inuse=getOption("snowflake_inuse", FALSE), verbose=FALSE, ...) { if (!missing(what)) warning ("argument 'what' is ignored in qShowViews(); use qShowTables() to override", call.=FALSE) qShowTables(where=where, schema=schema, what="views", cluster=cluster, verbose=verbose, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse, ...) } qShowTables <- function(where=NULL, schema=c("production", "bi", "ds_scratch"), namelike=NULL , rowCount=("bi" %in% schema) && (!justnames) , showDetails=FALSE , justnames=FALSE , filterByPerm=FALSE , includePerm=FALSE , connex=giveMeACon(verbose=verbose, snowflake_inuse=snowflake_inuse) , all.pfm=getOption("db.all.pfm") , what=c("tables", "views") , nomatch=c("all", "none") , showWarnings=TRUE , order=c("schema", "cluster", "name") , cluster=NULL , details="auto" , wh=getSnowflakeWH() , dbname=getSnowflakeDB() , snowflake_inuse=getOption("snowflake_inuse", FALSE) , msg_sf = getOption("snowflake_msg_sf", TRUE) , verbose=FALSE ) { ## TODO: do we need the connex argument here? ## TODO: Clean up these variable names. do.ts, CamelCase, camelCase, lowercase ## SNOWFLAKE -- Collect args and ship to comparable function if (snowflake_inuse) { verboseMsg(msg_sf, "calling sfShowTables() from qShowTables()", func="message", time=FALSE) ARGS <- collectArgs(except=c("snowflake_inuse")) return(do.call(sfShowTables, ARGS)) } if (!missing(details)) warning("details in qShowViews() and qShowTables() is just for snowflake. It will be ignored", .call=FALSE) if (getDBdriver() != "PostgreSQL") stop("qShowTables() only works for PostgreSQL -- Try using \n runQry(\"SHOW TABLES\")") if (!is.null(where) && all(where %in% c("production", "bi", "ds_scratch"))) { schema <- where where <- NULL } what <- match.arg(what) TtoV <- function(string) { if (what=="views") tableToView(string) else string } ## Schema should be a character or NULL. In previous versions of this function, there was a 'use.prod' boolean in its place if (is.logical(schema)) { stop("The parameter 'schema' should be a string name of the schema.\n Do Not use logical values for 'schema'") } ## Reminder to user on how to not-specify a schema if (!is.null(schema) && schema=="" && showWarnings) warning ("'schema' is blank. Did you mean to set it to NULL") # if `where` is a single word, treat it as `namelike` argument if (!is.null(where) && is.character(where) && !grepl(" ", where)) { namelike <- where where <- NULL } if (!is.null(namelike)) { if (!is.null(where)) stop("One of `where` and `namelike` must be NULL") where <- pasteC(TtoV("tablename ILIKE"), pasteQ(namelike, q="'%", C=NULL), "\n", C=" OR ") } if (!is.null(schema)) { where.schema <- pasteC(sprintf("schemaname ILIKE '%s'", schema), C=" OR ") where <- paste0("(", where.schema, ")", if (!is.null(where)) sprintf(" AND (%s)", where)) } Q.base <- sprintf("SELECT * FROM pg_catalog.pg_%s", what) if (!is.null(where)) Q <- paste(Q.base, "WHERE", gsub("^where", "", where)) else Q <- Q.base browser(expr=inDebugMode("qShowTables"), text="in qShowTables() right before execute") ## -==== ====- ## ## -==== EXECUTE ====- ## ## -==== ====- ## ret <- runQry(Q, connex=connex, cluster=cluster, verbose=verbose, all.pfm=all.pfm) if (is.null(ret) || !nrow(ret)) { if (showWarnings) warning(" No rows returned with matching table name. Query ran was:\n", verboseQry(Q, all=TRUE)) if (justnames) return(character(0)) return(ret) } setnames(ret, gsub("(table|view)", "", names(ret))) ## -==== ====- ## ## Do not check row counts on more than 9 tables, unless explicitly set by user if (missing(rowCount) && length(ret[["name"]]) > 9) rowCount <- FALSE if (isTRUE(rowCount)) try(ret[, rows := qRowCount(tbl=name, schema=schemaname, cluster=cluster)], silent=TRUE) # ret[, qRowCount(tbl=name, schema=schemaname), by=name] # If no rows returned, and there was a where clause, try without it if (!nrow(ret) && (Q.base != Q)) { message("ENTERING BROWSER... CHECK THE TEXT") browser(text=".. THIS MIGHT BE INCOMPLETE... \n\n What is the difference between `found` & `ret` ?\n") ret <- runQry(Q.base, verbose=verbose, cluster=cluster, all.pfm=all.pfm) found <- ret[["name"]] %cont% namelike ret <- ret[name %cont% namelike] } ## Add column for user permission. But only if flagged. Beware that qCanRead calls back to this function, so might hit infinite loop. if (includePerm) { if (filterByPerm || !justnames) ## dont bother running this, if returning just names and not filtering by perms ret[, canread := qCanRead(name, checkTableExists=FALSE, schema=schemaname, cluster=cluster, verbose=verbose)] ## Filter to only those allowed if (filterByPerm) ret <- ret[(canread)] } ## Drop detail columns if (!showDetails) { ## include schemaname only if schema is null colsToInclude <- c("name", "owner", "rows") if ("canread" %in% names(ret)) colsToInclude <- c(colsToInclude, "canread") if (is.null(schema) || length(schema) > 1) colsToInclude <- c("schemaname", colsToInclude) colsToInclude <- TtoV(colsToInclude) ## Drop all others ret[, setdiff(names(ret), colsToInclude) := NULL, with=FALSE] ## Reorder the columns, namely putting schemaname to end setcolorderpt(ret, colsToInclude, showWarnings=FALSE) } ret[, cluster := valueIfNull(cluster, getCluster())] ## SET THE ORDER ## OLD: order based on name. # kCols <- TtoV("name") # kCols <- TtoV(order) ## NEW: order based on order argument (first extracting matches from names(ret)) order <- removeNA(extractMatch(order, names(ret), use.names=FALSE, showWarnings=FALSE)) if (!length(order) && "name" %in% names(ret)) order <- "name" suppressWarnings(setkeyv(ret, TtoV(order))) ## sup if (justnames) { if (is.null(schema)) return(setNames(nm=ret[["name"]])) # self-name, so that they can then be lapply'd over return(setNames(nm=ret[, paste0(schema, ".", name)])) # self-name, so that they can then be lapply'd over } print(formnumb(ret, round=FALSE)) return(invisible(ret)) } qFindView <- function(..., showWarnings=TRUE, use.view=TRUE) { if (!missing(use.view)) warning("Arugment 'use.view' is ignored in qFindView()") ret <- qFindTable(..., showWarnings=showWarnings, use.view=TRUE) if (nrow(ret) && "definition" %in% names(ret)) { ret[, def_100chars := substr(definition, 1, 100)] # ret[, definition := {browser(); setNames(obj=lapply(definition, as.list), nm=schemaPaste(schema=schema, tbl=viewname))}] ret[, definition := list(mapply(function(n, d) setNames(obj=list(d), nm=n), d=definition, n=schemaPaste(schema=schema, tbl=viewname), USE.NAMES=FALSE, SIMPLIFY=FALSE))] } return(ret) } qFindTable <- function(pattern, ignoreCase=ignore.case, ignore.case=TRUE, exact=!any(grepl("\\%", pattern)), use.view=FALSE , all.pfm=getOption("db.all.pfm") , cluster=NULL , wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=NULL, snowflake_inuse=getOption("snowflake_inuse", FALSE) , msg_sf = getOption("snowflake_msg_sf", TRUE) , showWarnings=TRUE , verbose=FALSE ) { # EG: qFindTable("staging") or qFindTable("zip") # public.too : if NOT TRUE, then tables from schema public. are not included ## SNOWFLAKE -- Collect args and ship to comparable function if (snowflake_inuse) { verboseMsg(msg_sf, "calling sfShowTables() from qShowTables()", func="message", time=FALSE) ARGS <- collectArgs(except=c("snowflake_inuse")) ARGS$namelike <- ARGS$pattern; ARGS$pattern <- NULL ARGS$nomatch <- "none" ARGS$justnames <- TRUE return(as.data.table(do.call(sfShowTables, ARGS))) } if (is.null(schema) && grepl("\\.", pattern)) { stbl.splat <- schemaSplit(pattern) pattern <- stbl.splat[["table"]] schema <- stbl.splat[["schema"]] } if (!is.logical(use.view)) stop("Arugment 'use.view' in qFindTable() should be logical") if (ignoreCase) { pattern <- regExUpperLower(pattern) schema <- regExUpperLower(schema) } if (!exact) { if (!grepl("^\\%", pattern)) pattern <- paste0("%", pattern) if (!grepl("\\%$", pattern)) pattern <- paste0(pattern, "%") } Q <- paste0("SELECT * FROM pg_catalog.pg_tables WHERE tablename SIMILAR TO \'", pattern , "\'" , {if (!is.null(schema)) sprintf(" AND schemaname SIMILAR TO '%s'", schema)} , collapse="\nUNION\n") if (isTRUE(use.view)) Q <- tableToView(Q) runQry(Q, verbose=FALSE, results.not.expected=TRUE, cluster=cluster, wh=wh, dbname=dbname, schema=schema, snowflake_inuse=snowflake_inuse, all.pfm=all.pfm) } qViewExists <- function(..., cluster=NULL, use.view=TRUE, snowflake_inuse=getOption("snowflake_inuse")) { if (!missing(use.view)) warning("Arugment 'use.view' is ignored in qFindView()") qTableExists(..., cluster=cluster, use.view=TRUE, snowflake_inuse=snowflake_inuse) } qTableExists <- function(tbl, exact=TRUE, ignoreCase=TRUE, cluster=NULL, all.pfm=getOption("db.all.pfm"), use.view=FALSE, showWarnings=FALSE, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=NULL, snowflake_inuse=getOption("snowflake_inuse"), msg_sf = getOption("snowflake_msg_sf", TRUE), verbose=FALSE) { ## Returns the number of rows matching. ie 0 for FALSE. 1+ for TRUE ret <- qFindTable(pattern=tbl, exact=exact, ignoreCase=ignoreCase, cluster=cluster, use.view=use.view, wh=wh, dbname=dbname, schema=schema, snowflake_inuse=snowflake_inuse, all.pfm=all.pfm, msg_sf=msg_sf, showWarnings=showWarnings, verbose=verbose) return(nrow(ret)) } ## THIS COMMENT IS AN EXAMPLE OF qCanRead() ## EG: # tbl <- c("staging_raw_itunes", "staging_raw_spotify") # schema <- c("", "production") # user <- c("mdrake", "rsaporta") # # qCanRead(tbl=tbl, user=user, schema=schema) # makeProperColName <- function(tbl, col, AS.prefix=tbl, AS.suffix=NULL, sep="_", verbose=FALSE) { ## end a column name in ' AS' and the alias name will automatically ## be created ## All other values will not be postpended. ## Note that this allows you to manually determine alias by ## simply filling in the value after the " AS" if (!is.null(AS.suffix) && missing(AS.prefix)) AS.prefix <- NULL ## append the sep dot, prior to the paste0 so that no extra sep's present if (!is.null(AS.prefix)) AS.prefix <- paste0(AS.prefix, sep) if (!is.null(AS.suffix)) AS.suffix <- paste0(sep, AS.suffix) if (verbose) cat("\n\tsuffix is ", AS.suffix, "\n\tprefix is ", AS.prefix, "\n\n") repl <- paste0("\\1\\2"," ", AS.prefix, "\\1", AS.suffix) col <- gsub("(.+)( AS)$", repl, col, perl=TRUE) ## < if needed > # noAlias <- !grepl(" AS ", col) # col[noAlias] <- paste0(col[noAlias], " AS ", col[noAlias]) paste0(tbl, ".", col) } ### SPECIFIC TO ORCH getGenresForAllupcs <- function(id.col.AS="upc", dontExecute.justReturnQry=FALSE, verbose=TRUE, verbose.max.width=98L+(200*dontExecute.justReturnQry), cluster=NULL) { getReleaseDates(values=NULL, id.col="releaseId", id.col.AS=id.col.AS, minimal=TRUE , dontExecute.justReturnQry=dontExecute.justReturnQry, cluster=cluster, verbose=verbose) } getReleaseDates <- function(values, id.col="releaseId", id.col.AS=id.col , numericValues=FALSE, sep="__", origin="1970-01-01" , genre=TRUE , dontExecute.justReturnQry=FALSE, verbose=TRUE, verbose.max.width=98L+(200*dontExecute.justReturnQry) , minimal=is.null(values) , cluster=NULL) { force (minimal) ## TODO: This function is now outdated warning("This is somewhat outdated now. Especially the databases. \n\nPlease review code for getReleaseDates() in DB_Utils.r\n\n\n") if (length(genre) && !is.logical(genre)) stop ("genre should be a logical value") if (!isTRUE(genre)) # treat NULL / NA as FALSE genre <- FALSE quotes <- ifelse(numericValues, "", "\'") ## DROP NA's if (!is.null(values)) values <- unique(removeNA(values)) cols.rel <- c( paste(id.col, "AS", id.col.AS) , "releaseName AS 'release'", "releaseDate" , "artistId AS artistId" , "mkt_priority", "product_type_id" ) ## Some additional columns to include if (!minimal) cols.rel <- c(cols.rel, "vendor_catalog_number", "imprint" , "sale_start_date", "original_release_date" , "deletions", "manufacturer_upc", "version" , "subaccountId", "display_upc", "vendor_release_identifier" , "date_added AS", "last_updated AS", "date_created AS") cols.art <- c("artistName AS artist", "last_updated AS", "date_created AS") cols.label <- c("labelName AS label", "labelId", "priority AS label_priority") cols.genre <- c("genreName AS genre") cols.rel <- makeProperColName("R", cols.rel, AS.suffix="rel", verbose=FALSE, sep=sep) cols.art <- makeProperColName("A", cols.art, AS.suffix="art", verbose=FALSE, sep=sep) cols.label <- makeProperColName("L", cols.label, AS.suffix="label", verbose=FALSE, sep=sep) cols.genre <- makeProperColName("G", cols.genre, AS.suffix="genre", verbose=FALSE, sep=sep) ## IF adding columns from new tables, ## (a) append `select.what` and (ba) append the table name in the joins select.what <- c(cols.rel, cols.art, cols.label) if (genre) select.what <- c(select.what, cols.genre) if (minimal) { ## Drop meta-data dates select.what <- select.what[!grepl("created|updated|added", select.what)] ## Drop any foreign keys select.what <- select.what[!(grepl("Id$", select.what) & !grepl(paste0("AS ", id.col.AS, "$"), select.what))] } ## ------------ THE QUERY ------------ # Q <- paste( "SELECT ", paste(select.what, collapse=", ") , ", C1.country_code AS artist_country" , ", C2.country_code AS label_country" , "FROM dim_release R" , "LEFT JOIN dim_artist A ON R.artistId = A.artistId" , "LEFT JOIN dim_label L ON R.labelId = L.labelId" , "LEFT JOIN dim_country C1 ON C1.countryId = A.countryId" , "LEFT JOIN dim_country C2 ON C2.countryId = L.countryId" , if (genre) "LEFT JOIN dim_genre G ON R.genreId = G.genreId" , if (!is.null(values)) paste("WHERE", paste0("R.", id.col), "in", pasteQ(values, q=quotes)) # , "ORDER BY artistName" , sep=" " ) ## ------------ THE QUERY ------------ # ## Optionally, just output the Query if (dontExecute.justReturnQry) { if (verbose) cat("\n", verboseQry(Q, max.lines=35L, max.width=verbose.max.width, indentOr=TRUE), "\n", sep="") return(invisible(Q)) } ## Try to detach RPostgreSQL ## ------------------------------ ## try(dbDisconnect(con), silent=TRUE) try(detach(package:RPostgreSQL), silent=TRUE) require(RMySQL) ## ------------------------------ ## DT.releaseDates <- runQry(qry=Q, verbose.shortCircuit=FALSE, cluster=cluster) ## WRAPPING IN TRY SO AS TO NOT KILL THE WHOLE RUN IF FAILED succesfulRun <- FALSE try({ cleanReleaseDatesDT_(DT.releaseDates, minimal=minimal) succesfulRun <- TRUE }) if (!succesfulRun) warning("Qry was successful but this function failed at the point of cleaning the column types.\nPlease manually check") return(DT.releaseDates) } cleanReleaseDatesDT_ <- function(DT.releaseDates, id.col.AS=ifelse("upc" %in% names(DT.releaseDates), "upc", "releaseId") , minimal=FALSE , origin="1970-01-01" ,failOnMissingCols=!minimal, showWarnings=!minimal, verbose=FALSE) { idCols <- c( id.col.AS, "artistId", "product_type_id" , "subaccountId", "manufacturer_upc", "display_upc" , "vendor_release_identifier", "vendor_catalog_number") stringCols <- c( "artist", "release", "genre", "imprint" , "mkt_priority" , "artist_country", "label_country" , "version", "deletions" ) integerCols <- NULL dateCols <- c("releaseDate", "sale_start_date", "original_release_date") posixCols <- c("date_added__rel", "last_updated__rel", "date_created__rel" , "last_updated__art", "date_created__art") ## WRAPPING IN TRY SO AS TO NOT KILL THE WHOLE RUN IF FAILED fixColTypes_ (DT.releaseDates, character=stringCols, idcol=idCols , integer=integerCols, numeric=NULL, Date=dateCols , verbose=verbose, origin=origin , failOnMissingCols=failOnMissingCols, showWarnings=showWarnings) setattr(DT.releaseDates, "ReleaseDatesCleaned", TRUE) return(invisible(DT.releaseDates)) } infinitDBNotWorking.metas <- function(DT.nm) { # This is for use in TryCatch # error=function(e) {infinitDBNotWorking.metas("Some.DT.nm")} # where `Some.DT.nm` is probably `DB.Meta` or `DB.all.meta` fileToImport <- as.path(srcDir, "data_to_src", paste0(DT.nm, ".ManualExport.csv")) if (!file.exists(fileToImport)) stop("\nCould not fiind file '", fileToImport, "'\n") ## ELSE DT <- fread(fileToImport, colClass="character", na.strings=c("NA", "NULL")) cleanReleaseDatesDT_ (DT, "upc", minimal=TRUE, verbose=TRUE) return(setkey(DT, "upc")) } qShowErrs <- function(all=FALSE, verbose=TRUE, userid=NULL, lastN=1, cluster=NULL) { if (is.numeric(all) && missing(lastN)) { lastN <- all all <- FALSE } qry <- "SELECT * FROM stl_load_errors" ## TODO: Make a function where() which appropriately defaults to "" and appends AND as necessary whereClause_noAnd <- paste("WHERE", if (!is.null(userid)) sprintf("userid in (%s)", commaSep(userid))) whereClause <- paste(whereClause_noAnd, ifelse("WHERE " == whereClause_noAnd, "", "AND")) ## Find the max time if (!all) { ## Initialize to some very future time maxTime <- Sys.time()+5e9 maxTime <- sprintf("(SELECT max(starttime) FROM stl_load_errors %s)", ifelse(is.null(userid), "", whereClause_noAnd)) ## find the greatest less than each maxTime for (N in seq(lastN-1)) maxTime <- runQry(sprintf("SELECT max(starttime) as latest_time FROM stl_load_errors %s starttime < %s", whereClause, maxTime), verbose=FALSE, cluster=cluster)[1, pasteQ(latest_time)] qry <- sprintf("%s %s starttime >= %s", qry, whereClause, maxTime) } Errs <- runQry(qry, verbose=FALSE, cluster=cluster) ## Trim character cols trim_these <- nwhich(sapply(Errs, is.character)) Errs[, (trim_these) := lapply(.SD, trim ), .SDcols=trim_these] Errs[, raw_line_35 := substr(raw_line, 1, 35)] Errs[, raw_line_full := lapply(raw_line, function(x) list(x))] Errs[, raw_line := NULL] Errs[, errs_per_q := .N, by=c("userid", "query")] Errs[, uniq_errs_per_q := .N, by=c("userid", "query")] Errs[, cluster := getCluster()] Errs <- Errs[order(starttime, decreasing=TRUE)] if (verbose) print(unique(Errs, by=c("userid", "query"))[order(userid, starttime, decreasing=TRUE)]) return(invisible(Errs)) } # ------------------------------------------------ # qShowDist <- function(cluster=NULL) { Q <- paste( "SELECT slice, col, num_values, minvalue, maxvalue", "FROM svv_diskusage ", "WHERE name='users' ", " AND col=0", "ORDER BY slice, col;" ) runQry(Q, cluster=cluster) } # ------------------------------------------------ # qGetAll <- function(tbl, filter=list(), schema=NULL, limit=NULL, dont.check.table=FALSE, confirmed.single.quotes.are.escaped=FALSE, verbose=TRUE, ... ) { ## Quick query from a single table, using a list to filter. ## The list is combined using AND. ## (...) are wrapped up and collected into filter() ## ## (...) can be preceeded by 'min.___' or 'max.___' warning("NOT SURE IF YOU SHOULD STILL USE THIS FUNCTION qGetAll() -- it is very old") ## Allow for an unquoted argument if (!exists(as.character(substitute(tbl))) || !is.character(tbl)) tbl <- as.character(substitute(tbl)) if (length(filter)) { warning ("filter in qGetAll() will be changed in the future. Use the dots instead") } ## These are a few tables which I know I want dim for, and this saves the time of the extra query tbl <- tableNameClean(tbl, schema=schema) ## Allow for quick calls, like qGetAll(tbl, 100) if (is.numeric(filter) && length(filter) == 1 && is.null(names(filter))) { limit <- filter filter <- list() } ## Allow for quick table names like 'itunes' and 'spotify' if (!dont.check.table) { if (tbl == "analytics") tbl <- "fact_analytics" AllTables <- qShowTables(tbl) [["tablename"]] ## If there are no matches, that is a problem if (!length(AllTables)) stop("\nThere is no table in the DB with a name like '", tbl, "'\n") ## If there is exactly one match, use that. if (length(AllTables) == 1) { tbl <- AllTables[[1]] ## Otherwise there are multiple matches. If there is a perfect match, then leave tbl alone (do nothing) ## So if tbl is not in AllTables, that means there are multiple matches, none is perfect. ## Have the user choose manually } else if (!tbl %in% AllTables) { ## if tbl starts with raw_.. check if adding staging_.. will help if (grepl("^raw_", tbl) && paste0("staging_", tbl) %in% AllTables) { tbl <- paste0("staging_", tbl) ## check if adding 'dim_' to the table makes it legit } else if (paste0("dim_", tbl) %in% AllTables) { tbl <- paste0("dim_", tbl) } else { opts <- sprintf("%4i .... %s", seq(AllTables), AllTables) cat("\n\t\t\t Matching Tables Are:", pasteR(40), opts, "\n", sep="\n\t\t") opt <- readline("Please select an option: > ") if (tolower(opt) %in% c("x", "q")) return(message("Quitting by User Exit. No Queries were run.\n")) if (opt %in% AllTables) tbl <- opt else if (as.numeric(opt) %in% seq(AllTables)) tbl <- AllTables[as.numeric(opt)] else stop("\nThat was an invalid seleection. Valid options were a number betwee 1 and ", length(AllTables), ".") } } } # // end if '(!tbl %cont% "_")' verboseMsg(verbose, "Pulling from table ", tbl, time=FALSE) if (!is.list(filter) || length(names(filter)) != length(filter)) stop ("'filter' must be a named list.") ## COLLECT THE DOTS addlFilters <- list(...) filter <- c(filter, addlFilters) ## Determine which symbol to use Many <- sapply(filter, length) > 1 Mins <- grepl("^min\\.", names(filter)) Maxs <- grepl("^max\\.", names(filter)) names(filter) <- gsub("^(min|max)\\.", "", names(filter)) eq.symbol <- ifelse(Many, "IN", ifelse(Mins, ">=", ifelse(Maxs, "<=", "="))) verboseMsg(verbose, "Symbols being used are: ", paste(names(filter), eq.symbol, collapse=", "), time=FALSE) QuoteAndFlatten <- function(x) { if (is.character(x) || is.factor(x) || is.Date(x) || is.POSIXct(x)) pasteQ(x, wrap=ifelse(length(x)==1, "", "(")) else if (length(x) == 1) x else pasteQ(x, q="") } if (length(filter)) { ## check for \' in values if (filter %cont% "'" && !confirmed.single.quotes.are.escaped) stop("\nSome items in filter contain single quotes. Please make sure they are properly escaped then\n set 'confirmed.single.quotes.are.escaped=TRUE'\n") where <- sprintf("%s %s %s", names(filter), eq.symbol, sapply(filter, QuoteAndFlatten) ) where <- paste("WHERE", paste(where, collapse=" AND ")) } else where <- "" lim <- ifelse(is.numeric(limit) && is.finite(limit), sprintf(" LIMIT %i", limit), "") qry <- sprintf("SELECT * from %s %s %s", tbl, where, lim) runQry(qry, verbose=verbose) } createTblFromQry <- function(qry, tbl, schema, cluster=NULL, verbose=TRUE) { qry.create <- makeSQLtable(runQry(qry, verbose=FALSE, cluster=cluster), table.name=tbl, schema=schema, quiet=TRUE)[["QRY.create"]] runQry(qry.create, cluster=cluster, verbose=verbose) } qRenameTable <- function(tbl.old, tbl.new=tbl.old, schema=NULL, schema.old=schema, schema.new=schema.old, cluster=NULL, verbose=TRUE, verbose.qry=FALSE, msg.pre=NULL) { if (missing(tbl.old) && missing(tbl.new)) stop ("tbl must be specified") if (!identical(schema.old, schema.new)) { stop("Currently, redshift does not allow tables to be moved between schemas.\nIn other words, schema.old and schema.new must be the same") } if (missing(verbose) && !missing(msg.pre) && !is.null(msg.pre)) verbose <- FALSE stbl.old <- schemaPaste(schema.old, tbl.old) stbl.new <- schemaPaste(schema.new, tbl.new) ## TODO 20141126 stbl.new <- tbl.new if (identical(stbl.old, stbl.new) || (identical(schema.old, schema.new) && identical(tbl.old, tbl.new))) { warning ("old and new are identical ('", stbl.old, "'). No rename to perform") return (invisible(NULL)) } ## Verbose to user verboseMsg(verbose, msg.pre, "renaming from ", stbl.old, " to ", stbl.new, func="message") ## Create query qry <- sprintf("ALTER TABLE %s\nRENAME TO %s", stbl.old, stbl.new) ## Execute runQry(qry, cluster=cluster, verbose=verbose.qry) } makeDropQryFromCreate <- function(qry.create, instruction=c("DROP", "TRUNCATE")) { ## 20141010 - is this a better regex? gsub("(.*CREATE (VIEW|TABLE)\\s+)(.+)*?(\\s+.*)", "\\3", qrys.create) instruction <- match.arg(instruction) pat.create <- "\\bCREATE\\s+((VIEW|TABLE)\\s+([A-Za-z]+\\.)?[A-Za-z_][A-Za-z\\._]+)\\s+(AS\\s)?.+" repl <- paste0(instruction, " \\1") if (is.list(qry.create)) lapply(qry.create, function(x) trim(gsub(pat=pat.create, rep=repl, x=x))) else trim(gsub(pat=pat.create, rep=repl, x=qry.create)) } qDiskUsage <- function(limit=NULL, verbose=TRUE) { qry <- " select trim(pgdb.datname) as Database, trim(pgn.nspname) as Schema, trim(a.name) as Table, b.mbytes, a.rows from ( select db_id, id, name, sum(rows) as rows from stv_tbl_perm a group by db_id, id, name ) as a join pg_class as pgc on pgc.oid = a.id join pg_namespace as pgn on pgn.oid = pgc.relnamespace join pg_database as pgdb on pgdb.oid = a.db_id join ( select tbl, count(*) as mbytes from stv_blocklist group by tbl ) b on a.id = b.tbl order by mbytes desc, a.db_id, a.name; " runQry(qry, verbose=verbose) }