## This is part 01
##
## In the first part, we identified a few tables where we take the whole tables
## In this second part, we take just a few select columns from other important tables



# New Version of this process, began mid April 2016


setScience(proj="DeNormalizing", subProj="Escrow")
setGitBranchToSystem();  .g()

wh <- "SCIENCE"
dbname="prod"
schema="art_relations"

tbls <- c("publishing_escrow", "publishing_escrow_errors", "publishing_escrow_released", "vendor_contract")

frmt.row_count <- "select '%s' as tbl, count(*) as rows_%s from %s"
qry.row_count.mysql <- sprintf(frmt.row_count, tbls, 'expected', tbls) %>% pasteC(C = " UNION ")
qry.row_count.sf    <- sprintf(frmt.row_count, tbls, 'snowflake', dbschematbl(dbname=dbname, schema=schema, tbl=tbls)) %>% pasteC(C = " UNION ")


## ------------------------------------------------------------------------------- ##
## -------------------------------------------------------------------------------- ##
## The workflow
## -------------------------------------------------------------------------------- ##
##    For Each Table:
##    {
##      * Bring all of the data into memory  --  runQry("SELECT * FROM tbl")
##      * Save the data to disk              --  writeDT()
##      * Upload from disk to S3             --  s3_upload()
##      * Ingest from S3 to Snowflake        --  sfPopulateTable()
##    }
##
##    Because R cannot switch nicely between snowflake and mysql 
##     we first run all of the mysql items, then all of the snowflake items.
## ------------------------------------------------------------------------------- ##
## -------------------------------------------------------------------------------- ##



## ------------------------------------------ ##
## MYSQL PART                                 ##
## ------------------------------------------ ## 

## START WITH MYSQL
{
  setDBall("Mysql")

  ## Sometimes the Mysql package fails to connect.  If it does, this whole script should abort
  if (!(nrow(runQry("SHOW TABLES IN art_relations", verbose=FALSE)) > 100))
    stop("Something went wrong when connecting to the MySQL database")

  ll.f.out <- emptylist(tbls)
  ll.f_s3.out <- emptylist(tbls)

  for (tbl in tbls)
  {

    ## Clear up memory, just in case
    if (exists("DT"))
      rm(DT)
    gc()

    DT.desc <- runQry(sprintf("DESC %s", tbl))
    dateFields <- if ("Type" %in% names(DT.desc)) DT.desc[grepl("(time|date)(stamp)?", Type), Field]

    ## Select all of the data
    DT <- runQry(sprintf("SELECT * FROM %s", tbl))

    ## Slightly cleaning, as this will not transfer nicely to Snowflake
    for (col in dateFields) {
      if (col %in% names(DT)) {
        if (is.character(DT[[col]])) {
          catn("Checking for blanks in column '", col, "'", sep="")
          DT[grepl("0000-00-00", get(col)), (col) := NA]

          ## There are a few dates where the date is '00'.  Change those to '01'
          DT[grepl("\\-(0[1-9]|1[0-2])\\-00", get(col)), (col) := {
              if (.N)
                warning("Column '", col, "' has ", .N, " entries ending in '-00' -- these will be changed to '-01'", call.=FALSE)
              gsub("\\-(0[1-9]|1[0-2])\\-00", "-\\1-01", get(col))
            }]
        }
      } else {
        if (DT.desc[Field == col, Type] == "timestamp" && !is.POSIXct(DT[[col]]))
          warning("field '", col, "' is a timestamp in MySQL, but is stored as a ", is(DT[[col]])[[1]], " in R", call.=FALSE)
        if (DT.desc[Field == col, Type] %in% c("datestamp", "date") && !is.Date(DT[[col]]))
          warning("field '", col, "' is a ", DT.desc[Field == col, Type], " in MySQL, but is stored as a ", is(DT[[col]])[[1]], " in R", call.=FALSE)
      }
    }

    ## The filename anf subfolder where the data will be written to
    base.file.name <- dbschematbl(dbname=NULL, schema=schema, tbl=tbl) %>% timeStamp
    subfolder <- "escrow_exports"

    ## Output the DT to a text file, then upload the text file to S3
    f.out <- writeDT(DT, base.file.name=base.file.name, subfolder=subfolder)
    f_s3.out <- s3_upload(file=f.out, subbucket=as.path("mysql_sync", subfolder))

    ## Store the filenames for later use.  (Really only the s3 filename is needed)
    ll.f.out[[tbl]]    <- f.out
    ll.f_s3.out[[tbl]] <- f_s3.out
  } # // END for-loop

  ## Count how many rows are expected
  DT.expected_rows <- runQry(qry.row_count.mysql)
}

## ------------------------------------------ ##
## SNOWFLAKE PART                             ##
## ------------------------------------------ ## 
setSnowflake(wh=wh, dbname=dbname)
for (tbl in tbls)
  sfPopulateTable(bucket=ll.f_s3.out[[tbl]], tbl=tbl, schema=schema, dbname=dbname, truncate=TRUE, transient=TRUE, format_name="public.RICKS_TSV_HEADER", validation_mode=FALSE)
## Count how many rows are present
DT.snowflake_rows <- sfQry(qry.row_count.sf)



## ------------------------------------------ ##
## CONFIRM                                    ##
## ------------------------------------------ ## 
matchKey(DT.expected_rows, DT.snowflake_rows, keyCols="tbl", superset.ok=FALSE)
DT.confirm <- merge(DT.expected_rows, DT.snowflake_rows)
DT.confirm[, diff := rows_snowflake - rows_expected]
DT.confirm[, perc_missing := (rows_snowflake - rows_expected) / rows_expected]

wh_tbls.all_ok <- DT.confirm[diff == 0, tbl]
wh_tbls.zero_loaded <- DT.confirm[rows_snowflake == 0, tbl]
wh_tbls.some_errors <- DT.confirm[diff != 0 & rows_snowflake != 0, tbl]

if (length(wh_tbls.all_ok) == length(tbls) && all(sort(wh_tbls.all_ok) == sort(tbls))) {
  message("All ", length(tbls), " tables loaded OK!")
} else {
  if (length(wh_tbls.zero_loaded))
    warning("The following ", length(wh_tbls.zero_loaded), " tbl(s) did not load at all: ", pasteC(wh_tbls.zero_loaded, C=", "))
  if (length(wh_tbls.some_errors))
    warning("The following ", length(wh_tbls.some_errors), " tbl(s) did only partially loaded: ", pasteC(wh_tbls.some_errors, C=", "))
  catnn("More details: ")
  print(DT.confirm[diff != 0])
}


## ------------------------------------------ ##
## LOOKER                                     ##
## ------------------------------------------ ## 
if (FALSE) {
  sourceSupportFns(proj="Looker")
  create_lookml_model_from_schema(schema=schema, only_include_these_tbls=tbls, dbname=dbname, wh=wh, snowflake_inuse=TRUE)  
}