# -------------------------------------------------------------------------------------------------------------------------  #
  #  -----------------------------------------------------------------------------------------------------------------------   #
  #                                                                                                                            #
  #           File Name              :  automate new.r                                                                         #
  #           Last Updated Funclist  :  12 Feb 2015,  2:12 PM (Thursday)                                                       #
  #                                                                                                                            #
  #           Author Name            :  Rick Saporta                                                                           #
  #           Author Email           :  RSaporta@TheOrchard.com                                                                #
  #           Author URL             :  www.github.com/rsaporta                                                                #
  #                                                                                                                            #
  #           Packages Called        :  RPostgreSQL                                                                            #
  #           Packages Used via NS   :  lubridate                                                                              #
  #                                                                                                                            #
  #  -----------------------------------------------------------------------------------------------------------------------   #
  #                                                                                                                            #
  #   changeCluster      ( cluster, msg=NULL, tbl=NULL, verbose=TRUE )                                                         #
  #   SelectAndUnload    ( QRY.select, tbl, bucket, cluster, delim, manifest, gzip, verbose=TRUE, verboseQry=FALSE )           #
  #   createTable        ( QRY.create, tbl.final, tbl.tmp, schema, cluster, dontTruncate, verbose=TRUE, verboseQry=FALSE )     #
  #   reload             ( cluster                                                                                             #
  #                        , tbl.tmp, tbl.final, bucket, delim, manifest, schema, allow_0_rows=FALSE, verbose, verboseLoad )   #
  #   updateDeNormTable  ( tbl="analytics", schema="bi", delim="\v", manifest=FALSE, gzip=TRUE                                #
  #                        , cluster.in=getOption("db.defaultcluster.in")                                                      #
  #                        , cluster.out=getOption("db.defaultcluster.out"), folder=src.p("BI_tables")                         #
  #                        , projName="DeNormalizing", add_aggregated=FALSE, dontRename=FALSE, dontTruncate=FALSE              #
  #                        , skip.select=FALSE, skip.errorsIngest=skip.select, skip_to_part=0L                                 #
  #                        , fail_if_validate_fails=FALSE, stamp=dateStamp(), transacs.to.ignore=33:36, verbose=TRUE )         #
  #   gsubQry            ( qry, tbl.new, tbl.old, schema=NULL, schema.new=schema, schema.old=schema.new )                      #
  #   gsubQryDefaults    ( qry )                                                                                               #
  #                                                                                                                            #
  #                                                                                                                            #
  #                                                     <END FUNCS>                                                            #
  #  -----------------------------------------------------------------------------------------------------------------------   #
  # -------------------------------------------------------------------------------------------------------------------------  #

## Note this file is called by  ~/git/orch/src/MGMT_REPORT_FULL_AUTO.r
## and thus is in the supportFns folder for MGMTreport
## However, it really belongs to the DeNormalizing Project


changeCluster <- function(cluster, msg=NULL, tbl=NULL, verbose=TRUE, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema="bi", snowflake_inuse=getOption("snowflake_inuse", default=FALSE), verbose.showSettings=verbose) {
    ## SNOWFLAKE
    if (snowflake_inuse)
      return(setSnowflake(wh=wh, dbname=dbname, schema=schema))

    dbDisconnectAll()
    setDBall(cluster=cluster)
    if (verbose) {
      if (!is.null(tbl) && !is.null(msg))
        msg <- paste(msg, toupper(tbl))
      verboseMsg(!is.null(msg), sprintf("  -----------{  %30s  }-----------  ", msg), func="message")
      if (verbose.showSettings) {
        showDBsettings()
        cat("\n")
      }
    }
    return(invisible(cluster))
}

