# ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- # # ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- # # # # File Name : makeSQLtable.r # # Last Updated Funclist : 04 May 2015, 5:02 PM (Monday) # # # # Author Name : Rick Saporta # # Author Email : RickSaporta@gmail.com # # Author URL : www.github.com/rsaporta # # # # Packages Called : NA # # Packages Used via NS : NA # # # # ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- # # # # ingestIntoSQL ( DT, nmsToChange=c(), nmsToChange.Out=names(nmsToChange), keyColsForSQL=key(DT) # # , tbl=paste0("New_table_", timeStamp()), schema="ds_scratch", quiet=TRUE # # , numericDecimals=c(22, 9), intColsToCoerce=whichColsCanBeInteger(DT.out) # # , guessIntCols=TRUE, append=!drop, drop=FALSE, columnsDropAndAdd=!drop & !append # # , truncInsteadOfDrop=TRUE, permissions.update=FALSE # # , permissions.groups.to.update=c("dw_queriers", "biapps"), confirm=!append # # , add.ingestDate=FALSE, showStatusUpdate=TRUE, cluster=NULL, verbose.qry=FALSE # # , verbose=TRUE, dry.run=FALSE ) # # makeSQLaddCols ( DT, encode=NULL # # , table.name=gsub("^DT\\.", "", capture.output(substitute(DT))), schema=NULL # # , factorToChar=TRUE, quiet=FALSE, numericDecimals=c(18, 6) ) # # makeSQLtable ( DT, encode=NULL # # , table.name=gsub("^DT\\.", "", capture.output(substitute(DT))), schema=NULL # # , diststyle="EVEN", sortkey=key(DT), factorToChar=TRUE, quiet=FALSE # # , numericDecimals=c(18, 6), justDesc=FALSE # # , datetime_type=c("DATETIME", "TIMESTAMP", "TIMESTAMP_LTZ", "TIMESTAMP_NTZ", "TIMESTAMP_TZ"), boolean_type=c("BOOLEAN", "INT", "VARCHAR(1)"), comment=NULL ) # # escapeChars ( x ) # # cleanColNamesForSQL_ ( x, MustStartWithAlpha=TRUE ) # # chopQryInsert ( QRY.insert, maxBytes=1e+06, linebreak=",\n", verbose=TRUE ) # # # # # # # # ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- # # ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- # # TODO 2015-07-01 # If NOT dropping the table, confirm all the types are still the same # Especially for Integers .. make sure there are no numbers that exceed the type if (FALSE) { testRun_is_snowflake <- TRUE ## --------- TEST CASES -------------- ## ## Testing a single Column tmp_DT.test_one_column <- data.table(test=c("not quoted", "this \"line\" has quotes", "also not quoted")) catheader("singleColumn plus ingestDate", endl=2) ingestIntoSQL(DT=tmp_DT.test_one_column, schema="bi", tbl="TESTING_QUOTED_COLUMNS", dbname="dev", add.ingestDate=FALSE, drop=TRUE, snowflake_inuse=testRun_is_snowflake) catheader("singleColumn without ingestDate", endl=2) ingestIntoSQL(DT=tmp_DT.test_one_column, schema="bi", tbl="TESTING_QUOTED_COLUMNS", dbname="dev", add.ingestDate=TRUE, drop=TRUE, snowflake_inuse=testRun_is_snowflake) ## Testing two columns tmp_DT.test <- data.table(test=c("not quoted", "this \"line\" has quotes", "also not quoted"), value=101:103) catheader("Two columns plus ingestDate", endl=2) ingestIntoSQL(DT=tmp_DT.test, schema="bi", tbl="TESTING_QUOTED_COLUMNS", dbname="dev", append=FALSE, drop=TRUE, add.ingestDate=FALSE, snowflake_inuse=testRun_is_snowflake) catheader("Two columns without ingestDate", endl=2) ingestIntoSQL(DT=tmp_DT.test, schema="bi", tbl="TESTING_QUOTED_COLUMNS", dbname="dev", append=FALSE, drop=TRUE, add.ingestDate=TRUE, snowflake_inuse=testRun_is_snowflake) } ## ............ TO CREATE ................ # sameDesc <- function(DT, tbl, schema) { # ## TODO: This function will query the DB see if the schema.tbl exists, and if so, check if it is of the same format as tbl would be # Useuful when running ingestIntoSQL() multiple times on the same DT and dont want to drop each time, unless the schema is different. # } ## ............ TO CREATE ................ { ingestIntoSQL <- function( DT , nmsToChange = c() , nmsToChange.Out = names(nmsToChange) , keyColsForSQL = key(DT) , tbl = paste0("New_table_", timeStamp()) , quiet = TRUE , numericDecimals = c(22, 9) , intColsToCoerce = whichColsCanBeInteger(DT.out, ignoreIntegers=FALSE) , guessIntCols = TRUE , maxBytes_for_insertStatement = 2e7 , append = !drop , drop = FALSE , columnsDropAndAdd = !drop & !append , truncInsteadOfDrop = TRUE , permissions.update = FALSE , permissions.groups.to.update = c("dw_queriers", "biapps") , confirm = !append , add.ingestDate = FALSE , showStatusUpdate = TRUE , cluster=NULL ## Copying some DTs take a long time. If we don't mind modifiying the original DT (such as when the DT is already a copy or if on a nightly run), then set this flag to true , dont_take_copy_of_DT=FALSE , transient=TRUE ## Applies only to Snowflake ## Dont include datetime_type, since it will mess up the warning in makeSQLtable # , datetime_type = c("DATETIME", "TIMESTAMP_NTZ", "TIMESTAMP_TZ", "TIMESTAMP_LTZ", "TIMESTAMP") , boolean_type = c("BOOLEAN", "INT", "VARCHAR(5)", "VARCHAR(1)") ## If creating table or schema , comment_tbl=NULL , comment_tbl_footer=getSFCommentFooter() , comment_schema=NULL , wh = getSnowflakeWH() , dbname = getSnowflakeDB() , schema = "bi" , snowflake_inuse = getOption("snowflake_inuse") , dry.run=FALSE , verbose.qry = FALSE , verbose=TRUE , ... ## passed to makeSQLtable ) { ## Warpper function to makeSQLtable(). ## This function takes as input a DT, ## cleans it up (using a copy. No modification made to original) ## then inserts it into a SQL table. ## Will also create the table first and will optionally drop an existing table ## nmsToChange : can be a dict, where the names are the NEW values and the values are the OLD values ## guessIntCols : will take abs of remainder mod 1 and sum it. Anycol that sums to 0 gets coerced to integer ## TODO: Add an option for a minimal verbose, and add these outputs: # cat("Dropping existing table ....... ") # cat("Creating table ................ ") # cat("Inserting data into table ..... ") start_time <- now() ##TODO - we can test the DT first, before loading this package stopifnot(require(bit64, quietly=TRUE)) if (!missing(DT) && !nchar(getProjName())) stop ("When passing a DT to makeSQLtable(), there must be a project set, since writeDT needs to use out.p()\nHINT: call\n setScience(proj='misc', subl=FALSE, load=FALSE)", call.=FALSE) if (missing(boolean_type) && snowflake_inuse) boolean_type <- "BOOLEAN" else { boolean_type <- toupper(boolean_type) boolean_type <- match.arg(boolean_type) } if (!missing(columnsDropAndAdd)) stop("columnsDropAndAdd is not yet implemented") if (!is.numeric(maxBytes_for_insertStatement)) stop ("maxBytes_for_insertStatement must be an integer in bytes") ## If the DT argument is not a data.table ## Check to see if it is the name of a data.table. ## If so, get() it if (!is.data.table(DT) && is.character(DT) && exists(DT) && is.data.table(get(DT))) { nm.dt <- DT DT <- get(DT) } else { ## Capture the name for later use nm.dt <- as.character(capture.output(substitute(DT))) } ## If the DT argument is still not a table, that's an error if (!is.data.table(DT)) stop ("DT must be a data.table or the name of a data.table") ## Sometimes, user will mistakenly use the argument `create = TRUE`, as in "create the table if it does not exist" (or create the schema) ## However, there is no `create` flag; this function *always* creates the table & schema if it does not exist ## There however IS a flag called create_only, which *only* creates the table (and schema) but *does not populate the data* if (length(wh.create <- extract("create", names(list(...))))) { warning ("Looks like there is a create_only type argument ('", wh.create, "') in the dots to ingestIntoSQL\nThis will be interpreted by makeSQLtable as to only return the create qry and not the ingest query.\n\nHINT 1: If the intention is to create the table, without ingesting the data, you must do so manually (process not yet implemented)\n\nHINT 2: If the intention is to, say, create the schema if it does not exist, this will happen automatically", call.=FALSE) } if (snowflake_inuse && missing(comment_tbl) && is.null(comment_tbl)) { if (is.null(getProjName())) stop("project needs to be set for auto-comment of tbl\nHINT: set comment_tbl=NULL to not use auto-comment") ## OLD: comment_tbl <- pasteC(capture.output(substitute(DT))) %>% sprintf("Auto created from %s from project %s", ., getProjName()) comment_tbl <- sprintf("Auto created from %s from project %s", nm.dt, getProjName()) } ## Cleaning up my sloppy mess ## I allow for the use of append and drop interchangably. (Stick with this, as months from now I might use one or the other) ## However, in the code INSIDE the function, I use them interchangably. This is a mistake { if (!missing(append) && missing(drop)) drop <- append else if (missing(drop) && !missing(append)) append <- drop } ## Take a deep copy if (dont_take_copy_of_DT) { message("WARNING -- We are NOT taking a copy of the DT, meaning that the original DT will likely be modified") DT.out <- DT } else { verboseMsg(verbose, "Taking a copy of the DT") DT.out <- copy(DT) } ## COERCE TO INTEGER if (length(intColsToCoerce) && !identical(intColsToCoerce, FALSE)) { numericCols <- setdiff(nwhich(sapply(DT.out, is.numeric)), intColsToCoerce) if (verbose) message(sprintf("The following cols will be converted to integer:\n%s", paste_l(intColsToCoerce, 4, "\t ", maxLineWidth=getWidth(max=300), preline=" "))) if (verbose && length(numericCols)) message(sprintf("The following cols will be treated as numeric:\n%s", paste_l(numericCols, 4, "\t ", maxLineWidth=getWidth(max=300), preline=" "))) DT.out[, (intColsToCoerce) := lapply(.SD, as.integer), .SDcols=intColsToCoerce] } ## Any idCols which remain which have not been converted to ints, must have their blanks converted to NAs for (col in nwhich(sapply(DT.out, is.idcol)) ) { DT.out[get(col) == "", (col) := NA] } ## check that names & replacement match in length if (length(nmsToChange) != length(nmsToChange.Out)) stop ("'nmsToChange' and 'nmsToChange.Out' are not the same length") ## If there are names to change, change the DT and keyColsForSQL if (length(nmsToChange)) { setnames(DT.out, nmsToChange, nmsToChange.Out) ## Replace names in the keys from DT with those of DT.out keyColsForSQL %<>% msub(pat=nmsToChange, repl=nmsToChange.Out, ., whole=TRUE) %>% cleanColNamesForSQL_ } if (add.ingestDate) { if ("ingestDate" %in% names(DT.out)) stop ("cannot add 'ingestDate' column since a column with that name already exists") verboseMsg(verbose, "adding column 'ingestDate' to the DT (and hence to the table)") DT.out[, ingestDate := Sys.Date()] } ## Set key of the new DT.out setkeyIfNot(DT.out, keyColsForSQL, verbose=FALSE) ## Clean up the names of the DT.out; Key will be preserved. cleanColNamesForSQL_ (DT.out) ## Extract the tbl name from DT if (missing(tbl)) { ## if the orginal DT was a string name of a data.table, nm.dt will have already been assigned if (!exists("nm.dt", inherits=FALSE)) nm.dt <- as.character(capture.output(substitute(DT))) nm.dt <- sub("^DT\\.", "", nm.dt) ## Only replace the default value if the extracted value has more than one char and starts with a letter if (nchar(nm.dt) > 1 && grepl("^[A-Za-z]", nm.dt) && !(toupper(nm.dt) %in% c('NA', 'NULL'))) tbl <- nm.dt message(sprintf("Using schema = '%s'; tbl = '%s'; ", schema, tbl)) } ## Clean up tbl name tbl %<>% cleanColNamesForSQL_ %>% toupper dst <- dbschematbl(dbname=dbname, schema=schema, tbl=tbl) # browser(text = "Here is where it keeps crashing") # ## Different options to makeSQLtable based on if Snowflake or Redshift # if (snowflake_inuse) { # Addl_ARGS <- list(encode=NULL, upperColnames=TRUE, quoteColnames=FALSE, quoteBools=TRUE, boolean_type="VARCHAR(1)", diststyle=NULL, sortkey=NULL) # Addl_ARGS[names(list(...))] <- list(...) # QRY.list <- do.call(makeSQLtable, c(list(DT.out, table.name=tbl, schema=schema, quiet=quiet, numericDecimals=numericDecimals), Addl_ARGS)) # } else # QRY.list <- makeSQLtable(DT.out, encode="lzo", table.name=tbl, schema=schema, quiet=quiet, numericDecimals=numericDecimals, ...) encode <- if (!isTRUE(snowflake_inuse)) "lzo" else NULL ## If using snowflake, we will upload to S3. Therefore dont waste time compiling the QRY.insert portion, which has to spend several minutes quoting strings, etc create_only <- snowflake_inuse QRY.list <- makeSQLtable(DT.out, encode=encode, table.name=tbl, quiet=quiet, numericDecimals=numericDecimals, wh=wh, dbname=dbname, schema=schema, create_only=create_only, snowflake_inuse=snowflake_inuse, boolean_type=boolean_type, ...) if (create_only) { QRY.create <- QRY.list[[1]] } else { QRY.create <- QRY.list[["QRY.create"]] QRY.insert <- QRY.list[["QRY.insert"]] } ## Add in comments for snowflake if (snowflake_inuse) { if (!is.null(comment_tbl)) { comment_clause <- paste0("\nCOMMENT = '", comment_tbl, " ", comment_tbl_footer, "';") ## Paste the comment to the end of the create qry by replacing the last semicolon with the comment_clause (which contains a new semicolon) QRY.create %<>% gsub("\\s*;?\\s*$", comment_clause, .) } if (transient) { QRY.create %<>% gsub("\\bCREATE TABLE\\b", "CREATE TRANSIENT TABLE", ., ignore.case=TRUE) } } ## check if Schema exists, if not, create it ## For now this only works on snowflake if (snowflake_inuse && !sfSchemaExists(schema=schema, dbname=dbname, wh=wh, verbose=FALSE) ) { verboseMsg(verbose, "Schema '", schema, "' does not yet exist -- creating it.", sep="") sfCreateSchema(schema=schema, dbname=dbname, wh=wh, comment=comment_schema, overwrite_if_exists=FALSE, check_if_exists=FALSE, showWarnings=FALSE, verbose=FALSE) } if (!snowflake_inuse) ## TODO: remove this line once established for non-snowflake warning ("Cannot check if schema exists for non-snowflake use", call.=FALSE) ## check if table exists already tbl.exists <- as.logical(qTableExists(tbl, dbname=dbname, schema=schema, snowflake_inuse=snowflake_inuse, msg_sf=FALSE, verbose=FALSE)) browser(expr=inDebugMode("SQL", "ingestIntoSQL"), text="in ingestIntoSQL() right before Qry execution and the dry.run clause.") ## Allow for the actual queries to be skipped. For debugging, etc if (!dry.run) { ## optionally set/change cluster if (!is.null(cluster)) { if (cluster != getCluster()) warning("the cluster argument in ingestIntoSQL() sets the cluster for the whole working environment") setDBall(cluster=cluster) } ## If the table exists, grab the columns currently in the table to check against the columns in the DT currentCols <- if (tbl.exists) qShowCols(tbl=tbl, schema=schema, dbname=dbname, verbose=FALSE) else NULL ## If the table exists already AND we are not dropping it, it is important that all of ## the colnames of DT be in the existing table, otherwise the query will crash. Might also segfault. if (tbl.exists && !drop) { if (length(cols.missing <- setdiffnocase(names(DT.out), currentCols))) { warning(warningCols(paste("The following columns are missing from ", dst, "and will be added"), cols.missing)) ## The strsplit/lapply is because snowflake does not allow for multiple statments in one call; ## However, the strsplit should be in the sfQry logic ## TODO ^^^^^^^ makeSQLaddCols(DT.out[, cols.missing, with=FALSE], table.name=tbl, dbname=dbname, schema=schema, quiet=quiet, encode=encode, snowflake_inuse=snowflake_inuse) %>% strsplit(";") %>% {.[[1]]} %>% lapply(.%>% runQry(verbose=FALSE, snowflake_inuse=snowflake_inuse) ) } } ## If table already exists and we are NOT appending if (tbl.exists && !append) { ## TODO: Run a query, to check the table description & column type, not just column names. If identical, then TRUNC instead of drop ## check for same names truncate <- truncInsteadOfDrop && all(names(DT.out) %in% currentCols) QRY.drop <- sprintf("%s TABLE %s\n", ifelse(truncate, "TRUNCATE", "DROP"), dst) verboseMsg(showStatusUpdate, ifelse(truncate, "* Truncating", "* Dropping"), " Table \"", dst, "\" ", sep="", minw=50) runQry(QRY.drop, verbose=verbose.qry, snowflake_inuse=snowflake_inuse) ## re-check if table exists, specificall to not run QRY.create if it does. tbl.exists <- as.logical(qTableExists(tbl, schema=schema, dbname=dbname)) } ## If NOT appending OR table does not yet exist, create it if (!tbl.exists) { ## try to turn on the warehouse if necessary if (snowflake_inuse) try(suppressWarnings(sfWarehouseOn(wh, set_warehouse=FALSE, showWarnings=FALSE, verbose=FALSE)), silent=TRUE) ## message the user verboseMsg(showStatusUpdate, "* Creating Table \"", dst, "\" ", sep="", minw=50) ## create table runQry(QRY.create, verbose=verbose.qry, snowflake_inuse=snowflake_inuse, wh=wh, dbname=dbname) if (permissions.update && !snowflake_inuse) qUpdatePerms(tbl=tbl, schema=schema, dbname=dbname, group=permissions.groups.to.update, verbose=TRUE, failOnError=FALSE, showWarnings=TRUE, snowflake_inuse=snowflake_inuse) } ## ------------------------------------------------------------------------------------------------------------------------ ## ## INSERT ## ------------------------------------------------------------------------------------------------------------------------ ## ## --------------------------- NOT SNOWFLAKE ------------------------------------- ## if (!snowflake_inuse) { ## Check if the insert statement is too large. Assuming approx 1 char == 1 byte ## If it is, chop it up and iterate over each one # * not needed - if (nchar(QRY.insert) > .95 * maxBytes_for_insertStatement) { warning ("2015-07-01:: Not sure if chopQryInsert() works correctly or not. There is a browser statement that was left in ingestIntoSQL() right before calling QRY.insert. Removing it for now\n\n2015-09-20 QRY.insert is no longer used for snowflake, so no worries") browser(expr=FALSE, text = "at chopping Query .... ") verboseMsg(showStatusUpdate, "Chopping Insert Query", sep="", minw=50) QRY.insert <- chopQryInsert(QRY.insert, max=maxBytes_for_insertStatement * 0.95, linebreak=",\n", verbose=FALSE) # * not needed - } L.ins <- length(QRY.insert) verboseMsg(showStatusUpdate || L.ins > 1, "* Inserting into Table \"", dst, "\" " , if (L.ins > 1) sprintf("using %i separate queries", L.ins), sep="", minw=50) # * not needed - if (L.ins == 1) # * not needed - runQry(QRY.insert, verbose=verbose.qry) # * not needed - else sapply(QRY.insert, runQry, verbose=verbose.qry, snowflake_inuse=snowflake_inuse) ## --------------------------- SNOWFLAKE ------------------------------------- ## } else { ## Make sure the WH is on try(suppressWarnings(sfWarehouseOn(wh, set_warehouse=FALSE, showWarnings=FALSE, verbose=FALSE, notify=FALSE)), silent=TRUE) ## Convert bools before writing to disk. The advantage to converting here as opposed to sfPopulateTable(), is that we already copied the DT. in sfPopulateTable() the DT would need to be copied a second time convertBooleans_(DT.out, boolean_type=boolean_type) ret <- sfPopulateTable(DT=DT.out, tbl=tbl, schema=schema, dbname=dbname, wh=wh) ## NOTE: no need to check for errors in ret; sfPopulateTable() will check, and will communicate to the user } ## ------------------------------------------------------------------------------------------------------------------------ ## } ## ------------ ## ## Confirm same ## ## ------------ ## browser(expr=inDebugMode("sqlconfirm", "ingestIntoSQL_confirm"), text="in ingestIntoSQL() AFTER updating, BEFTORE confirming") if (confirm) { verboseMsg(verbose, "Beggining Confirmation") confirm.failed <- FALSE QRY.confirm <- paste("SELECT * FROM ", dst) DT.confirm <- runQry(QRY.confirm, verbose=verbose.qry, results.not.expected=FALSE, snowflake_inuse=snowflake_inuse, wh=wh, dbname=dbname, msg_sf=FALSE) ## Convert IDcols as necessary if (length(wh.idcols <- nwhich(sapply(DT.out, is.idcol)))) DT.confirm[, (wh.idcols) := lapply(.SD, as.idcol), .SDcols=wh.idcols] ## Convert integer64 cols if (length(wh.int64Cols <- nwhich(sapply(DT.out, is.integer64)))) DT.confirm[, (wh.int64Cols) := lapply(.SD, as.integer64), .SDcols=wh.int64Cols] # old: int64Cols <- nwhich(sapply(DT.out, is.integer64)) # old: for (col in int64Cols) { # old: DT.confirm[, (col) := as.integer64(get(col))] # old: } ## Convert logical cols, if using snowflake if (length(wh.logcols <- nwhich(sapply(DT.out, is.logical))) && snowflake_inuse) { DT.confirm[, (wh.logcols) := lapply(.SD, as.logical), .SDcols=wh.logcols] } ## Convert factor cols if (length(wh.factorcols <- nwhich(sapply(DT.out, is.factor)))) { for (fcol in wh.factorcols) DT.confirm[, (fcol) := factor(get(fcol), levels=levels(DT.out[[fcol]]))] } s.t({ setkey(DT.out) setkey(DT.confirm) }, quiet.if.less.than=2, title="set keys for both DT.out and DT.confirm (across all columns in each DT)") ## Check that the CLASS matches for each column if (any(wh.diff_classes <- !mapply(identical, lapply(DT.out, class)[order(names(DT.out))], lapply(DT.confirm, class)[order(names(DT.confirm))]))) { warning ("Classes are different for DT.confirm than DT.out. This may impact the confirmation process.\nOffending columns are: ", pasteQand(nwhich(wh.diff_classes)), call.=FALSE) } ## check same dimension N.out <- nrow(DT.out) N.confirm <- nrow(DT.confirm) ## Check if there are Zero Rows in DT.confirm if (!nrow(DT.confirm)) { warning("DT.confirm has 0 Rows -- this normally means that nothing has loaded", call.=FALSE) confirm.failed <- TRUE wh.cn.val <- FALSE ## for verbose message at end of function ## Check if one has more rows than the other } else if (!identical(N.out, N.confirm)) { warning(sprintf("The DT in memory has %s rows while the table in the DB has %s rows", formnumb(N.out, round=FALSE), formnumb(N.confirm, round=FALSE)), call.=FALSE) ## Try to print the missing rows try({ if (N.out > N.confirm) suppressWarnings(print(DT.out[!DT.confirm])) else suppressWarnings(print(DT.confirm[!DT.out])) }, silent=TRUE) confirm.failed <- TRUE wh.cn.val <- FALSE ## for verbose message at end of function ## If rows are the same, check the details } else { ## Bank tha names, we use them a lot nms.dt.out <- names(DT.out) nms.dt.conf <- names(DT.confirm) if (!identical(sort(nms.dt.out), sort(nms.dt.conf))) { msg1 <- if (length(mis1 <- setdiff(nms.dt.conf, nms.dt.out))) { ## only a failure if we had dropped the table. Otherwise, straggling columns is possible confirm.failed <- confirm.failed || (drop && !truncInsteadOfDrop) sprintf("The following columns are in the DB table but NOT in the original DT\n%s\n", paste_l(mis1, 5)) } msg2 <- if (length(mis2 <- setdiff(nms.dt.out, nms.dt.conf))) { ## This is a failure confirm.failed <- TRUE sprintf("The following columns are in the original DT but NOT in the DB table\n%s\n", paste_l(mis2, 5)) } warning("\n", msg1, msg2) ## dif further # warning(sprintf("The DT in memory has %s rows while the table in the DB has %s rows", formnumb(N.out), formnumb(N.confirm))) } else { setcolorder(DT.confirm, names(DT.out)) } ## if DT.confirm is a superset of DT.out, then trim DT.confirm ## This also reorders the columns for DT.confirm to match DT.out if (all(nms.dt.out %chin% nms.dt.conf)) DT.confirm <- DT.confirm[, names(DT.out), with=FALSE] ## Check values are the same vals.check <- mapply(all.equal, DT.out, DT.confirm, tolerance=1e-4) %>% sapply(head, 1) if (!is.logical(vals.check) && any(wh.mode <- grepl("^Modes", vals.check))) { vals.check[wh.mode] <- apply(DT.confirm[, wh.mode, with=FALSE] == DT.out[, wh.mode, with=FALSE] | (is.na(DT.confirm[, wh.mode, with=FALSE]) & is.na(DT.out[, wh.mode, with=FALSE])) , 2, function(x) isTRUE(all(x))) } ## Convert to logical, but preserve names vals.check %<>% {setNames(obj=as.logical(.), nm=names(.))} ## all.equal sometimes produces bad results. If any values are NOT T/F, then use manual check if (any(wh.na_issue <- is.na(vals.check))) vals.check[wh.na_issue] <- manual.all.equal(DT.confirm[, wh.na_issue, with=FALSE], DT.out[, wh.na_issue, with=FALSE]) wh.cn.val <- !sapply(vals.check, isTRUE) ## factors will be messed up. Check for factors for none-same factors.to.check <- DT.out[, wh.cn.val, with=FALSE][, nwhich(sapply(.SD, is.factor))] ## They should be equal or NA. if (length(factors.to.check)) vals.check[factors.to.check] <- colSums(DT.confirm[, factors.to.check, with=FALSE] == DT.out[, factors.to.check, with=FALSE] | is.na(DT.confirm[, factors.to.check, with=FALSE])) == N.confirm ## factors will be messed up. Check for factors for none-same int64.to.check <- DT.out[, wh.cn.val, with=FALSE][, nwhich(sapply(.SD, is.integer64))] ## They should be equal or NA. if (length(int64.to.check)) vals.check[int64.to.check] <- colSums(DT.confirm[, int64.to.check, with=FALSE] == DT.out[, int64.to.check, with=FALSE] | is.na(DT.confirm[, int64.to.check, with=FALSE])) == N.confirm posix.to.check <- DT.out[, wh.cn.val, with=FALSE][, nwhich(sapply(.SD, is.POSIX))] ## They should be equal or NA. if (length(posix.to.check)) vals.check[posix.to.check] <- colSums(## They should all be less than a second apart abs(DT.confirm[, lapply(.SD, as.numeric), .SDcols=posix.to.check] - DT.out[, lapply(.SD, as.numeric), .SDcols=posix.to.check]) < 1 | is.na(DT.confirm[, posix.to.check, with=FALSE]) ) == N.confirm ## If any posix.to.check are still off, check the timezones if (length(posix.to.check) && any(!vals.check[posix.to.check])) { for (col in posix.to.check) { ind <- sample(seq(nrow(DT.out)), max(100000, nrow(DT.out)/5)) tz.out <- DT.out[ind, unique(format(get(col), "%Z"))] if (length(tz.out) != 1) warning("\nColumn '", col, "' has mixed timezones within the same column (on the R side);\nThis probably contributed to the mismatch with snowflake") else { tz.confirm <- DT.confirm[ind, unique(format(get(col), "%Z"))] if (length(tz.confirm) != 1) warning("\nWTF?!! Column '", col, "' from DT.confirm has mixed timezones;\nThis should not be the case. When it comes from snowflake it should have just one timezone.\n\nPLEASE INVESTIGATE\n") else if (tz.confirm != tz.out) warning("\nColumn '", col, "' has different timezones from R to Snowflake;\nThis probably contributed to the mismatch with snowflake\n\n in R : ", tz.out, "\n in SF: ", tz.confirm, "\n") } } } if (any(wh.cn.val <- !sapply(vals.check, isTRUE))) { confirm.failed <- TRUE warning(warningCols("Confirm Failed: Values are different", if (all(wh.cn.val)) " ALL COLUMNS " else names(DT.confirm)[wh.cn.val])) } } ## Output for user if (!confirm.failed) { message("Confirmed that ", dst, " has been written properly") ## cleanup suppressWarnings(rm(DT.confirm, wh.cn.val, QRY.confirm)) } else if (all(wh.cn.val)) { message("The issue might be in the confirmation process, since all ", length(wh.cn.val), " columns failed\nDT.confirm and DT.out being assigned to environment 'conf'") conf <- new.env() assign("DT.out", DT.out, envir=conf) assign("DT.confirm", DT.confirm, envir=conf) assign("conf", conf, envir=globalenv()) e <- environment() rm(conf, envir=e) } else if (any(wh.cn.val)) { message("Here is a sample of what failed") cat("DT.confirm: \n"); print(DT.confirm[, wh.cn.val, with=FALSE]) cat("DT.out: \n"); print(DT.out[, wh.cn.val, with=FALSE]) } if (confirm.failed) { sprintf("Total time to execute ingestIntoSQL() was %s", fwTDiff(start=start_time, end=now())) return(invisible(FALSE)) } } ## ------------ ## sprintf("Total time to execute ingestIntoSQL() was %s", fwTDiff(start=start_time, end=now())) return(invisible(TRUE)) } } makeSQLaddCols <- function(DT, encode=NULL, table.name=gsub("^DT\\.", "", capture.output(substitute(DT))), factorToChar=TRUE, quiet=FALSE, numericDecimals=c(18,6), upperColnames=FALSE, quoteColnames=TRUE , wh = getSnowflakeWH(), dbname = getSnowflakeDB(), schema = NULL, snowflake_inuse = getOption("snowflake_inuse")) { cols <- makeSQLtable(DT=DT, encode=encode, table.name=table.name, schema=schema, factorToChar=factorToChar, quiet=quiet, numericDecimals=numericDecimals, upperColnames=upperColnames, quoteColnames=quoteColnames, justDesc=TRUE, dbname=dbname, wh=wh, snowflake_inuse=snowflake_inuse) pasteC(sprintf("ALTER TABLE %s ADD COLUMN %s", dbschematbl(dbname=dbname, schema=schema, tbl=table.name), cols), C=";\n") } get_pat_ok_sql_col_names <- function() { ## This is the pattern for valid charachters in a SQL column name ## This is used by the function makeSQLtable() and create_and_populate_table_form_csv() pat <- "^[A-Za-z][A-Za-z0-9\\-_]*$" return(pat) } makeSQLtable <- function(DT, replace=FALSE, encode=NULL, table.name=tbl, tbl=gsub("^DT\\.", "", capture.output(substitute(DT))), diststyle="EVEN", sortkey=key(DT), factorToChar=TRUE, quiet=FALSE , numericDecimals=c(18,6) , justDesc=FALSE , datetime_type = c("DATETIME", "TIMESTAMP_NTZ", "TIMESTAMP_TZ", "TIMESTAMP_LTZ", "TIMESTAMP") , boolean_type = c("BOOLEAN", "INT", "VARCHAR(5)", "VARCHAR(1)") , comment=NULL , comment_footer = if (snowflake_inuse) getSFCommentFooter() , quoteBools=grepl("CHAR", boolean_type, ignore.case=TRUE) ## NOT USED YET , upperColnames=FALSE , quoteColnames=!snowflake_inuse , create_only = FALSE , min_char_idcol = 100 # the min length for idCols , wh = getSnowflakeWH() , dbname = getSnowflakeDB() , schema = NULL , snowflake_inuse = getOption("snowflake_inuse") ) { ## justDesc : if TRUE returns only the modified DESC table. This is useful for other functions which use the logic here. ## todo, breakout the inner logic into its own function if (snowflake_inuse) { if (missing(datetime_type)) { message("when using makeSQLtable() for Snowflake, don't forget to specify datetime_type. The default is no timezone (TIMESTAMP_NTZ), treated as UTC.\nUse , datetime_type=\'TIMESTAMP_TZ\' otherwise") datetime_type <- "TIMESTAMP_NTZ" } if (missing(boolean_type)) boolean_type <- "INT" } else { datetime_type <- match.arg(datetime_type) boolean_type <- match.arg(boolean_type) } ## We'll be using the names(DT) a few times, as well as modifying them if upperColnames is TRUE nms <- copy(names(DT)) ## If not quoting column names, make sure there are no offending characters if (!quoteColnames) { pat.ok <- get_pat_ok_sql_col_names() offending_nms <- nms[!grepl(pat.ok, nms)] if (length(offending_nms)) { stop ("some column names of DT are not permitted. Namely \n", pasteC(offending_nms, C=", "), "\n\nHINT: Try setting quoteColnames=TRUE") } } ## Make sure all column names are unique if (anyDuplicated(nms)) { stop("There are duplicate column names in DT") } if (is.call(substitute(DT)) && missing(table.name) && grepl("\\(|\\)|\"", table.name)) table.name <- paste0("New_table_", timeStamp()) ## Capture the info descr <- desc(DT, quiet=TRUE) ## Add a column for SQL type descr[, type := NA_character_] ## For debugging browser(expr=inDebugMode(c("SQL", "makeSQLtable", "makeSQLtable_top")), text="in makeSQLtable(), right before types.") ## idCols need special treatment. They could be either integers or strings ## if any non-numeric non-whiespace character found, consider it string idCols <- descr[.("idcol"), Columns] idCols_is_string <- nwhich(sapply(idCols, function(col) any(grepl("[[:alpha:]]", DT[[col]])))) if (length(idCols_is_string)) { message("The folllowing idcol columns will be treated as STRINGS: ", pasteQand(idCols_is_string, w="")) ## Note we calculate vchars.idcols for ALL idcols (not just idCols_is_string) so that the ifelse() downstream can work smoothly vchars.idcols <- (1+sapply(DT[, .SD, .SDcols=idCols], mnchar, na.rm=TRUE)) %>% pmax(min_char_idcol) } ## Translate R column types to SQL column types num_max <- sprintf("NUMERIC (%i, %i)", max(38, numericDecimals[[1]]), numericDecimals[[2]]) num_exact <- sprintf("NUMERIC (%i, %i)", numericDecimals[[1]], numericDecimals[[2]]) descr[.("numeric"), type := num_exact ] descr[.("integer64"), type := num_max ] descr[.("integer"), type := "INTEGER" ] descr[.("POSIXct"), type := datetime_type] descr[.("Date"), type := "DATE" ] descr[.("logical"), type := boolean_type ] descr[.("idcol"), type := ifelse(Columns %in% idCols_is_string, sprintf("VARCHAR(%i)", vchars.idcols), num_max)] ## "character" and "factor" need if-clauses, because the assignment to type depends on ## a function of Columns, which Columns cannot be NA for. ## However, descr[.('value')] will issue NA rows if 'value' is not one of the column values ## Specifically, the DT[.. ,SDcols=NA] will throw an error ## ## character if ("character" %in% descr$Class) s.t(descr[.("character"), type := sprintf("VARCHAR(%i)", 1+sapply(DT[, .SD, .SDcols=Columns], mnchar, na.rm=TRUE)) ], title="calculate what the max varchar is -- note that this step can be skipped for snowflake.", quiet.if.less.than=10) ## factor if ("factor" %in% descr$Class) { if (!factorToChar) { warning ("factor to enum() is not yet implemented, since it is not available in REDSHIFT") descr[.("factor"), type := sprintf("VARCHAR(%i)", 1+sapply(DT[, .SD, .SDcols=Columns], mnchar, na.rm=TRUE)) ] } else { descr[.("factor"), type := sprintf("VARCHAR(%i)", 1+sapply(DT[, .SD, .SDcols=Columns], mnchar, na.rm=TRUE)) ] } } ## Make sure we got all of column types (ie, any non-standard classes) if (any(is.na(descr$type))) { warning("Some types remain NA. Will quit now") print(dput(descr)) stop("Some types remain NA. Quitting.") } ## Add compression, except for booleans if (!is.null(encode)) { descr[!.("logical"), type := paste(type, "ENCODE", encode)] } ## Convert column names to upper case, if flagged if (upperColnames) { descr[, Columns := toupper(Columns)] nms <- toupper(nms) } ## the order of desc needs to be the order of the columns of the DT setkey(descr, Columns) descr <- descr[.(nms)] if (snowflake_inuse) descr[grepl("^VARCHAR", type, ignore.case=TRUE), type := "VARCHAR"] if (justDesc) return(descr[, paste(Columns, type)]) ## Add in schema and dbname table.name <- dbschematbl(dbname=dbname, schema=schema, tbl=table.name) QRY <- sprintf( "CREATE%s TABLE %s\n(\n %s\n)" , ifelse(isTRUE(replace), " OR REPLACE", "") , table.name , descr[, pasteC(Columns, type, C=",\n ")] ) if (!isTRUE(snowflake_inuse)) { if (!is.null(diststyle)) QRY <- paste0(QRY, "\n\n", "DISTSTYLE ", diststyle) if (!is.null(sortkey)) QRY <- paste0(QRY, "\n\nSORTKEY \n(\n" , pasteC(" ", sortkey, C=",\n") ,"\n)\n" ) } if (!is.null(comment)) { if (grepl("\'", comment)) { warning ("cannot use single quotes (') in comment. Removing altogher") comment <- gsub("\'", "", comment) } QRY <- paste0(QRY, "\nCOMMENT = '", paste(comment, comment_footer), "'") } ## Add a semicolon QRY <- paste0(QRY, ";") %>% setQry ## Output if (!quiet) cat("\nCreated Query is:\n\n", QRY, "\n\n", sep="") if (create_only) return(QRY) ## ------------- ------------- ## ## ------------- DONE WITH THE 'CREATE' PORTION ------------- ## ## ------------- ------------- ## ## ------------- NOW COMES THE 'INSERT' PORTION ------------- ## ## ------------- ------------- ## ## Some of the data that in R is not compatible or needs slight cleanup before converting to SQL ## eg, booleans need to be converted to lowercase (or other type), numbers need to be rounded, strings need to be quoted ## Thus we create a copy of the DT and modify accordingly ## For debugging browser(expr=inDebugMode(c("SQL", "makeSQLtable", "makeSQLtable_middle", "makeSQLtable_insert")), text="in makeSQLtable(), right before types.") ## Convert booleans to appropriate type ## Add quotes to non number columns DT.quoted <- copy(DT) if (upperColnames) setnames(DT.quoted, toupper(names(DT.quoted))) bools <- descr[Class=="logical", Columns] numbs <- descr[Class %in% c("numeric", "integer"), Columns] ## Quote all columns except numbers and booleans ## (note that even if quoting booleans, that is handled separately, so dont include in 'toquote') toquote <- setdiff(descr$Columns, c(bools, numbs)) escapeChars <- function(x) { gsub("('|\\\\)", "\\\\\\1", x) } ## Clean up BOOLEANS by (1) converting to appropriate string and (2) optinoally adding quotes if (length(bools)) { true_str <- 'true' false_str <- 'false' if (grepl("INT", boolean_type)) { true_str <- 1 false_str <- 0 } else if (boolean_type == "VARCHAR(1)") { true_str <- 't' false_str <- 'f' } if (quoteBools) { true_str <- paste0("'", true_str, "'") false_str <- paste0("'", false_str, "'") } ## convert from TRUE/FALSE to appropriate string format ## NOTE that NA's will be handled below DT.quoted[, c(bools) := lapply(bools, function(x) ifelse(get(x), true_str, false_str))] } ## Clean up NUMBS by rounding any numerics to appropriate digits. if (length(numbs)) { verboseMsg(verbose, "Converting ", length(numbs), " numeric columns") DT.quoted[, c(numbs) := lapply(numbs, function(x) round(get(x), numericDecimals[[2]]) ) ] } if (length(toquote)){ verboseMsg(verbose, "Cleaning & quoting ", length(toquote), " string columns") s.t(DT.quoted[, c(toquote) := lapply(toquote, function(x) paste0("'", escapeChars(as.character(get(x))), "'"))], title="quote and clean (some) char columns", quiet.if.less.than=10) } ## Collapse each row into a single string insert.values <- s.t(apply(DT.quoted, 1, pasteC, C=","), title="create insert.values via apply()", quiet.if.less.than=10) ## --------- NULL CLEANUP --------- ## ## Convert NA to NULL pat.NA <- "(\\b)NA(\\b)" # "(^|,|\\\"|\\\')NA(,|\\\"|\\\'|$)" # NA enclosed in commas, or quotes, or start or end of string. At start or end of line, it might have only one quote repl.NULL <- "\\1NULL\\2" safetybreak <- 0 while (any(grep(pat.NA, insert.values)) && safetybreak <= 1e3) { insert.values <- gsub(pat.NA, repl.NULL, insert.values) safetybreak <- safetybreak + 1 } ## Remove quotes around NULLs insert.values <- gsub("'NULL'", "NULL", insert.values) ## ------------------------------- ## ## Collapse all the rows into a single string, with parens around each row insert.values <- pasteC(sprintf("(%s)", insert.values), C=",\n") ##### FINAL OUTPUT WILL LOOK LIKE # # INSERT INTO table (column1, column2, …) # VALUES # (value1, value2, …), # (value1, value2, …) ,...; # table.columns <- pasteQ(names(DT.quoted), q=if (quoteColnames) "\"" else "") QRY.insert <- sprintf("INSERT INTO %s %s\nVALUES\n%s;", table.name, table.columns, insert.values) %>% setQry return(invisible(list(QRY.create=QRY, QRY.insert=QRY.insert))) } ## ------------------------------------------------------------------------------------------- ## cleanColNamesForSQL_ <- function(x, MustStartWithAlpha=TRUE, removespace=TRUE, removetrailingunderscores=TRUE, removedoubleunderscores=TRUE) { ## Changes the names of the data.table to lowercase, SQL-compliant names ## if x is NULL or names(x) is NULL, x will be returned unchanged. xisnames <- is.character(x) ## Check that x is a data.frame, string or NULL if (!inherits(x, "data.frame") && !xisnames && !is.null(x)) stop ("x must be a data.frame or data.table (or the names of such)") ## Grab the names of the DT (or x itself) nms <- if (xisnames) x else names(x) ## allow for graceful exit if no names, or if x itself is null. if (!length(x)) return(x) ## make lowercase and trim whitespace nms %<>% tolower() %>% trim() ## Remove offending characters nms %<>% removeNonAlphaNumeric(replace="_") ## Check that all column names start with a letter if (MustStartWithAlpha && any(non <- !grepl("^[A-Za-z]", nms))) { warning(warningCols("The following column names do not start with a letter\nand will be prepended with 'ABC_' ", paste0("'", nms[non], "'"))) nms[non] <- paste0("ABC_", nms[non]) } if (removespace) nms %<>% gsub("\\s", "_", .) if (removedoubleunderscores) nms %<>% gsub("_+", "_", .) if (removetrailingunderscores) nms %<>% gsub("_$", "", .) ## if x is just the names, return the cleaned version ## Otherwise, set the names of the DT and return that if (xisnames) return(nms) else return(setnames(x, nms)) } ## ------------------------------------------------------------------------------------------- ## chopQryInsert <- function(QRY.insert, maxBytes=2e7, linebreak=",\n", verbose=TRUE) { ## Certain APIs only allows for a limited size Qry - eg, snowflake allows for 1MB ## Insert queries can easily surprass that. Thus chop into smaller parts. ## regex pattern for 'INSERT INTO (..) VALUES' pat.insert <- "^\\s*INSERT\\s+INTO\\s+[A-Za-z0-9\\-_\\.]+\\s+\\(.+\\)\\s+VALUES\\s+" ## Input check is.char_of_length1(QRY.insert, fail.if.not=TRUE) is.char_of_length1(linebreak, fail.if.not=TRUE) ## If QRY.insert does not need chopping, return unchanged if (nchar(QRY.insert) < maxBytes * 0.95) return (QRY.insert) ## Check that regex pattern is present in QRY.insert -- if not, fail if (!grepl(pat.insert, QRY.insert)) stop ("QRY.insert should be a string of length 1 and starting with pattern 'INSERT INTO (..) VALUES'\nSuch a pattern could not be detected") ## Check that linebreak is present in QRY.insert -- if not, return unchanged if (!grepl(linebreak, QRY.insert)) { warning ("Could not find linebreak (", clean.capture.output(linebreak), ") in QRY.insert") return(QRY.insert) } ## Check that the escaped version of linebreak does NOT exist in QRY.insert -- if so, return unchanged escaped_linebreak <- escapeRegEx(clean.capture.output(linebreak, quote=FALSE)) if (grepl(escaped_linebreak, QRY.insert)) { warning ("an escaped verion of the linebreak string exists in the QRY.insert\nSplitting the string when this is the case is not yet implemented") return(QRY.insert) } if (FALSE) { browser(text = "chopQryInsert before splat_body") nc <- nchar(QRY.insert) formatBytes(nc) maxBytes <- 2e7 formatBytes(maxBytes) } ## Extract the header info header <- extract_gregexpr(pat.insert, QRY.insert) body <- gsub(pat.insert, "", QRY.insert) ## remove final semicolon if present. Trim whitespace. Assign back into body body %<>% trim %>% gsub(pat="\\s*;$", repl="", x=.) ## Split to max size s.t({ splat_body <- strsplitToMaxCharSize(string=body, pattern=linebreak, ignore.case=FALSE, chars=(0.99 * maxBytes) - nchar(header), showWarnings=FALSE, clearTrailingEmptyString=TRUE) }, msg=sprintf("Time it took to split the body into %i substrings is:", length(splat_body))) ## CONFIRM. Every line of splat_body should start and end with a parens if (any(wh <- !grepl("^\\(.*\\)$", splat_body))) { assign("offending_body", splat_body[wh], envir=globalenv()) stop ("Internal Error: 'splat_body' does not start and end with parens. The ", sum(wh), " offenses have been assigned to the object offending_body in the R Global environment") } ret <- paste0(header, " ", splat_body, ";") if (verbose) { nc <- nchar(ret) fc <- . %>% nchar %>% formnumb sm <- summary(nchar(ret)) cat(sprintf("The %s character query has been chopped into %s queries\n each of median length %s characters (min = %s, max = %s)", fc(QRY.insert), formnumb(length(ret), round=FALSE), formnumb(sm[["Median"]]), formnumb(sm[["Min."]]), formnumb(sm[["Max."]])), sep="", fill=TRUE) } return(invisible(ret)) } ## ------------------------------------------------------------------------------------------- ##