# swap_in_new_table.r

sfzArchiveTable <- function(tbl, schema, schema_zArchive=schema, tbl_zArchive=timeStamp(paste(tbl, bak_word, sep=sep_bak)), dbname_zArchive=ifelse(bak_word=="zArchived","BackUps",dbname), wh=getSnowflakeWH(), dbname=getSnowflakeDB(), transient=TRUE, bak_word="zArchived", sep_bak="_", verbose.qry=TRUE, verbose=TRUE) {

  ## Check if the table to archive exists
  if (!qTableExists(dbname=dbname, schema=schema, tbl=tbl))
    stop("The table to zArchive ('", dbschematbl(dbname=dbname, schema=schema, tbl=tbl), "') does not exist.")

  if (grepl(paste0("^", escapeRegEx(sep_bak)), bak_word))
    warning("sep_bak ('", sep_bak, "') was detected in bak_word ('", bak_word, "'). This is redundant")

  ## RETURN
  sfCloneTable(tbl_from=tbl, tbl_to=tbl_zArchive, schema_from=schema, schema_to=schema_zArchive, dbname_from=dbname, dbname_to=dbname_zArchive, transient=transient, wh=wh, verbose.qry=verbose.qry, verbose=verbose)
}


sfCloneTable <- function(tbl_from, tbl_to, schema_from=schema, schema_to=schema, dbname_from=dbname, dbname_to=dbname_from, schema, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), transient=TRUE, verbose.qry=TRUE, verbose=TRUE) {

  if (!missing(schema) && !missing(schema_from) && missing(schema_to))
    warning("'schema_to' is missing but 'schema' and 'schema_from' are given;  'schema' is ignored.  'schema_to' will be set to same value as 'schema_from' ---- IS THIS WHAT YOU INTENDED?")

  ## INPUT CHECK
  ## --------------------------------------- ##
  if (!is.logical(transient) || length(transient) != 1)
    stop ("'transient' must be a either TRUE or FALSE;  it is of class '", class(transient)[[1]], "' with length ", length(transient))
  if (is.na(transient))
    stop("'transient' must be either TRUE or FALSE.  It is NA")
  is.char_of_length1(tbl_to)
  is.char_of_length1(tbl_from)
  is.char_of_length1(schema_to)
  is.char_of_length1(schema_from)
  ## --------------------------------------- ##

  # dst.old <- dbschematbl(dbname=dbname, schema=schema, tbl=tbl_from)
  # dst.new <- dbschematbl(dbname=dbname, schema=schema, tbl=tbl_to)
  # fmt.clone <- "CREATE OR REPLACE %s TABLE %s CLONE %s"
  if (!sfSchemaExists(schema=schema_to, dbname=dbname_to, wh=wh, verbose=FALSE))
    sfCreateSchema(schema=schema_to, dbname=dbname_to, wh=wh, verbose=TRUE)

  sprintf("CREATE OR REPLACE %s TABLE %s CLONE %s"
    , ifelse(isTRUE(transient), "TRANSIENT", "")
    , dbschematbl(dbname=dbname_to, schema=schema_to, tbl=tbl_to)
    , dbschematbl(dbname=dbname_from, schema=schema_from, tbl=tbl_from)
    ) %>%
  sfQry(wh=wh, verbose.qry=verbose.qry)
}

## NOTE: 2015-10-05:  changed argument to 'tbl_zArchive' from 'tbl.zArchive'
swap_in_new_table <- function(tbl_old=tbl.old, tbl_new=tbl.new, schema_old=schema, schema_new=schema, schema, tbl_zArchive=timeStamp(paste0(tbl_old, "_old_asof")), zArchive.old=TRUE, transient=TRUE, drop_new_after_rows_confirmed=TRUE, tbl.new, tbl.old, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), verbose.qry=TRUE, verbose=TRUE) {
## Replaces tbl_old with tbl_new, while optionally zArchiving the old table.
## If the rowcount for tbl_old and tbl_new are the same, tbl_new is dropped

  try({verboseMsg(verbose, ifelse(drop_new_after_rows_confirmed, "Renaming", "Copying"), "table:  ", dbschematbl(dbname=dbname, schema=schema_old, tbl=tbl_old), " ~~~> ", dbschematbl(dbname=dbname, schema=schema_new, tbl=tbl_new) , func="message", time=FALSE)})

  ## message about DEPRECATED
  {
    if (!missing(tbl.old) || !missing(tbl.new))
      warning ("tbl.old and tbl.new have been DEPRECATED. use undescore version tbl_new and tbl_old")
  }


  dst_old <- dbschematbl(dbname=dbname, schema=schema_old, tbl=tbl_old)
  dst_new <- dbschematbl(dbname=dbname, schema=schema_new, tbl=tbl_new)

  ## Check if tbl_old exists
  exists_old <- as.logical(qTableExists(dbname=dbname, schema=schema_old, tbl=tbl_old))
  exists_new <- as.logical(qTableExists(dbname=dbname, schema=schema_new, tbl=tbl_new))

  if (!exists_new)
    stop("Could not find the new (replacing) table '", dst_new, "'")

  ## if tbl_new is already transient, the new one must be transient too
  ## I am not sure if the other way around is true too or not?
  is_transient_new <- (sfShowTables(schema="bi", name="Accounting", nomatch="none")$kind == "TRANSIENT")
  if (is_transient_new && !transient) {
    warning ("The replacing table ('", dst_new, "') is TRANSIENT and therefore the final table must be as well.\n  Overriding user set value and using:\n      transient <- TRUE")
    transient <- TRUE
  }

  ## Optionally, zArchive the old table
  if (zArchive.old) {
    if (!exists_old)
      warning("A current version of the desination table '", dst_new, "' does not exist. There is nothing to zArchive")
    else
      sfzArchiveTable(tbl=tbl_old, schema=schema_old, dbname=dbname, dbname_zArchive=dbname, bak_word="old_asof", verbose.qry=verbose.qry)
  }

  ## CLONE
  sfCloneTable(tbl_from=tbl_new, tbl_to=tbl_old, schema_from=schema_new, schema_to=schema_old, dbname=dbname, verbose.qry=verbose.qry)

  ## Optionally drop tbl_new
  if (drop_new_after_rows_confirmed) {
    rows_old <- qRowCount(tbl=dst_old, msg_sf=FALSE, wh=wh)
    rows_new <- qRowCount(tbl=dst_new, msg_sf=FALSE, wh=wh)

    if (rows_old == rows_new)
      sprintf("DROP TABLE %s", dst_new) %>% sfQry(wh=wh, verbose.qry=verbose.qry)
    else
      warning (sprintf("\nSomething went wrong - table row count did not match for the destination\n  table %s (%s) and the table which was intended to replace\n     it %s (%s).\n\nTable %s was *NOT* dropped.", dst_old, rows_old, dst_new, rows_new, dst_new))
  }

  return(dst_old)
}