# validation failed.r ## --------------------------------------------------------------- ## ## validation 2 failed. ## compare the fact table (tbl.fact) with the bi table (tbl.final) ## --------------------------------------------------------------- ## setScience("DeNormalizing", create=FALSE, subl=FALSE, load=FALSE) source("~/git/orch/src/MgmtReport/supportFns/colsByTable.r") loadIfNotExists("DT.rowcounts.fact", verbose=TRUE) loadIfNotExists("DT.rowcounts.aggd", verbose=TRUE) loadIfNotExists("minDate", verbose=TRUE) loadIfNotExists("maxDate", verbose=TRUE) loadIfNotExists("DT.rowcounts", verbose=TRUE) loadIfNotExists("run_start_time", verbose=TRUE) assignIfNotExist(cluster.out, getOption("db.defaultcluster.out")) tbl <- assignIfNotExist(tbl, "analytics" , envir=environment(), verbose=TRUE) tbl.fact <- assignIfNotExist(tbl.fact, "fact_analytics" , envir=environment(), verbose=TRUE) tbl.final <- assignIfNotExist(tbl.final, "analytics" , envir=environment(), verbose=TRUE) schema <- assignIfNotExist(schema, "bi" , envir=environment(), verbose=TRUE) minDate <- assignIfNotExist(minDate, "2013-01-01" , envir=environment(), verbose=TRUE) maxDate <- assignIfNotExist(maxDate, qMaxDate(tbl, verbose=FALSE, cluster=cluster.out) - 1, envir=environment(), verbose=TRUE) ## kCols.validate <- assignIfNotExist(kCols.validate, c("date", "storeid", "transac_typeid"), envir=environment(), verbose=TRUE) colsToAgg <- c("units") desc("DT.rowcounts", tight=TRUE) message("Using run_start_time = ", run_start_time) ## Either manually set the dateChecking or auto set it { dateChecking <- as.Date("2014-09-01") if (nrow(DT.rowcounts[!(is_ok)]) == 1) dateChecking <- DT.rowcounts[!(is_ok), date] else dateChecking <- DT.rowcounts[!(is_ok), tail(head(date, -1), 1)] } #### -------------- colsToCompare <- copy(colsToCompare.ll[[tbl]]) colsToCompare <- colsToCompare[colsToCompare != ".rowcount"] colsToCompare <- colNamesFromVector(colsToCompare) colsToCompare.ratio <- paste0(colsToCompare, ".ratio") selfname_(colsToCompare.ratio) #### -------------- ## CREATE THE QUERIES qry.validate.aggd <- makeQry(tbl=tbl.final, schema=schema, colsToPull=c(date = dateCol.ll[[tbl]], "storeid", "transac_typeid"), colsToAgg=colsToCompare.ll[[tbl]][colsToCompare.ll[[tbl]] != ".rowcount"], dateCol=dateCol.ll[[tbl]], minDate=minDate, maxDate=maxDate) qry.validate.fact <- makeQry(tbl=tbl.fact, schema="production", colsToPull=c(date = dateCol.ll[[tbl.fact]], "storeid", "transac_typeid"="transactiontypeid", "processed_last_2days"=sprintf("processeddaytime > '%s'", today()-2) ), colsToAgg=colsToCompare.ll[[tbl.fact]][colsToCompare.ll[[tbl.fact]] != ".rowcount"], dateCol=dateCol.ll[[tbl.fact]], minDate="2013-01-01", maxDate=maxDate, where = if (tbl == "analytics") sprintf("(processeddaytime < '%s' OR processeddaytime IS NULL)", run_start_time)) ## EXECUTE THE QUERIES ## ---------------------- ## bi { DT.validate.aggd <- runQry(qry.validate.aggd, cluster=cluster.out, verbose=FALSE) ## Clean up any NAs in transac_typeid & units (these were introduced from the errors table in the SQL join) DT.validate.aggd[is.na(transac_typeid), transac_typeid := 0L] DT.validate.aggd[is.na(units), units := 0L] } ## fact { DT.validate.fact_noErrs <- runQry(qry.validate.fact, cluster=cluster.in) if (tbl.fact == "fact_analytics") { DT.validate.errs <- runQry(convertQry.fa_to_faerrors(qry.validate.fact)) DT.validate.fact <- rbind(DT.validate.fact_noErrs, DT.validate.errs)[, lapply(.SD, sumn), .SDcols=colsToCompare, keyby=kCols.validate] } else { ## This step isn't necessary, but it is useful to have separate the two DTs (when using fact_analytics) for debugging purposes DT.validate.fact <- copy(DT.validate.fact_noErrs) } ## If the DT used periodi, change it to "date" (for now, applies only to fact_sales) addDateCols.periodid_(DT=DT.validate.fact, newCol.nms="date", dropPeriodCols=TRUE, showWarnings=FALSE) ## Clean up transactiontype 26 ~~> 4 DT.validate.fact[storeid == 1 & transac_typeid == 26, transac_typeid := 4] } ## CONFIRM ALL HAVE ROWS { if (!nrow(DT.validate.fact)) stop ("DT.validate.fact has no rows") if (!nrow(DT.validate.aggd)) stop ("DT.validate.aggd has no rows") } ## MERGE THE TWO ## ------------------- suffix <- c(".aggd", ".fact") matchKey(DT.validate.fact, DT.validate.aggd, kCols.validate) DT.validate <- merge(DT.validate.aggd, DT.validate.fact, suffix=suffix, all=TRUE) setInfo(DT.validate, "Created as the merge of the Qry-pulls from tbl.fact and tbl.final. 'is_ok' is FALSE if any pairwise columns fail to be the same (between fact and final") ## FIND THE ERRORS. ## ------------------- ## If any col to compare is not the same, it is an error DT.validate[, is_ok := TRUE] for (col in colsToCompare) DT.validate[, is_ok := is_ok & equals0(get(paste0(col, suffix[[1]])) - get(paste0(col, suffix[[2]]))) ] DT_ERROR.failedValidation <- DT.validate[!(is_ok)] setInfo(DT_ERROR.failedValidation, "Created from DT.validate[!(is_ok)]. DT.validate was the merge of the Qry-pulls from tbl.fact and tbl.final. 'is_ok' is FALSE if any pairwise columns fail to be the same (between fact and final") sortColsIgnoreSuffix_ <- function(DT, suffixes, startCols=key(DT), sortAllColsAlphabetically.exceptStartCols=FALSE) { if (!is.null(startCols) && any(startCols %ni% names(DT))) stop("Some cols in startCols are not in DT: ", pasteQand(setdiff(names(DT), startCols))) nms <- names(DT) selfname_(nms) patSuffix <- function(suffix) paste0(escapeRegEx(suffix), "$") for (suffix in suffixes) nms <- gsub(patSuffix(suffix), "", nms) if (sortAllColsAlphabetically.exceptStartCols) { nms.ret <- c(startCols, sort(nms)[sort(nms) %ni% startCols]) ## Otherwise, sort only the duplicated columns, according to when they first appeared } else { for (nm in nms[duplicated(nms)]) { wh <- which(nms == nm) if (length(wh)) { } nms[-wh] } &&& LEFT OFF HERE } } ## SHOW DT.validate[date == dateChecking] DT_ERROR.failedValidation[date == dateChecking] ## Pick a date DT_ERROR.failedValidation[date < today() - 60][date == max(date)] DT.validate.fact_noErrs matchKey(DT.validate.aggd, DT.validate.fact, "date") DT.validate <- merge(DT.validate.aggd, DT.validate.fact, suffix=c(".aggd", ".fact"), all=TRUE) for (col in colsToCompare) { DT.validate[, paste0(col, ".ratio") := get(paste0(col, ".aggd")) / get(paste0(col, ".fact"))] DT.validate[get(paste0(col, ".aggd")) == 0 & get(paste0(col, ".fact")) == 0, paste0(col, ".ratio") := 1] } "&&&& 2014-12-15:: I am not sure that 'sumn' is the way to go here. It should me 'unique' or maybe even 'meann' 2014-01-07:: Actually, for now at least, they are all identical" # compare # DT.validate.sumn <- DT.validate[, lapply(.SD, sumn), keyby=date, .SDcols=colsToCompare.ratio ] # compare # DT.validate.meann <- DT.validate[, lapply(.SD, meann), keyby=date, .SDcols=colsToCompare.ratio ] # compare # DT.validate.unique <- DT.validate[, lapply(.SD, unique), keyby=date, .SDcols=colsToCompare.ratio ] # compare # compareDTs(DT.validate.sumn, DT.validate.unique) # compare # compareDTs(DT.validate.sumn, DT.validate.meann) DT.validate <- DT.validate[, lapply(.SD, sumn), keyby=date, .SDcols=colsToCompare.ratio ]