snowflakeStageCreate <- function(target_name, bucket, access_key_id=getOption("aws_aki"), secret_access_key=getOption("aws_sak"), connex=giveMeACon(verbose=FALSE), verbose=TRUE) {
  stop ("2015-07-20 --  there are other stage create functions    Are you sure youw ant to use this?")

  qry <- sprintf("CREATE STAGE %s URL = '%s' CREDENTIALS=(AWS_KEY_ID='%s' AWS_SECRET_KEY='%s');", target_name, bucket, access_key_id, secret_access_key)
  return(qry)
}

unloadTableToS3Bucket <- function(tbl, schema, bucket=s3_p("BI_Tables", tbl, stamp, prefix=tbl), stamp=timeStamp(), gzip=TRUE, cluster=NULL, force=FALSE, trialRun=TRUE, minDate=NULL, maxDate=NULL, dateCol=NULL, limit=if (trialRun) 1000 else NULL, wh=if (snowflake_inuse) getSnowflakeWH() else NULL, dbname=if (snowflake_inuse) getSnowflakeDB() else NULL, snowflake_inuse=getOption("snowflake_inuse", default=FALSE), manifest=TRUE, ...) {
  ##  ... : arguments to unloadQry, including bucket and manifest
  warning("As of 2015-07-07 unloadTableToS3Bucket() now returns the bucket instead of the query", call.=FALSE)

  if (!missing(limit) & !missing(trialRun))
    warning ("Manually setting 'limit' will render 'trialRun' useless")

  qry  <- makeQry(tbl=tbl, schema=schema, dbname=dbname, colsToPull="*", aggFunc=NULL, colsToAgg=NULL, minDate=minDate, maxDate=maxDate, dateCol=dateCol, limit=limit)

  if (!isTRUE(force)) {
    print(qry)
    ok <- readline(prompt=sprintf("%60s\nOk to run the above query on cluster %s and saving\nto bucket '%s'\n(y / n)? >", pasteR(57), ifelseNULL(cluster, "[default cluster]", cluster), bucket))
    if (!identical(tolower(substr(ok, 1, 1)), "y")) {
      cat ("\nExiting without executing query\n")
      return(invisible(NULL))
    } else cat("\n\n")
  }

  unloadQry(qry=qry, bucket=bucket, gzip=gzip, wh=wh, dbname=dbname, schema=schema, snowflake_inuse=snowflake_inuse, manifest=manifest, ...)

  return(bucket)
}

{
sfUnloadQry <- function(qry, bucket=NULL, stage_name=NULL, format_name="PROD.PRODUCTION.RICKS_DEFAULT_REDSHIFT_DUMP", on_error=c("continue", "skip_file", "abort_statement")
  , validation_mode=FALSE
  , overwrite=FALSE
  , basename=timeStamp("DataExport", seconds=TRUE)
  , manifest=TRUE, gzip=TRUE, delimiter="\t", esc.delim=TRUE
  , aws_aki=getOption("aws_aki", default=stop("aws_aki not found"))
  , aws_sak=getOption("aws_sak", default=stop("aws_sak not found"))

  , access_key_id="DEPRECATED"
  , secret_access_key="DEPRECATED"
  , just_qry=FALSE
  , connex=giveMeACon(verbose=FALSE, snowflake_inuse=snowflake_inuse, wh=wh, dbname=dbname, schema=schema), wh=if (snowflake_inuse) getSnowflakeWH() else NULL, dbname=if (snowflake_inuse) getSnowflakeDB() else NULL, schema=getSnowflakeSchema(), snowflake_inuse=getOption("snowflake_inuse", default=FALSE), verbose=TRUE)  {

  force(qry)

  warning ("Still left to do is to control the size of the file -- have it be 100Gigs or around there")

  if (missing(bucket) && missing(stage_name))
    bucket <- s3_p("SNOWFLAKE_OUTPUT", getProjName(showWarnings=FALSE), getSubProj(showWarnings=FALSE), warn_on_as.path=FALSE)

  # if (!is.null(stage_name)) {
  #   location <- sprintf("@%s/", stage_name) %>% gsub("//$", "/", .)
  # } else {
  #   location <- sprintf("%s \n    CREDENTIALS=(AWS_KEY_ID='%s' AWS_SECRET_KEY='%s')", gsub("//$", "/", bucket), aws_aki, aws_sak)
  # }

  location <- ifelseNULL(stage_name, no=sprintf("@%s", stage_name), yes=bucket) %>% 
                  removeText(pat="/+$", x=.) %>% 
                  paste0("/", basename)

  # if (is.null(format_name))
  #   format_name <- "PROD.PRODUCTION.RICKS_DEFAULT_REDSHIFT_DUMP"

  ## EG from website: 
  # COPY INTO @my_stage/result/data_ 
  # FROM (SELECT * FROM orderstiny)
  # file_format=(format_name='vsv' compression='gzip');



  # COPY INTO %s FROM %s
  #   CREDENTIALS = (AWS_KEY_ID = <string> AWS_SECRET_KEY = <string>)
  #   [FILE_FORMAT = (FORMAT_NAME = '<format_identifier>' | TYPE = <format_type> [ <file_format_options> ] ) ]
  #   [<copy_options>]
  #   [VALIDATION_MODE = <option>]

  if (!missing(access_key_id))
    stop("'access_key_id' has been DEPRECATED ---  use 'aws_aki'")
  if (!missing(secret_access_key))
    stop("'secret_access_key' has been DEPRECATED ---  use 'aws_sak'")

  ret <- 
    sfPopulateTable(
        qry=qry
      , location=location
      , cmd = "COPY"
      , stage_name=stage_name
      , format_name=format_name
      , bucket=bucket
      , overwrite=overwrite
      , on_error=on_error
      , validation_mode=validation_mode
      , aws_aki=aws_aki
      , aws_sak=aws_sak
      , wh=wh
      , dbname=dbname
      , schema=schema
      , just_qry=just_qry
      , verbose=verbose
    )

  if (just_qry)
    return(just_qry)

  verboseMsg(verbose, "Snowflake results dropped into \n  ", location, "\n", time=FALSE)
  return(location)
}
} # // outter bar

