if (FALSE) { sprintf("SELECT MAX(processeddaytime) FROM production.fact_analytics WHERE storeid = 1 AND download_activity_date >= '2015-06-01' AND processeddaytime < '%s'", processed_lessthan) -> qqq ## RS <- runQry(qqq, cluster=cluster) RS %<>% setnames("Redshift") SF <- sfQry(qqq, wh=wh, dbname=dbname, schema=schema) SF %<>% setnames("Snowflake") ## print(SF); print(RS) } compare_snowflake_to_redshift <- function(tbl , colsToAgg=c(".RU", ".rowcount") , dateCol=getDateColForTBL(tbl, schema=schema) , colsToPull=c(dateCol, "storeid", "transactiontypeid") , wh="cron_jobs_small" , schema="production" , dbname = getOption("snowflake_dbname") , rowcount_addedto_colsToAgg = TRUE # , maxDate = qMaxDate(tbl, snowflake=TRUE, dbname=dbname, simplify=TRUE) , maxDate=NULL , processed_lessthan = NULL , refresh_connex=TRUE # , processed_lessthan = "2015-07-10 07:00:21" # , cluster=getOption("db.defaultcluster.in") , cluster=04 , verbose=TRUE ) { if (tbl != "fact_analytics") processed_lessthan <- NULL if (length(cluster) > 1) stop ("Only one cluster at a time") ## Todo, iterate over the combos setDBall(cluster=cluster) tbl %<>% tableNameClean ## TEMP - until download_activity_date / dayid issue is fixed in cluster 04 if (tbl == "fact_analytics" && dateCol == "download_activity_date") { message("Using dayid instead of dateCol for fact_analytics") # quickEmail(subject="Using dayid instead of dateCol for fact_analytics", getRS()) dateCol <- "dayid" } force(dateCol) force(colsToPull) if (!is.null(maxDate) || !is.null(processed_lessthan)) message("Make sure you are using maxDate and processed_lessthan only for debugging purposes -- otherwise, whats the point") ## Check that we user can read from each cluster if (is.null(names(cluster))) names(cluster) <- sprintf("Cluster_%02i", cluster) canread <- unlist(sapply(cluster, function(cl) qCanRead(tbl=tbl, schema=schema, cluster=cl)), use.names=TRUE) if (any(!canread)) { stop (warningCols("user cannot read for the following: ", nwhich(!canread))) } suffixes <- c(".rs", ".sf") ## Wrapper function to append suffixes psfx1 <- function(string) paste0(string, suffixes[[1]]) psfx2 <- function(string) paste0(string, suffixes[[2]]) whereIn <- if (!is.null(processed_lessthan)) sprintf("processeddaytime < '%s'", processed_lessthan) colsWithaggFunc <- if (!is.null(processed_lessthan)) c(max_processeddaytime = "MAX(processeddaytime)") ## Clean up colsToAgg after the query nms_of_colsToAgg <- colNamesFromVector(colsToAgg) ## Add a rowCount to colsToAgg if (rowcount_addedto_colsToAgg && "rows" %ni% names(colsToAgg) && ".rowcount" %ni% colsToAgg) { colsToAgg %<>% c(rows=".rowcount", .) nms_of_colsToAgg %<>% c("rows", .) } ## Create Query Qry <- makeQry(tbl=tbl, schema=schema, colsToAgg=colsToAgg, colsToPull=colsToPull, colsWithaggFunc=colsWithaggFunc, dateCol=dateCol, maxDate=maxDate, whereIn=whereIn, limit=NULL, key="colsToPull") ## Get Redshift Data DT.redshift <- runQry(Qry, cluster=cluster, snowflake_inuse = FALSE, connex=giveMeACon(refresh=refresh_connex)) # assign("DT.redshift", DT.redshift, envir=globalenv()) suppressWarnings(rm(con, connex, envir=globalenv())) suppressWarnings(rm(con, connex, envir=getOption("con.environ", globalenv()))) message("TODO: Switch from redshift to snowflake involves unloading package") setSnowflake(wh=wh, dbname=dbname, start=TRUE, verbose=TRUE) DT.snowflake <- runQry(Qry, warehouse=wh, snowflake_inuse = TRUE, connex=sfGetCon(refresh=refresh_connex)) # assign("DT.snowflake", DT.snowflake, envir=globalenv()) ## ------------------------------------------- ## ## COMPARE ## Merge, and organize the columns for fun matchKey(DT.snowflake, DT.redshift, key=colNamesFromVector(colsToPull)) DT.compare <- merge(DT.redshift, DT.snowflake, suffix=suffixes, by=colNamesFromVector(colsToPull), all=TRUE) setcolorderpt(DT.compare, endCols=zipperCombine(psfx1(nms_of_colsToAgg), psfx2(nms_of_colsToAgg))) ## Check that there are no unmatched rows col.row1 <- psfx1("rows") col.row2 <- psfx2("rows") DT.compare[, psfx1("missing_from") := is.na(get(col.row1)) & !is.na(get(col.row2))] DT.compare[, psfx2("missing_from") := is.na(get(col.row2)) & !is.na(get(col.row1))] if (any(DT.compare[, get(psfx1("missing_from")) | get(psfx2("missing_from"))])) { err.msg.unmatched <- sprintf("\nThere are %5i group(s) of rows %s and \nThere are %5i group(s) of rows %s" , DT.compare[get(psfx1("missing_from")), .N] , psfx1("missing_from") , DT.compare[get(psfx2("missing_from")), .N] , psfx2("missing_from") ) warning (err.msg.unmatched) } ## Check each pair of columns same_suffix <- ".SAME" for (col in nms_of_colsToAgg) { DT.compare[, paste0(col, same_suffix) := equals0(.SD[[1]] - .SD[[2]], na.check=TRUE) | (is.na(.SD[[1]]) & is.na(.SD[[2]])), .SDcols=paste0(col, suffixes)] } ## Check which row is good across all columns colsSameCheck <- nms_of_colsToAgg %>% {setNames(nm=., obj=paste0(., same_suffix))} DT.compare[, all_row_values_are_equal := rowSums(.SD, na.rm=TRUE) == length(nms_of_colsToAgg), .SDcols=colsSameCheck] ## A ROW CAN BE NOT-GOOD IN THREE WAYS ## 1. Missing from SF ## 2. Missing from RS ## 3. Different Values # DT.err.missing_from_sf <- DT.compare[(missing_from.sf)] # DT.err.missing_from_rs <- DT.compare[(missing_from.rs)] # DT.err.different_values <- DT.compare[!all_row_values_are_equal & !missing_from.sf & !missing_from.rs] DT.err.different_values <- DT.compare[(!all_row_values_are_equal)] if (nrow(DT.err.different_values)) { ## OLD: ~~ if (nrow(DT.compare[!(all_row_values_are_equal)])) { ## OLD: ~~ # dates_off <- DT.compare[, c(list(number_of_rows_off = sum(!all_row_values_are_equal), out_of_total_rows=.N, perc_of_rows_incorrect=percFalse(all_row_values_are_equal)), lapply(.SD, sumn)), keyby=dateCol, .SDcols=c("all_row_values_are_equal", unlist(lapply(suffixes, function(sfx) sprintf("%s%s", nms_of_colsToAgg, sfx))))] ## OLD: ~~ dates_off <- DT.compare[, list(number_of_rows_off = sum(!all_row_values_are_equal), out_of_total_rows=.N, perc_of_rows_incorrect=fwp(percFalse(all_row_values_are_equal))), keyby=dateCol][number_of_rows_off > 0] ## OLD: ~~ DT.compare[get(dateCol) == dates_off[[c(1, 1)]]] ## OLD: ~~ ## OLD: ~~ err.msg <- DT.compare[!(all_row_values_are_equal), sprintf("For table '%s':\n There are %i rows (out of %i) that are not good, across %i different dates\n mindate is %s and maxdate is %s.\n Also, there are %i different stores and %i different transaction types", tbl, .N, nrow(DT.compare), lunique(get(dateCol)), min(get(dateCol)), max(get(dateCol)), lunique(storeid), lunique(transactiontypeid))] ## identify What's wrong: DT.err.different_values[, whats_wrong := "UNKOWN?? "] DT.err.different_values[, whats_wrong := {nwhich(sapply(unique(.SD, by=NULL), "!")) %>% gsub(same_suffix, "" ,.) %>% pasteQand(q="") %>% paste("is different from RS <~~> SF")}, .SDcols=colsSameCheck, by=colsSameCheck] DT.err.different_values[(missing_from.sf), whats_wrong := "Data did not make it to Snowflake"] DT.err.different_values[(missing_from.rs), whats_wrong := "Superfluous data in Snowflake (missing from Redshift)"] err.msg <- DT.err.different_values[, .N, keyby=whats_wrong][, sprintf("%i row-group%s where %s", N, ifelse(N==1, "", "s"), gsub("\brows\b", "row count", whats_wrong))] %>% pasteC(C="\n\t") %>% sprintf("For table '%s':\n\t%s", tbl, .) warning(err.msg) setattr(DT.err.different_values, "err.msg", err.msg) return(invisible(DT.err.different_values)) } if ("rows" %in% names(DT.redshift) && "rows" %in% DT.snowflake) { rowcount.rs <- DT.redshift[["rows"]] %>% sumn rowcount.sf <- DT.snowflake[["rows"]] %>% sumn } else { rowcount.rs <- nrow(DT.redshift) rowcount.sf <- nrow(DT.snowflake) } ## ELSE if (rowcount.rs != rowcount.sf) warning ("The row counts are off") ## Message that all is well if (!(rowcount.rs != rowcount.sf || nrow(DT.compare[!(all_row_values_are_equal)]))) message (sprintf("Snowflake and RedShift match for table '%s' in schema '%s'\n when checked by %s along %s", tbl, schema, pasteQand(colsToPull), nms_of_colsToAgg %>% gsub("rows", "row count", .) %>% pasteQand(q=""))) return(invisible(DT.compare)) }