SelectAndUnload <- function(QRY.select, tbl, bucket, cluster, delim, manifest, gzip, check_bucket_exists=TRUE, verbose=TRUE, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeSchema(), snowflake_inuse=getOption("snowflake_inuse", default=FALSE), verboseQry=FALSE, verbose.showSettings=verbose, notify=TRUE) {
## tbl :: used only for notifications and messages

    ## The bucket where the data is getting dumped should not already exist
    if (s3_bucketExists(bucket))
      warning ("The destination bucket for unloading the data appears to already exist. This might cause issues for AWS\nThe bucket given is:  '", bucket, "'")

    changeCluster(cluster=cluster, msg="STARTING UNLOAD", tbl=tbl, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse, verbose=verbose, verbose.showSettings=verbose.showSettings)
    qUnload <- unloadQry(QRY.select, bucket=bucket, delimiter=delim, manifest=manifest, gzip=gzip, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse, verbose=verboseQry)

    ## CONFIRM - the data should be in the bucket
    if (isTRUE(check_bucket_exists) && !s3_bucketExists(bucket))
      stop ("\nA sefety check failed, namely, the bucket was NOT created.\nSelectAndUnload(tbl='", tbl, "') failed to create the bucket '", bucket, "'\n\nHINT 1: This probably means that unloadQry() failed, in othwer words the SELECT query failed.\n        Check the Logs.\n\nHINT 2: If you changed the SQL for creating analytics, did you remember to change the 'errors' SQL as well?\n        Did you update the views?")

    if (isTRUE(notify))
      notifyAndEmail(sprintf("%s - Done Selecting & Unload", toupper(tbl)))
    else
      return(invisible(NULL))
}

### -------- 01 CREATE / CLEAR TABLE TO INGEST -------- ###
createTable <- function(QRY.create, tbl.final, tbl.tmp, cluster, dontTruncate, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema, snowflake_inuse=getOption("snowflake_inuse", default=FALSE), verbose=TRUE, verboseQry=FALSE) {
  changeCluster(cluster=cluster, msg="CREATING TABLE", tbl=tbl, verbose=verbose)
  if (!qTableExists(tbl.tmp, schema=schema, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse))
      runQry(QRY.create, verbose=verboseQry, cluster=cluster, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse)
  else if (!dontTruncate)
      runQry(sprintf("TRUNCATE TABLE %s", schemaPaste(schema, tbl=tbl.tmp)), cluster=cluster, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse, verbose=TRUE)
  else 
      stop ("Table ", dbschematbl(dbname=dbname, schema=schema, tbl=tbl.tmp), " exists and dontTruncate is set to TRUE.\n  Cannot proceed")
}


reload <- function(cluster, tbl.tmp, tbl.final, bucket, delim, manifest, allow_0_rows=FALSE, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema, snowflake_inuse=getOption("snowflake_inuse", default=FALSE), verbose, verboseLoad)
{
  changeCluster(cluster=cluster, msg="STARTING   LOAD", tbl=tbl.final, verbose=verbose)
  qLoad   <- loadFromBucket(bucket=bucket, tbl=tbl.tmp, schema=schema
                , delimiter=delim, confirm_dates=TRUE
                , create.if.not.exist=FALSE, truncate.if.exist=FALSE
                , allow_0_rows=allow_0_rows
                , verbose=verboseLoad, manifest=manifest)


  notifyAndEmail(sprintf("%s - Done Loading & Ingesting", toupper(tbl.tmp)))

  ## CONFIRM TABLE EXISTS AND HAS ROWS
  if (!runQry(sprintf("SELECT count(*) AS rows FROM %s", schemaPaste(schema, tbl=tbl.tmp)), cluster=cluster)[1, rows > 0])
     stop ("verify rows failed for ", schemaPaste(schema, tbl=tbl.tmp))
}