unloadQry <- function(qry, bucket, manifest=TRUE, gzip=TRUE, delimiter="\t", esc.delim=TRUE, access_key_id=getOption("aws_aki"), secret_access_key=getOption("aws_sak"), connex=giveMeACon(verbose=FALSE, snowflake_inuse=snowflake_inuse, wh=wh, dbname=dbname, schema=schema), wh=if (snowflake_inuse) getSnowflakeWH() else NULL, dbname=if (snowflake_inuse) getSnowflakeDB() else NULL, schema=getSnowflakeSchema(), snowflake_inuse=getOption("snowflake_inuse", default=FALSE), verbose=TRUE)  {

  ### ERROR CHECK 
  ### --------------------------------------------- ###
  is.char_of_length1(qry,               fail.if.not=TRUE)
  is.char_of_length1(bucket,            fail.if.not=TRUE)
  is.char_of_length1(delimiter,         fail.if.not=TRUE)
  is.char_of_length1(access_key_id,     fail.if.not=TRUE, hint="check that   options('aws_aki'='YOUR_aws_access_key_id')  is set")
  is.char_of_length1(secret_access_key, fail.if.not=TRUE, hint="check that   options('aws_sak'='YOUR_aws_secret_access_key')  is set")
  ## do not allow "'" (single quote) as delimiter
  if (delimiter %in% c("'", "'\\''", "\\'"))
    stop ("'delimiter' cannot be the single quote (\"\\'\")")
  ### --------------------------------------------- ###

  if (inVerboseMode.dbcon() || verbose)
    showDBsettings()

  if (verbose) {
      verbose.max.width <- getOption("width", 120) - 10
      verbose.max.lines <- 12
       hr <- paste0(pasteR(verbose.max.width+3), "\n")
       cat("\n\n\t    Running query [Began at ", timeStamp(frmt="%R %p") ,"] \n   ", hr
          , verboseQry(qry, max.lines=verbose.max.lines, max.width=verbose.max.width, indentAnd=TRUE, indentOr=TRUE)
          , "\n   ", hr, "\t    And unloading it to \"", bucket, "\"\n   ", sep="")
  }


  ### CLEAN

  ## NULL and other values are equivalent to FALSE for these 
  manifest <- ifelse(isTRUE(manifest), TRUE, FALSE)
  gzip     <- ifelse(isTRUE(gzip),     TRUE, FALSE)

  ## make sure qry and delimiter are not quoted
  remove_quotes <- . %>% {if (grepl("^'", .) && grepl("'$", .)) gsub("^'|'$", "", .) else .}
  qry        %<>% remove_quotes   
  delimiter  %<>% remove_quotes         

  ## Escape Chars
  qry <- escapeSpecialChars(qry, charsToEscape=c("\\", "'"))

  ## clean tabs
  if (esc.delim)
    delimiter <- escapeSpecialChars(delimiter, charsToEscape=c("\\"))


  ### PIECE IT ALL TOGETHER
  qry.full <- sprintf(
    "UNLOAD('%s')
     to '%s'
     credentials 'aws_access_key_id=%s;aws_secret_access_key=%s'
     delimiter '%s'
     %s%s
     ;
    "
    , qry, bucket, access_key_id, secret_access_key
    , delimiter
    , ifelse(manifest, " manifest\n", " ")
    , ifelse(gzip, " gzip\n", "")
    )

  browser(expr=inDebugMode("unload", "AWS"), text="right before executing in unloadQry()")

  ## EXECUTE -- reminder: We do not need to catch the output, since the query itself ships the output to disk
  if (snowflake_inuse)
    sfQry(qry=qry.full, connex=connex, wh=wh, dbname=dbname, schema=schema, snowflake_inuse=snowflake_inuse)
  else 
    suppressWarnings(dbGetQuery(connex, qry.full))

  return(invisible(qry.full))
}


loadFromBucket <- function(bucket, tbl, manifest=TRUE, delimiter="\t", gzip=TRUE
                        , dateformat="auto", timeformat=NULL
                        , esc.delim=TRUE
                        , allow_0_rows=FALSE
                        , create.if.not.exist=TRUE,  truncate.if.exist=FALSE
                        , access_key_id=getOption("aws_aki"), secret_access_key=getOption("aws_sak")
                        , connex=giveMeACon(verbose=FALSE)
                        , confirm=TRUE, confirm_dates=confirm
                        , wh=if (snowflake_inuse) getSnowflakeWH() else NULL, dbname=if (snowflake_inuse) getSnowflakeDB() else NULL, schema=NULL, snowflake_inuse=getOption("snowflake_inuse", default=FALSE)
                        , verbose=TRUE)
{
  ### ERROR CHECK 
  ### --------------------------------------------- ###
  is.char_of_length1(tbl,               fail.if.not=TRUE)
  is.char_of_length1(schema,            fail.if.not=TRUE)
  is.char_of_length1(bucket,            fail.if.not=TRUE)
  is.char_of_length1(delimiter,         fail.if.not=TRUE)
  is.char_of_length1(access_key_id,     fail.if.not=TRUE, hint="check that   options('aws_aki'='YOUR_aws_access_key_id')  is set")
  is.char_of_length1(secret_access_key, fail.if.not=TRUE, hint="check that   options('aws_sak'='YOUR_aws_secret_access_key')  is set")
  ## do not allow "'" (single quote) as delimiter
  if (delimiter %in% c("'", "'\\''", "\\'"))
    stop ("'delimiter' cannot be the single quote (\"\\'\")")
  ## bucket should exist
  if (!s3_bucketExists(bucket))
      stop ("loadFromBucket(tbl='", tbl, "') since the bucket '", bucket, "' appears to not exist\nIf you previously ran an unloadQry(), that may have failed. Check the Logs.")
  ## TODO:  This is really just sloppiness on my part
  if (!confirm && isTRUE(!allow_0_rows))
      stop ("'confirm' cannot be FALSE when allow_0_rows is not true")
  ### --------------------------------------------- ###

  stbl <- schemaPaste(schema=schema, tbl=tbl)

  if (inVerboseMode.dbcon() || verbose) {
    showDBsettings()
    cat("\n")
  }

  ### CHECK IF TABLE EXISTS, AND HANDLE ACCORDING TO FLAGS
  tableExists <- as.logical(qTableExists(tbl=tbl, schema=schema, exact=TRUE, ignoreCase=FALSE))

  if (create.if.not.exist && !tableExists) {
    stop ("\n    The table  ", stbl, "  does not exist. Cannot auto-create it.\n    Please create it manually then run loadFromBucket() again.\n\n    Hint: If you have a query, consider running\n          makeSQLtable(runQry(paste(qry, \"limit 1\"), verbose=FALSE), table.name=schemaPaste(schema=", gsub("\\[1\\] ", "", capture.output(schema)), ", \"", tbl, "\"))\n (dumping query)\n")
  }

  if (truncate.if.exist && tableExists) {
    verboseMsg(verbose, "   Truncating table ", stbl)
    dbGetQuery(conn=connex, statement=sprintf("truncate %s", stbl))
  }


  ## CLEAN QUERY PARAMS

  ## NULL and other values are equivalent to FALSE for these 
  manifest <- ifelse(isTRUE(manifest), TRUE, FALSE)

  ## make sure and delimiter are not quoted
  if (grepl("^'", delimiter) && grepl("'$", delimiter))
    delimiter  <- gsub("^'|'$", "", delimiter)
  gzip     <- ifelse(isTRUE(gzip),     TRUE, FALSE)

  ## clean tabs
  if (esc.delim)
    delimiter <- escapeSpecialChars(delimiter, charsToEscape=c("\\"))

  ## If not using a manifest file, bucket should not have prefix (ie, it should end with a ..last_sub_folder/ )
  ## Other wise, append "manifest" to bucket
  if (!manifest && !grepl("/$", bucket))
    bucket <- paste0(splitOnLast(bucket, "/") [[c(1, 1)]], "/")
  else if (manifest && !grepl("manifest$", bucket))
    bucket <- paste0(bucket, "manifest")


  ### PIECE IT ALL TOGETHER
  qry.full <- sprintf(
    "
     COPY %s
     from '%s'
     credentials 'aws_access_key_id=%s;aws_secret_access_key=%s'
     delimiter '%s'
     %s%s%s%s
     ;
    "
      , stbl, bucket, access_key_id, secret_access_key
      , delimiter
      , ifelse(manifest, " manifest\n"  , "")
      , ifelse(gzip,     " gzip\n"      , "")
      , ifelse(!is.null(dateformat), sprintf("dateformat '%s'\n", dateformat), "")
      , ifelse(!is.null(timeformat), sprintf("timeformat '%s'\n", timeformat), "")
    )

  ## EXECUTE
  verboseMsg(verbose, "   Populating ", stbl, " from bucket ", bucket)

  ## Debugging
  browser(expr=inDebugMode("load", "AWS"), text="right before executing in loadFromBucket()")

  ret <- try(suppressWarnings(dbGetQuery(conn=connex, qry.full)))

  ## Check for error and dump the query if something went wrong
  if (isErr(ret)) {
    hr <- paste0("\n   ", pasteR("-", 60), "\n")
    cat(hr, "          DUMPING QUERY", hr, qry.full, hr, "\n")
    message("FAILURE: Could not load data. Dumping query")
    return(qry.full)
  }

  ## TODO:  This could be a standalone function
  if (confirm || !allow_0_rows) {
    if (!qTableExists(tbl=tbl, schema=schema))
      warning (sprintf("Something went wrong.\nThe table  %s  does not exist in the DB.  Using Cluster %02i", stbl, getCluster())) 
    else {
      dateCols <- c("activity_date", "download_activity_date", "accounting_month", "download_accounting_month", "date", "activity_month544")
      cols <- qShowCols(tbl=tbl, schema=schema)
      rows <- qRowCount(tbl=tbl, schema=schema)

      hasRows <- rows > 0

      ## FAIL IF NO ROWS
      if (!allow_0_rows && !hasRows)
        stop("Loading did not take for ", stbl, " -- the table was created but there are no rows.\nHINT: Check that there is sufficient Disk Space on the cluster")

      dateCols <- intersect(dateCols, cols)
      hasDates <- confirm_dates && length(dateCols)

      date.msg <- ""
      if (hasDates && hasRows) {
        maxDates <- qMaxDate(tbl=tbl, schema=schema, dateCols=dateCols, verbose=FALSE, simplify=TRUE) #[1, sapply(.SD, function(x) as.character(as.Date(x, origin=.origin)))]

        if (lunique(maxDates) == 1)
          date.msg <- sprintf("\nThe latest date is %s (for %s %s)", unique(maxDates), plrl("columns", length(maxDates)), pasteQand(names(maxDates)))
        else 
          date.msg <- sprintf("\nThe latest dates are %s", commaSep(sprintf(" %s (%s)", maxDates, names(maxDates))))
      }

      row.msg  <- sprintf("and has %.0f rows", rows)
      row.msg  <- sprintf("and has %s (%.2f B) rows", formnumb(rows), rows / 1e9)
      msg <- boxit(sprintf("Table %s exists %s%s", stbl, ifelse(hasRows, row.msg, "but it has no rows."), ifelse(hasDates, date.msg, "")), min_width=90)
      message(msg)
    }
  }

  return(invisible(qry.full))
}