updateDeNormTable <- function(tbl="analytics"
                            , UNLOAD_ONLY=TRUE
                            , delim='\v'
                            , manifest=FALSE
                            , gzip=TRUE
                            , cluster.in=getOption("db.defaultcluster.in")
                            , cluster.out=getOption("db.defaultcluster.out")
                            , folder=src.p("BI_tables")
                            , projName="DeNormalizing" # for science
                            , add_aggregated = FALSE
                            , dontRename = FALSE
                            , dontTruncate = FALSE
                            , skip.select = FALSE  ## Usually for debugging or re-running after fail point
                            , skip.errorsIngest = skip.select  ## Usually for debugging or re-running after fail point
                            , skip_to_part = 0L
                            , fail_if_validate_fails=FALSE
                            , stamp=dateStamp()
                            , transacs.to.ignore = 33:36
                            , wh=getSnowflakeWH()
                            , dbname=getSnowflakeDB()
                            , schema="bi"
                            , snowflake_inuse=getOption("snowflake_inuse", default=FALSE)
                            , verbose=TRUE
                            ) {

## ARGS: 
## dontTruncate : if table exists && isTRUE(dontTruncate) then will NOT truncate and will fail
## ----
  library(RPostgreSQL)

  ### ------ Dependent Functions ----------- ###
  gsubQry <- function(qry, tbl.new, tbl.old, schema=NULL, schema.new=schema, schema.old=schema.new) {
    pat  <- schemaPaste(schema.old, tbl.old)
    repl <- schemaPaste(schema.new, tbl.new)
    gsub(pat, repl, qry)
  }

  gsubQryDefaults <- function(qry) {
    qry <- gsubQry(qry, tbl.new=tbl.tmp, tbl.old=paste0("aggregated_", tbl.final),  schema.new=schema, schema.old="production")
    qry <- gsubQry(qry, tbl.new=tbl.tmp, tbl.old=paste0("aggregated_", tbl.final),  schema.new=schema, schema.old="bi.")
    qry <- gsubQry(qry, tbl.new=tbl.tmp, tbl.old=tbl.final,  schema.new=schema, schema.old="production")
    qry <- gsubQry(qry, tbl.new=tbl.tmp, tbl.old=tbl.final,  schema.new=schema, schema.old=schema)
    return(qry)
  }
  ### ------ Dependent Functions ----------- ###

  if (is.numeric(skip_to_part) && skip_to_part < 2)
    message("\n------------------------\n", "USING CLUSTER.IN = ", cluster.in, "\n------------------------\n")

  if (is.null(skip_to_part))
    skip_to_part <- 0L
  if (tolower(substr(skip_to_part, 1, 6))=="ingest")
    skip_to_part <- 3L
  if (isTRUE(skip.select))
    skip_to_part <- max(2, skip_to_part)
  if (skip_to_part > 7)
    stop ("skip_to_part must be an integer between 1 and 7")


  ## If the porject is set incorrectly, adjust it
  if (!identical(getProjName(), projName) && missing(folder)) {
    # stop ("Call setScience() first or set folder explicitly ")
    setScience(projName, subProj=paste0(tbl, "_autoload"), create=FALSE, load=FALSE, subl=fact_sales)
    folder <- src.p("BI_tables")
  }

  if (!file.exists(folder)) {
    stop ("folder '", folder, "' does not exit")
  }

  ## TODO:  Have better logic for when to run this
  if (!qShowViews(justnames=TRUE, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse) %cont% "bi.fa_errors_view" && tbl == "analytics") {
    stop("It looks like bi.fa_errors_view is missing\nWhile we could run CreateAllViews(), it is most likely that it was already ran and had encountered a previous error.\nHINT: Check the log files")
    source("~rsaporta/git/orch/src/DeNormalizing/BI_tables/Create All Views.r")
  }


  ## which tables we can create
  tblsAvail <- setdiff(list.dirs(path=folder, full.names=FALSE), c("zArchived", ""))

  ## confirm tbl name and create tmp tables
  {
    tbl.final <- extractMatch(tbl, tolower(tblsAvail))

    ## FROM OLDER VERSIONS: REMOVE "aggregated_" PREFACE
    if (add_aggregated && !grepl("^aggregated_", tbl.final))
      tbl.final <- paste0("aggregated_", tbl.final)

    tbl.tmp    <- sprintf("%s_new_%s",     tbl.final, stamp)
    tbl.zArch  <- sprintf("%s_removed_%s", tbl.final, stamp)

    if (is.na(tbl) || !length(tbl) || is.na(tbl.final) || !length(tbl.final))
      stop ("Invalid tbl value -- crashing at top of updateDeNormTable()")
  }

  folder.using <- extractMatch(tbl, tblsAvail, ignore.case=TRUE)
  if (length(folder.using) != 1)
    stop ("tbl matches zero or more than one folder in tblsAvail -- should this be allowed? (why not?)")

  files.using <- extractFilesFromFolder(as.path(folder, folder.using), ext="sql", full=TRUE)

  ## Read in the SQL files
  QRY.create <- sqlFileToQry( extractMatch("create",   files.using, use.names=TRUE) )
  QRY.select <- sqlFileToQry( extractMatch("populate", files.using, use.names=TRUE) )

  ## TODO:  Split QRY.03 into pieces
  QRY.update <- sqlFileToQry( extractMatch("update",   files.using, use.names=TRUE) )

  tbl.fact <- extractMatch(gsub("aggregated_", "", tbl), c(analytics="fact_analytics", accounting="fact_sales"), use.names.of.table=TRUE, ignore.case=TRUE)
  col.maxdate.fact <- extractMatch(gsub("aggregated_", "", tbl), c(analytics="processeddaytime", accounting="dateCreated"), use.names.of.table=TRUE, ignore.case=TRUE)
  col.maxdate.fact <- unname(col.maxdate.fact)

  if (length(tbl.fact) != 1)
    stop ("tbl matches zero or more than one fact_ table -- should this be allowed? (why not?)")


  # lastProcessedDate <- runQry("SELECT max(processeddaytime) FROM production.fact_analytics", cluster=cluster.in, verbose=FALSE)
  # lastProcessedDate <- runQry("SELECT max(datecreated) AS datecreated, max(accountingperiodid) AS accountingperiodid FROM production.fact_sales", cluster=cluster.in, verbose=FALSE)
  lastProcessedDate <- makeQry(tbl=unname(tbl.fact), colsToPull=NULL, colsToAgg=c(maxdateprocessed=col.maxdate.fact), aggFunc="max", schema="production") %>% 
                          runQry(cluster=cluster.in, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse, verbose=FALSE)
  lastProcessedDate <- lastProcessedDate[["maxdateprocessed"]]
  lastProcessedDate <- as.Date(lastProcessedDate) ## as.Date() might be the wrong function to use for times near-midnight

  ### TODO: Check against lastProcessed for agg table


  suppressWarnings(rm(qUnload, qLoad))

  ## Where files will be placed
  bucket <- s3_p("BI_Tables", tbl, stamp, prefix=tbl)
  cat (" ---- bucket ::   '",bucket,"'\n", sep="")

  ## change names in queries
  QRY.create <- gsubQryDefaults(QRY.create)
  QRY.update <- gsubQryDefaults(QRY.update)
  ## no need for QRY.select

  ## Capture the time that the ran start. This will be used for filtering in the validation step
  run_start_time <- now(tz="UST")
  message("Using   run_start_time = '", run_start_time, "'")

  browser(expr=inDebugMode(c("automate", "update", "unload")), text="in updateDeNormTable() before unload or load")

  ### -------- PART 1 SELECT & UNLOAD (02.SQL) -------- ###
  if (skip_to_part <= 1)
    SelectAndUnload(QRY.select=QRY.select, tbl=tbl, bucket=bucket, delim=delim, manifest=manifest, gzip=gzip, cluster=cluster.in, wh=wh, dbname=dbname, snowflake_inuse=snowflake_inuse, verbose=all(!identical(cluster.in, cluster.out)))

  if (UNLOAD_ONLY) {
    message("Only Unloading. Bucket is\n  ", bucket, "\n")
    return(bucket)
  }

  ### -------- PART 2 CREATE / TRUNCATE TABLE TO INGEST (01.SQL) -------- ###
  if (skip_to_part <= 2)
    createTable(QRY.create=QRY.create, tbl.final=tbl.final, tbl.tmp=tbl.tmp, schema=schema, cluster=cluster.out, dontTruncate=dontTruncate, verbose=all(!identical(cluster.in, cluster.out)))


  ### -------- PART INTERMEDIATE STEP: Free up Disk Space -------- ###
  ## Drop any archived table named "tbl_removed_*", to free up disk space
  ## TODO:  Every two weeks, take a snapshot
  {
    ## Drop All tabes named Tbl_Removed_*  -- including the most recent Tables
    existing_tables <- qShowTables(namelike=sprintf("%s_removed_%%", tbl.final), schema=schema, justnames=TRUE, rowCount=FALSE, showWarnings=FALSE, wh=wh, dbname=dbname)
    if (length(existing_tables) > 1) {
      verboseMsg(verbose, sprintf("Dropping %i archived table%s:\n    %s", length(existing_tables),  ifelse(length(existing_tables) > 1, "s", ""), pasteQ(existing_tables, w="")))
      runQry(pasteC(sprintf("DROP TABLE %s;\n", sapply(existing_tables, schemaPaste, schema=schema))), verbose=TRUE)
    }
  }


  ### -------- PART 3 INGEST DATA INTO TABLE (loading from bucket) -------- ###
  if (skip_to_part <= 3)
    reload(cluster=cluster.out, tbl.tmp=tbl.tmp, tbl.final=tbl.final, bucket=bucket, delim=delim, manifest=manifest, schema=schema, allow_0_rows=FALSE, verbose=TRUE, verboseLoad=TRUE)

  ### -------- SAFETY CHECK: VERIFY / VALIDATE 1 of 2   -------- ###
  ## SO FAR ONLY WORKS FOR ANALYTICS
  validation.success <- FALSE
  if (tbl.fact == "fact_analytics")
  try({
    ## Verify value
    minMonthToCheck   <- lubridate::floor_date(Sys.Date() - 150, unit="month")
    # maxDate           <- as.Date(min(lubridate::floor_date(Sys.Date(), unit="month") + 15, Sys.Date() - 6))
    maxDate           <- lastProcessedDate

    # minDayIdToCheck   <- dateTodayid(minMonthToCheck)
    # maxDayIdToCheck   <- dateTodayid(maxDate) ## Useful when debugging, or if checking a table created a few days ago

    storeids.to.check <- c(1, 7, 187, 376, 487, 497)
    colsToAgg         <- c("units")

    kCols.validate    <- c("date", "storeid", "transac_typeid")
  
    q.validate.fact <- makeQry(tbl=tbl.fact, schema="production"
                            , colsToPull=c(date=sql1stOfMonth("download_activity_date"), "storeid", transac_typeid="transactiontypeid")
                            , colsToAgg=colsToAgg, minDate=minMonthToCheck, maxDate=maxDate
                            , dateCol="download_activity_date"
                            # OLD , whereIn=sprintf("storeid in (%s) and processeddaytime < '%s'", commaSep(storeids.to.check), lastProcessedDate+1))
                            , whereIn=sprintf("storeid IN (%s) AND (processeddaytime < '%s' OR processeddaytime IS NULL AND NOT(transactiontypeid IN (%s)))", commaSep(storeids.to.check),   run_start_time, pasteC(transacs.to.ignore, C=",")))
    DT.validate.fact_noErrs <- runQry(q.validate.fact, cluster=cluster.in, verbose=FALSE)

    ## Pull and aggregate the data from the errors table as well, then rbind it to DT.validate.fact, aggregating again.
    if (tbl.fact == "fact_analytics") {
      q.validate.errors  <- convertQry.fa_to_faerrors(q.validate.fact)
      DT.validate.errors <- runQry(q.validate.errors, cluster=cluster.in, verbose=FALSE)
      DT.validate.fact   <- rbind(DT.validate.fact_noErrs, DT.validate.errors)[, lapply(.SD, sumn), keyby=kCols.validate, .SDcols=colsToAgg]
    } else {
      DT.validate.fact <- DT.validate.fact_noErrs
    }

    ## Clean up transactiontype 26 ~~> 4
    DT.validate.fact[storeid == 1 & transac_typeid == 26, transac_typeid := 4]

    ## Pull and aggregate the new table
    q.validate.aggd <- makeQry(tbl=schemaPaste(schema, tbl.tmp), colsToPull=c(date=sql1stOfMonth("activity_date"), "storeid", transac_typeid="transac_typeid")
                      , colsToAgg=colsToAgg, minDate=minMonthToCheck, maxDate=maxDate, dateCol="activity_date", whereIn=list(storeid=storeids.to.check))
    DT.validate.aggd <- runQry(q.validate.aggd, cluster=cluster.out, verbose=FALSE)
    ## Clean up any NAs that might have been introduced from including the errors table
    DT.validate.aggd[is.na(transac_typeid), transac_typeid := 0L]
    DT.validate.aggd[is.na(units), units := 0L]

    if (!nrow(DT.validate.fact))
      stop ("DT.validate.fact has no rows")
    if (!nrow(DT.validate.aggd))
      stop ("DT.validate.aggd has no rows")

    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)
    DT.validate[, paste0(colsToAgg, suffix) := lapply(.SD, removeNA, repl=0), .SDcols=paste0(colsToAgg, suffix)]
    DT.validate[, aggd_perc_of_fact := (units.aggd - units.fact) / units.fact]
    DT.validate[equals0(aggd_perc_of_fact, tol=.001), aggd_perc_of_fact := 0]
    failedValidation <- DT.validate[(0!=aggd_perc_of_fact)]

    ## VALIDATION FAILURE
    if (nrow(failedValidation))  {
      notifyIfError(expr=all(failedValidation[, date] == lubridate::floor_date(Sys.Date(), unit="month") )
                   , subj=paste0(tbl, " - Validation pt 1 failed.")
                   , msg=paste0("validation failed for tbl='", tbl, "'.\nsee object  failedValidation\n(stopped before running '03 Updates')")
                   , fail=TRUE)
      ## ELSE
      warning ("See object failedValidation -- all failures are for current month (meaning, might be a day behind, or might be missing several days)")
    } else {
      message("validation successfull -- renaming DB")
    }
    ## If this process (Wrapped in TRY to avoid failing the whole function) reaches this point, then assign TRUE to validation.success -- no need to return anything to outside of try()
    validation.success <- TRUE
  ## ELSE, as in, tbl.fact is NOT fact_analytics, assume validation to be succesful
  }) else {
    validation.success <- TRUE
  }
  ## --- end verify / validate (for fact_analytics only, for now) --- ##


  ### -------- PART 4 UPDATES (03.SQL) -------- ###
  if (skip_to_part <= 4) {
    changeCluster(cluster=cluster.out, msg="RUNNING UPDATES ON", tbl=tbl, verbose=all(!identical(cluster.in, cluster.out)))
    runQry(QRY.update, cluster=cluster.out, proceed.past.erros=TRUE, verbose=TRUE)
    notifyAndEmail(sprintf("%s - Done Updating", toupper(tbl)))

    verboseMsg(verbose, "\n\n", pasteR("-~", n=50), "\n  * Done Running SQL files for '", tbl, "'. *\n  * What remains is verifying, then archiving the existing table,  then renaming the new one. *\n", pasteR("-~", n=50), "\n", sep="")
  } ## // END part 4


  ### -------- PART 5 RENAME TABLES -------- ###
  if (skip_to_part <= 5) {
    if (!dontRename) {
      if (tbl != tbl.tmp && validation.success) {
          ## check if tbl.final already exists, and if so, archive it
          if (qTableExists(schema=schema, tbl=tbl.final) == 1)
                 qRenameTable(tbl.old=tbl.final, tbl.new=tbl.zArch, schema=schema, verbose=TRUE, msg.pre="[Archiving previous table ] ")
          ## rename tbl.tmp to tbl.final
          qRenameTable(tbl.old=tbl.tmp, tbl.new=tbl.final, schema=schema, verbose=TRUE, msg.pre="[Temporay table to current] ")
      }
    } else {
      message ("Don't forget to rename your table")
      message("Current table is ", schemaPaste(schema, tbl=tbl.tmp))
      message("FINAL   table is ", schemaPaste(schema, tbl=tbl.final))
    }
  } ## // END part 5

  ### -------- SAFETY CHECK: VERIFY / VALIDATE 2 of 2   -------- ###
  ## NOTE: For analytics, This takes a long time, though relatively not too long. About 5~10 minutes. 
  ##   Consider running this only on larger updates.
  if (!dontRename && tbl %in% c("accounting", "analytics"))
  try({

      verboseMsg(verbose, "\n\nVerifyication process (part 2 of 2) BEGINNING")

      ## MOVED dateCol.ll, dimCols.ll, colsToCompare.ll  TO  colsByTable.r
      source("~/git/orch/src/MgmtReport/supportFns/colsByTable.r")

      #### --------------
      colsToCompare <- copy(colsToCompare.ll[[tbl]])
      colsToCompare <- colNamesFromVector(colsToCompare)
      # colsToCompare # if (!is.null(names(colsToCompare)))
      # colsToCompare #   colsToCompare <- ifelse(names(colsToCompare) == "", colsToCompare, names(colsToCompare))

      colsToCompare.ratio <- paste0(colsToCompare, ".ratio")
      selfname_(colsToCompare.ratio)
      #### --------------

      minDate <- "2013-01-01"
      maxDate <- qMaxDate(tbl, verbose=FALSE, cluster=cluster.out) - 1
      ## We use Sys.Date() - 2 to allow for slight discrepency in most recent date. 
      ## TODO:  Can we minmize that discrepency
      qry.rowcounts.aggd <- makeQry(tbl=tbl.final, schema=schema, colsToPull=dimCols.ll[[tbl]], colsToAgg=colsToCompare.ll[[tbl]], dateCol=dateCol.ll[[tbl]], minDate=minDate, maxDate=maxDate)

      DT.rowcounts.aggd <- runQry(qry.rowcounts.aggd, cluster=cluster.out)

      qry.rowcounts.fact <- makeQry(tbl=tbl.fact, schema="production", colsToPull=dimCols.ll[[tbl.fact]], colsToAgg=colsToCompare.ll[[tbl.fact]], dateCol=dateCol.ll[[tbl.fact]], minDate="2013-01-01", maxDate=maxDate, where = pasteC(c(sprintf("NOT(transactiontypeid IN (%s)) ", pasteC(transacs.to.ignore, C=",")), if (tbl == "analytics") sprintf("(processeddaytime < '%s' OR processeddaytime IS NULL)", run_start_time)), C=" AND "))

      DT.rowcounts.fact_noErrs <- runQry(qry.rowcounts.fact, cluster=cluster.in)

      if (tbl.fact == "fact_analytics") {
        DT.rowcounts.errs <- runQry(convertQry.fa_to_faerrors(qry.rowcounts.fact))
        DT.rowcounts.fact <- rbind(DT.rowcounts.fact_noErrs, DT.rowcounts.errs)[, lapply(.SD, sumn), .SDcols=colsToCompare, keyby=date]
      } else {
        ## This step isn't necessary, but it is useful to have separate the two DTs (when using fact_analytics) for debugging purposes
        DT.rowcounts.fact <- copy(DT.rowcounts.fact_noErrs)
      }

      ## If the DT used periodi, change it to "date" (for now, applies only to fact_sales)
      addDateCols.periodid_(DT=DT.rowcounts.fact, newCol.nms="date", dropPeriodCols=TRUE, showWarnings=FALSE)


      matchKey(DT.rowcounts.aggd, DT.rowcounts.fact, "date")
      DT.rowcounts <- merge(DT.rowcounts.aggd, DT.rowcounts.fact, suffix=c(".aggd", ".fact"), all=TRUE)
      for (col in colsToCompare) {
        DT.rowcounts[, paste0(col, ".ratio") := get(paste0(col, ".aggd")) / get(paste0(col, ".fact"))]
        DT.rowcounts[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.rowcounts.sumn   <- DT.rowcounts[, lapply(.SD, sumn),   keyby=date, .SDcols=colsToCompare.ratio ]
      # compare #   DT.rowcounts.meann  <- DT.rowcounts[, lapply(.SD, meann),  keyby=date, .SDcols=colsToCompare.ratio ]
      # compare #   DT.rowcounts.unique <- DT.rowcounts[, lapply(.SD, unique), keyby=date, .SDcols=colsToCompare.ratio ]
      # compare #   compareDTs(DT.rowcounts.sumn, DT.rowcounts.unique)
      # compare #   compareDTs(DT.rowcounts.sumn, DT.rowcounts.meann)
      DT.rowcounts <- DT.rowcounts[, lapply(.SD, sumn), keyby=date, .SDcols=colsToCompare.ratio ]

      ## Acceptible "verified":
      ## The ratios for colsToCompare (other than rowcounts) must all be 1
      DT.rowcounts[, is_ok := all(equals1(.SD)), .SDcols = setdiff(colsToCompare.ratio, "row_count.ratio"), by=date]
      if (!all(DT.rowcounts$is_ok)) {
        folder.valid_jesus <- jesusForData(DT.rowcounts.fact, DT.rowcounts.aggd, DT.rowcounts, run_start_time, minDate, maxDate, envir=environment(), sub="validation2_failed")
        message(paste0("| ----- VALIDATION FAILED ----- |",
          unlist(strsplit(msgBox(silent=TRUE, paste(
            c( "Validation failed when comparing rows"
              , "jesus'ing DT.rowcounts  DT.rowcounts.fact  DT.rowcounts.aggd  run_start_time  minDate  maxDate"
              , folder.valid_jesus
              , "If re-running updateDeNormTable() make sure to set  dontRename=TRUE"
              , ""
              , "NOTE: When this fails, you want to pick a date (where !is_ok) and"
              , sprintf("      investigate (a) %s vs. (b) %s", schemaPaste("production", tbl.fact), schemaPaste(schema, tbl.final))
        ), sep="\n   ")), "\n")), "|", collapse="\n"))
      }
      notifyIfError(fail=fail_if_validate_fails, warn = !fail_if_validate_fails
                   , expr=all(DT.rowcounts$is_ok), subj=paste0(tbl, " - Verification part2 failed"), msg=
        sprintf("%s\n      ----     \nStopping ETL (just AFTER renaming the the existing tbl and renaming the '_new' table)\nMake sure to manually clean this\n", 
        if (DT.rowcounts[, all(date[!is_ok] == max(date))])
          sprintf("The most recent date, '%s', did not reconcile for tbl='%s'", max(DT.rowcounts$date), tbl)
        else 
          DT.rowcounts[!(is_ok), sprintf("The following date(s) did not reconcile for tbl='%s':\n  %s", tbl, pasteQand(date))]
      ))


      rc.fact <- round(DT.rowcounts.fact[date >= "2014-01-01", sum(row_count)], -6) / 1e9
      rc.aggd <- round(DT.rowcounts.aggd[date >= "2014-01-01", sum(row_count)], -6) / 1e9
      message(sprintf("  ----- For '%s' -----\nTotal rows for 2014 in fact table is %2.03f Billion\nTotal rows for 2014 in aggd table is % 2.03f Billion\nA decrease of %s", tbl, rc.fact, rc.aggd, fwp(percOf(rc.aggd, rc.fact))))

      ## VERIFY SPECIFIC STORES ---  
      ## 2014-12-02 TODO:  What would be a meaningful check here?  Make sure good for accounting and analytics
      # skip       if (FALSE)
      # skip       {
      # skip         subProj <- "VERIFY"
      # skip         setDBall(cluster=cluster.out)
      # skip         qry.verify <- sprintf("SELECT storeid, accounting_month, sum(gest) from %s where accounting_month > '2014-06-30' and storeid in (7, 187) group by 1, 2", schemaPaste(schema, tbl=tbl))
      # skip         DT.verify <- runQry(qry.verify, verbose=FALSE)[order(storeid, accounting_month)]
      # skip         # &&& 20141121 Update this to something more meaningful: 
      # skip         #    cat(pasteC(DT.verify[storeid == 7, sprintf("V: %s - %s", accounting_month, asCurr(sum, 0))], C="\n"))
      # skip       }

      verboseMsg(verbose, "Verifyication process (part 2 of 2) COMPLETED")
      ## NOTE:  This process will fail a few lines above, at the notifyIfError() if the validation is incorrect
      ## "return"
      TRUE
  })
  ### -------- END: SAFETY CHECK: VERIFY / VALIDATE 2 of 2   -------- ###



  ### -------- PART 6 DROP OLD TABLES -------- ###
  if (skip_to_part <= 6) {
    ## Drop All tabes named name Tbl_Removed_* except for the most recent Tables
    existing_tables <- qShowTables(namelike=sprintf("%s_removed_%%", tbl.final), schema=schema, justnames=TRUE, rowCount=FALSE, showWarnings=FALSE, wh=wh, dbname=dbname)
    if (length(existing_tables) > 1) 
      runQry(pasteC(sprintf("DROP TABLE %s;\n", sapply(head(sort(existing_tables), -1), schemaPaste, schema=schema))), verbose=TRUE)
  }

  ## SEE (when debugging): 
  if (FALSE)
    sort(qShowTables(namelike=sprintf("%s_removed_%%", tbl.final), schema=schema, justnames=TRUE, showWarnings=FALSE, wh=wh, dbname=dbname))

  ## Verbose output to user or log
  message("Table Created: ", schemaPaste(schema, tbl=ifelse(dontRename, tbl.tmp, tbl.final)) )

  if (skip_to_part <= 7) {
    if (skip_to_part > 6)
      message("No parts have been run, except for the validation checks")
  }


  return(invisible(TRUE))
}