# -------------------------------------------------------------------------------------------------------------------------  #
  #  -----------------------------------------------------------------------------------------------------------------------   #
  #                                                                                                                            #
  #           File Name              :  SF_Utils - Table Creations.r                                                           #
  #           Last Updated Funclist  :  22 Jul 2015,  9:53 AM (Wednesday)                                                      #
  #                                                                                                                            #
  #           Author Name            :  Rick Saporta                                                                           #
  #           Author Email           :  RickSaporta@gmail.com                                                                  #
  #           Author URL             :  www.github.com/rsaporta                                                                #
  #                                                                                                                            #
  #           Packages Called        :  NA                                                                                     #
  #           Packages Used via NS   :  NA                                                                                     #
  #                                                                                                                            #
  #  -----------------------------------------------------------------------------------------------------------------------   #
  #                                                                                                                            #
  #   upload_file_to_snowflake ( file_fullpath, stage_name, prefix=NULL, tbl_out="!unspecified_tbl_destination"                #
  #                              , bucket=NULL, timestamp=timeStamp(), wh=getSnowflakeWH(), dbname=getSnowflakeDB()            #
  #                              , schema=getSnowflakeDB(), relativeTo=ingestDir, verbose=TRUE )                               #
  #   sfGetBucketFromStage     ( stage_name, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeSchema()         #
  #                              , showWarnings=TRUE, verbose.qry=verbose, verbose=TRUE )                                      #
  #   sfPopulateTable          ( tbl, stage_name=NULL, bucket=NULL, bucket_relative_to_stage, header="NOT USED"                #
  #                              , on_error=c("continue", "skip_file", "abort_statement"), delete_files_after_load=FALSE       #
  #                              , validation_mode=FALSE                                                                       #
  #                              , aws_aki=getOption("aws_aki", default=stop("aws_aki not found"))                             #
  #                              , aws_sak=getOption("aws_sak", default=stop("aws_sak not found"))                             #
  #                              , wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeSchema()                   #
  #                              , verbose.rows=verbose, verbose=TRUE )                                                        #
  #   sfShowFileFormat         ( justnames=FALSE, wh=getSnowflakeWH(), dbname=getSnowflakeDB()                                 #
  #                              , schema=getSnowflakeSchema(), verbose=FALSE )                                                #
  #   sfShowStages             ( justnames=FALSE, wh=getSnowflakeWH(), dbname=getSnowflakeDB()                                 #
  #                              , schema=getSnowflakeSchema(), verbose=FALSE )                                                #
  #   sfCreateFormatAndStage   ( stage_name, comment_stage="", format_name, comment_format="", header, delimiter="             #
  #                              ,", gzip=FALSE, quote_string=c("'", "\"", "NONE"), trim_space=FALSE                           #
  #                              , overwrite_if_exists=FALSE, create_schema_if_not_exist=TRUE, wh=getSnowflakeWH()             #
  #                              , dbname=getSnowflakeDB(), schema=getSnowflakeDB(), verbose=TRUE )                            #
  #   sfCreateStage            ( stage_name=name, bucket, format_name=NULL, comment=NULL                                       #
  #                              , comment_footer=getSFCommentFooter()                                                         #
  #                              , aws_aki=getOption("aws_aki", default=stop("aws_aki not found"))                             #
  #                              , aws_sak=getOption("aws_sak", default=stop("aws_sak not found"))                             #
  #                              , check_if_exists=TRUE, overwrite_if_exists=FALSE, wh=getSnowflakeWH()                        #
  #                              , dbname=getSnowflakeDB(), schema=getSnowflakeSchema(), name, verbose=TRUE )                  #
  #   sfCreateFileFormat.CSV   ( format_name=name, header, gzip=FALSE, delimiter="\t"                                          #
  #                              , quote_string=c("NONE", "'", "\""), trim_space=FALSE, NULL_IF=c("")                          #
  #                              , error_on_column_count_mismatch=TRUE, comment=NULL                                           #
  #                              , comment_footer=getSFCommentFooter(), wh=getSnowflakeWH(), dbname=getSnowflakeDB()           #
  #                              , schema=getSnowflakeSchema(), name, check_if_exists=TRUE, overwrite_if_exists=FALSE          #
  #                              , verbose=TRUE )                                                                              #
  #                                                                                                                            #
  #                                                                                                                            #
  #                                                     <END FUNCS>                                                            #
  #  -----------------------------------------------------------------------------------------------------------------------   #
  # -------------------------------------------------------------------------------------------------------------------------  #


"~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"

# ---------------------------------------------------- #
#         Personal Staging Bucket Best Practice        #
# ---------------------------------------------------- #
# A bucket_base can hold several tables
# Each table should have its own subfolder in the bucket_base
# Within each subfolder there can be multiple iterations of the table
# It is helpful, for easy cleaning and purging, to colllect these into another subfolder, by date or datetime
# Finally, the prefix_ should be the name of the input source 
#   * sometimes the same as the table, if transferring from redshift to snowflake
#   * No need for timestamping the prefix if using timestamp as the subfolder
# 
# ^^^ WRONG:  What if multiple source files to one tbl_out and we need to reprocess a file
# 
# bucket_for_stage           =  root/snowflake_stages/stage_name/
# bucket_for_file_processing =  root/snowflake_stages/stage_name/tbl_out/sourcefile.ext/datetimestamp/ #no need for prefix_
# ---------------------------------------------------- #
# ^^^ WRONG:  Again
#    The above only applies to table EXPORTS
#    For files, we want to simply upload the file, and kepe track of when it was processed
# 
# ---------------------------------------------------- #
#  UPDATED AND CORRECT: 
#     bucket_for_stage          = root/snowflake_stages/stage_name/
#     bucket_for_file_from_ETL  = root/snowflake_stages/stage_name/tbl_dest/tbl_from/datetimestamp/
#     bucket_for_file_from_Disk = root/snowflake_stages/stage_name/tbl_dest/relative/path/onLocal/filename.ext  # Any timestamp should be on the filename prior to now
# ---------------------------------------------------- #
#
## SOME STAGES
# sfCreateStage("query_unloads", bucket=s3_p("snowflake_stages", "query_unloads"), format_name="prod.public.BI_TABLES_DUMP_FROM_SNOWFLAKE_NOHEADER", comment="Auto Location for Unloading Queries for sfPopulateTable()", dbname="prod", schema="public")
# sfCreateFileFormat.CSV("query_unloads", bucket=s3_p("snowflake_stages", "query_unloads"), format_name="prod.public.BI_TABLES_DUMP_FROM_SNOWFLAKE_NOHEADER", comment="Auto Location for Unloading Queries for sfPopulateTable()", dbname="prod", schema="public")

if (FALSE) {
sfCreateFileFormat.CSV(format_name="CSV_GZIP_noHeader_noQuote", header=FALSE, gzip=TRUE, delimiter=",", quote_string="NONE", comment="No Header, No Quote, GZIP", schema="public", dbname="prod")
sfCreateFileFormat.CSV(format_name="CSV_GZIP_noHeader_Quote", header=FALSE, gzip=TRUE, delimiter=",", quote_string="\"", comment="No Header, No Quote, GZIP", schema="public", dbname="prod")

  ## writeDT EXPORT
sfCreateFileFormat.CSV(format_name="ricks_writeDT_export_with_esc", header=TRUE, gzip=FALSE, escape_string="\\\\", delimiter="\t", quote_string="\"", NULL_IF=c("NA"), dbname="prod", schema="BI", comment="Default export from R function writeDT() will be of this format")
## OLD AS OF 2015-10-09 (didnt have escape): sfCreateFileFormat.CSV(format_name="ricks_writeDT_export", header=TRUE, gzip=FALSE, delimiter="\t", quote_string="\"", NULL_IF=c("NA"), dbname="prod", schema="BI", comment="Default export from R function writeDT() will be of this format")  
}

#

"~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"

# 
# 
# "COPY INTO ", bucket, " FROM "{ [<namespace>.]<table_name> | <query> }
#     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>]
# 
# 
# 
# COPY INTO <external_location> FROM { [<namespace>.]<table_name> | <query> }
#     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 (FALSE) {  

  sfShowFileFormat()
  
  schema <- "production"
  tbl <- "staging_raw_apple_music"

  sfPopulateTableFromBucket(tbl=tbl.apple, schema=schema, stage=stage_name, bucket=bucket_apple, validation_mode=NULL)
}


## This is how I created the two main file formats
if (FALSE) {
  sfCreateFileFormat.CSV("RICKS_TSV_HEADER", header=TRUE, gzip=FALSE, delimiter="\t", quote_string="\"", dbname="prod", schema="public")
  sfCreateFileFormat.CSV("RICKS_CSV_HEADER", header=TRUE, gzip=FALSE, delimiter=",",  quote_string="\"", dbname="prod", schema="public")
}

create_and_populate_table_form_csv <- function(
    file
  , tbl=gsub("\\..*", "", basename(file))
  , schema="bi"
  , dbname=getSnowflakeDB()
  , wh=getSnowflakeWH()
  , header=TRUE
  , clean_column_names = FALSE
  , encoding_for_fread = c("unknown", "UTF-8", "Latin-1")
  , replace=FALSE
  , populate=TRUE
  , datetime_type = c("TIMESTAMP_NTZ", "TIMESTAMP_TZ", "TIMESTAMP_LTZ", "TIMESTAMP")
  , boolean_type = c("INT", "VARCHAR(5)", "VARCHAR(1)")
  , on_error=c("continue", "skip_file", "abort_statement")
  , integer64="numeric" ## passed to fread
  , delete_files_after_load=FALSE
  , sep=NULL ## used only to determine the format_name
  , format_name = if (is.null(sep)) NULL else if (sep == "\t") "PUBLIC.RICKS_TSV_HEADER" else if (sep == ",") "PUBLIC.RICKS_CSV_HEADER"
  , verbose=TRUE
) {
## This is a wrapper function to sfPopulateTable(file_name =..) 
##  Where, first, before shipping the file up to S3 and populating the table
##   the file is fread into R, and atable schema is created, then the table is itself created

  encoding_for_fread <- match.arg(encoding_for_fread)
  datetime_type <- match.arg(datetime_type)
  boolean_type  <- match.arg(boolean_type)
  on_error      <- match.arg(on_error)

  is.char_of_length1(file, fail=TRUE)
  is.char_of_length1(tbl, fail=TRUE)
  is.char_of_length1(schema, fail=TRUE)

  dst <- dbschematbl(dbname=dbname, schema=schema, tbl=tbl)

  if (!sfSchemaExists(schema=schema, verbose=FALSE)) {
    dbschematbl(dbname=dbname, schema=schema, tbl=NULL) %>% sprintf("CREATE SCHEMA %s", .) %>% sfQry(wh=wh)    
  }

  ## read in file, converted to data.table
  DT <- fread(file, header=header, integer64=integer64, encoding=encoding_for_fread)

  pat.ok <- get_pat_ok_sql_col_names()
  nms <- copy(names(DT))
  offending_nms <- nms[!grepl(pat.ok, nms)]
  if (length(offending_nms)) {
    msg <- paste("some column names of DT are not permitted as SQL column names. Namely \n", pasteC(offending_nms, C=",  "))
    ## If not flagged to clean the column names, fail
    if (!clean_column_names)
      stop(msg, "\n\nHINT: To have create_and_populate_table_form_csv() change the column names, set    clean_column_names=TRUE")

    setnames(DT, cleanColNamesForSQL_(clean_names_to_simple_alpha(nms, words_to_rem=NULL)))
    message(msg, " -- They have been cleaned")
    nms <- copy(names(DT))
  }

  ## Convert IDcols such as UPC etc
  setIDCols(DT)

  ## get and execute qry to create table
  QRY.create <- makeSQLtable(DT, tbl=tbl, schema=schema, dbname=dbname, create_only=TRUE, replace=replace, datetime_type=datetime_type, boolean_type=boolean_type, snowflake_inuse=TRUE, quiet=TRUE)
  ## clean up whitespace for bug on ODBC driver
  QRY.create <- gsub("\\) \\)", "))", gsub("\\n", " ", QRY.create)) %>% trim
  ret_create <- sfQry(QRY.create, wh=wh)

  if (!populate) {
    print(ret_create)
    return(dst)
  }

  ## populate the table from file
  ret <- sfPopulateTable(file_name=file, tbl=tbl, schema=schema, dbname=dbname, wh=wh, format_name=format_name, delete_files_after_load=delete_files_after_load, on_error=on_error, verbose=TRUE)
  print(ret)

  if ("first_error" %in% names(ret) && any(!is.na(ret$first_error))) {
    warning(call.=FALSE, "\n", ret[, sprintf("There was at least one error loading the data\n  Error Message: \"%s\"\n  Line Number:  %i\n  Character Number:  %i", first_error, first_error_line, first_error_character)])
  }

  if (replace) {
    nr.db <- qRowCount(tbl=tbl, schema=schema, wh=wh, dbname=dbname, snowflake=TRUE, msg_sf=FALSE)
    nr.f  <- fRowcount(file=file, header=header)
    confirmed <-  nr.f  ==  nr.db
    if (confirmed)
      verboseMsg(verbose, "Number of rows (", nr.f, ") confirmed in both, file and database", sep="")
    else
      warning("Number of rows differs between file (", nr.f, ") and database (", nr.db, ")")
  }

  return(dst)
}



upload_file_to_snowflake <- function(file_fullpath, stage_name, prefix=NULL, tbl_out="!unspecified_tbl_destination", bucket=NULL, timestamp=timeStamp(), wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeDB(), relativeTo=ingestDir, verbose=TRUE) {
  force(tbl_out); force(stage_name);

  subfolder <- file_fullpath %>% subbucketNameFromFileName(.) %>% gsub("^/", "", .)
  if (subfolder == "")
    subfolder <- c()

  use_stage  <- !missing(stage_name) && !is.null(stage_name)
  use_bucket <- !missing(bucket)     && !is.null(bucket)

  if (use_stage) {
      ## when using stage, bucket should be NULL
      if (use_bucket)
          warning("When stage_name is given explicitly, bucket is ignored in upload_file_to_snowflake()", call.=FALSE)
      "root/snowflake_stages/stage_name/"
      bucket_for_stage <- sfGetBucketFromStage(stage_name=stage_name, wh=wh, dbname=dbname, schema=schema, verbose=FALSE)
      bucket_for_file <- s3_p(root=NULL, bucket_for_stage, tbl_out, subfolder, file=basename(file_fullpath), warn_on_as.path=FALSE)
      "root/snowflake_stages/stage_name/tbl_dest/relative/path/onLocal/filename.ext"
  } else if (use_bucket) {
      bucket_for_file <- bucket
  } else {
      bucket_for_file <- ifelse (!missing(tbl_out), s3_p("snowflake_unsorted_files", tbl_out, timestamp, warn_on_as.path=FALSE), s3_p("snowflake_unsorted_files", warn_on_as.path=FALSE))
  }

  s3_upload(file.local=file_fullpath, bucket=bucket_for_file, verbose=verbose)
}


sfGetBucketFromStage <- function(stage_name, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeSchema(), showWarnings=TRUE, verbose.qry=verbose, verbose=TRUE) {
  force(stage_name)
  if (grepl("\\.", stage_name)) {
    parts <- strsplit(stage_name, "\\.")[[1]]
    if (length(parts) == 3) {
      dbname <- parts[[1]]
      schema <- parts[[2]]
      stage_name <- parts[[3]]
    } else if (length(parts) == 2) {
      schema <- parts[[1]]
      stage_name <- parts[[2]]
    } else
      warning("stage_name '", stage_name, "' created ", length(parts), " pieces when split on '.'")
  }

  DT.stages <- sfShowStages(wh=wh, dbname=dbname, schema=schema, verbose=verbose.qry, set=FALSE)

  setkey(DT.stages, name)
  buckets <- DT.stages[.(toupper(stage_name)), url]

  if (showWarnings && any(is.na(buckets)))
    warning (stage_name[is.na(buckets)] %>% sprintf("Could not find stage '%s'", .) %>% pasteC(C="\n"), call.=FALSE)

  return(buckets)
}



{
sfPopulateTable <- function(
    tbl = NULL
  , cmd = c("COPY", "INSERT", "UPSERT", "UPDATE")
  , stage_name = if (using_qry) "query_unloads" else NULL
  , format_name = NULL

  , DT=NULL
  , bucket=NULL
  , bucket_relative_to_stage
  , file_name=NULL
  , header="NOT USED"  ## Logical or an integer
  , on_error=c("continue", "skip_file", "abort_statement")
  , delete_files_after_load=FALSE
  , validation_mode=FALSE
  , timestamp=timeStamp()
  , aws_aki=getOption("aws_aki", default=stop("aws_aki not found"))
  , aws_sak=getOption("aws_sak", default=stop("aws_sak not found"))
  , qry
  , location
  , encoding_for_fread = c("unknown", "UTF-8", "Latin-1")
  , boolean_type  = c("INT", "VARCHAR(5)", "VARCHAR(1)") ## Applies only when DT is used
  , clean_column_names=FALSE ## passed through to create_and_populate_table_form_csv
  , integer64="numeric" ## passed through to create_and_populate_table_form_csv
  , transient=TRUE   ## only applies for qry
  , overwrite=FALSE  ## only applies for qry
  , truncate=FALSE ## applies to all, so long as table exists
  , wh=getSnowflakeWH()
  , dbname=getSnowflakeDB()
  , schema=getSnowflakeSchema()
  , just_qry = FALSE
  , using_qry = !missing(qry) && !is.null(qry)
  , verbose.rows=verbose
  , verbose=TRUE
) {



## TODO 1:  Allow for override of these file format options
## This would involve changing the CREATE INTO statment format type
##    [FILE_FORMAT = (FORMAT_NAME = '<format_identifier>' | TYPE = <format_type> [ <file_format_options> ] ) ]
##    https://documentation.snowflakecomputing.com/manuals/sql-reference/sql/copy-into-table.html
##
  # list(
  # , FIELD_DELIMITER = NULL # '<character>' | 'NONE'
  # , RECORD_DELIMITER = NULL # '<character>' | 'NONE'
  # , SKIP_HEADER = NULL # <integer>
  # , DATE_FORMAT = NULL # '<string>' | 'AUTO'
  # , TIMESTAMP_FORMAT = NULL # '<string>' | 'AUTO'
  # , ESCAPE = NULL # '<character>' | 'NONE'
  # , ESCAPE_UNENCLOSED_FIELD = NULL # '<character>' | 'NONE'
  # , TRIM_SPACE = NULL # TRUE | FALSE
  # , FIELD_OPTIONALLY_ENCLOSED_BY = NULL # '<character>' | 'NONE'
  # , NULL_IF = NULL # ( '<string>' [, '<string>'] ... )
  # , ERROR_ON_COLUMN_COUNT_MISMATCH = NULL # TRUE | FALSE
  # )

  if (missing(tbl) && !missing(DT)) {
    DT.nm <- capture.output(substitute(DT))
    tbl <- nm_from_DT.nm(DT.nm)
  }

  ## make sure necessary arguments are present
  force(tbl); force(stage_name); force(bucket); force(file_name); force(format_name); force(dbname); force(schema); force(using_qry);
  encoding_for_fread <- match.arg(encoding_for_fread)
  boolean_type <- match.arg(boolean_type)

  ## SANITY CHECK -- only one of DT, file_name, stage, bucket
  if (!missing(DT) && !nchar(getProjName()))
    stop ("When passing a DT to sfPopulate(), there must be a project set, since writeDT needs to use out.p()\nHINT: call\n       setScience(proj='misc', subl=FALSE, load=FALSE)", call.=FALSE)
  if (!missing(DT) && !missing(format_name))
    stop ("format_name is not used when DT is explicity")
  if (!missing(DT) && !missing(file_name))
    stop ("Only one of DT or file_name could be used")
  if ((!missing(DT) || !missing(file_name))  &&  !(missing(stage_name) && missing(bucket)))
    stop ("stage_name and bucket should NOT be used when DT or file_name are being used")

  if (truncate && overwrite) {
    warning("Both truncaste and overwrite are TRUE.  This will have the effect of first truncating the table, then dropping it.\n\nHINT: Did you mean to not set overwrite?")
  }

  if (truncate && qTableExists(tbl=tbl, dbname=dbname, schema=schema, snowflake_inuse=TRUE, msg_sf=FALSE, verbose=FALSE)) {
      dbschematbl(tbl=tbl, dbname=dbname, schema=schema) %>% 
        sprintf("TRUNCATE TABLE %s", .) %>% 
        sfQry()
  }

  if (using_qry) {
    ## Exactly only one of tbl, location, bucket, stage to be specified; The exception is that stage_name has a default value if no other is specified
    if (sum(!is.null(tbl), !is.null(bucket), !missing(stage_name), !missing(location)) %>% {!(. == 1 || (. == 0 && !is.null(stage_name)))})
      stop ("When using 'qry' exactly one of 'tbl', 'location', 'stage_name', 'bucket' must be specified.\n When unloading a query, it can either go to a table or to a location.\nthe location can be given explicitly or it can be pieced together from stage_name or bucket", call.=FALSE)
    ##  TODO: any other checks?

    ## if unloading to a tbl, use CREATE TABLE AS ... 
    if (!is.null(tbl)) {
      start_time <- now()
      qry_full <- sprintf("CREATE %s %s TABLE %s AS\n%s", ifelse(overwrite, "OR REPLACE", ""), ifelse(transient, 'TRANSIENT', ""), dbschematbl(tbl=tbl, schema=schema, dbname=dbname), qry=qry) %>% trim %>% paste0(";")
      if (just_qry)
        return(qry_full)
      ret <- sfQry(qry_full, wh=wh, verbose=verbose)
      end_time <- now()
      rows <- qRowCount(tbl=tbl, schema=schema, dbname=dbname, wh=wh, msg_sf=FALSE, snowflake_inuse=TRUE)
      message(sprintf("Unloading the query took %s\nTABLE %s now has %s rows", fwTDiff(start_time, end_time), dbschematbl(tbl=tbl, schema=schema, dbname=dbname), formnumb(rows, round=FALSE)))
      return(invisible(ret))
    }


    # unloading a complex query... not needded for now    ## Check if query is complex; in which case need to first unload to bucket, then load fr
    # unloading a complex query... not needded for now    is_complex <- length(gregexpr(pat="\\bSELECT\\b", qry, ignore.case=TRUE)[[1]]) > 1
    # unloading a complex query... not needded for now    if (!is.null(tbl) && is_complex) {
    # unloading a complex query... not needded for now      message("COMPLEX QUERY DETECTED -- WILL NEED TO UNLOAD TO BUCKET, THEN LOAD FROM BUCKET")
    # unloading a complex query... not needded for now
    # unloading a complex query... not needded for now      if (!qTableExists(tbl=tbl, schema=schema, dbname=dbname, snowflake_inuse=TRUE))
    # unloading a complex query... not needded for now        stop ("Complex queries will not auto-create the table.  No idea how to program this")
    # unloading a complex query... not needded for now
    # unloading a complex query... not needded for now      ## PARAMS USING      
    # unloading a complex query... not needded for now      bucket <- s3_p("snowflake_stages", "query_unloads", "temp_location_for_tbl_populate", prefix=timeStamp("query_unload", seconds=TRUE))
    # unloading a complex query... not needded for now      format_name <- "PROD.PUBLIC.CSV_GZIP_noHeader_Quote"
    # unloading a complex query... not needded for now
    # unloading a complex query... not needded for now      sfPopulateTable(qry=qry, schema=schema, dbname=dbname, location=bucket, format_name=format_name, just_qry=FALSE)
    # unloading a complex query... not needded for now      message("~~~~~~~~~~~~~~~ LOADING ~~~~~~~~~~~~~~~")
    # unloading a complex query... not needded for now      return(sfPopulateTable(tbl=tbl, schema=schema, dbname=dbname,   bucket=bucket, format_name=format_name, just_qry=FALSE))
    # unloading a complex query... not needded for now    }
  }


  ## SANITY CHECK -- proj is required when using DT
  if (!missing(DT) && is.null(getProjName()))
    stop ("project MUST be set when using DT in sfPopulateTable()", call.=FALSE)

  if (!missing(file_name) && missing(format_name)) {
    if (tolower(extractExt(file_name)) == "tsv")
      format_name = "prod.public.RICKS_TSV_HEADER"
    if (tolower(extractExt(file_name)) == "csv")
      format_name = "prod.public.RICKS_CSV_HEADER"
  }

  ## Check that the table exists;  if not, try to create it.
  if (!using_qry && !is.null(tbl)) {
     if (overwrite) {
        stop("The argument 'overwrite' only works when using a 'qry'.  Did you mean to use TRUNCATE?  Or should the code here be changed to allow for DROP TABLE?")
      }


    if (!qTableExists(dbname=dbname, schema=schema, tbl=tbl, msg_sf=FALSE, snowflake_inuse=TRUE)) {
      if (!missing(file_name)) {
        message("The table does not exist -- will try to create ", dbschematbl(dbname=dbname, schema=schema, tbl=tbl))
        ## temp for bug in ODBC ## create_and_populate_table_form_csv(file=file_name, dbname=dbname, schema=schema, tbl=tbl, wh=wh, format_name=format_name, integer64=integer64)
        ## temp for bug in ODBC ##
        ## Note that there is a bug in the ODBC DRIVER

## 2015-10-03: create_and_populate_table_form_csv is helpful because it CREATES the table
##             however, it then calls sfPopulate, which is NOT needed;
##             Thus the return statement here. 
##             A better solution is to change the function to simply 'create_table_from_csv' and drop the call to sfPopulate(). 
    ## TODO:   But first, we need to make sure that there are no other scripts that depend on this
        return(create_and_populate_table_form_csv(file=file_name, dbname=dbname, schema=schema, tbl=tbl, wh=wh, format_name=format_name, integer64=integer64, encoding=encoding_for_fread, clean_column_names=clean_column_names))
        try({create_and_populate_table_form_csv(file=file_name, dbname=dbname, schema=schema, tbl=tbl, wh=wh, format_name=format_name, integer64=integer64, encoding=encoding_for_fread, clean_column_names=clean_column_names)})
        if (!qTableExists(dbname=dbname, schema=schema, tbl=tbl, msg_sf=FALSE, snowflake_inuse=TRUE))
          stop ("Tried to create the table but failed")
      }
      else
        stop ("Table does not exist, please create it first.\nIf you have a CSV available, you can use create_and_populate_table_form_csv()\nIf you are using a DT, use function ingestIntoSQL(DT, ..)   with all the same parameters")
    }
  ## tbl should not be NULL, then a qry must be specified; otherwise it's an error
  } else if (!using_qry) {
    stop ("either qry or tbl should be specified and NOT NULL")
  }

  cmd <- match.arg(cmd)

  if (!missing(format_name) && !missing(stage_name) && !is.null(stage_name))
    warning ("format_name is ignored when using a stage;  stages have their own format attached to them")

  ## make sure header is not given
  if (!missing(header))
    warning("'header' should be set when creating the file format. It is ignored in sfPopulateTable() and sfPopulateTableFromFile()", call.=FALSE)

  if (!missing(DT)) {
    ## CONVERT BOOLEANS TO INTEGERS
    if (length(boolCols <- nwhich(sapply(DT, is.logical)))) {
      DT <- copy(DT)
      convertBooleans_(DT, boolean_type=boolean_type)
    }
    verboseMsg(verbose, "Writing DT to temp file", func="message")
    file_name <- writeDT(DT, subfolder="exports_via_ingestintosql", base.file.name=paste0(tbl, "_TEMPFILE_", timeStamp(sec=TRUE)), ext="tsv", sep="\t", proppernames=FALSE, header=TRUE)
    # BAD FORMAT (old as of 2015-10-09): format_name <- "PROD.BI.RICKS_writeDT_export"
    format_name <- "PROD.BI.RICKS_writeDT_export_with_esc"
    on.exit(unlink(file_name), add=TRUE)
  }

  if (!is.null(file_name)) {
    verboseMsg(verbose, "Uploading file to bucket", func="message")
    ## UPLOAD THE FILE
    ## TODO:  Allow for a stage or a bucket
    bucket_for_file <- upload_file_to_snowflake(file=file_name, tbl_out=tbl, timestamp=timestamp, stage_name=stage_name, bucket=bucket, schema=schema, dbname=dbname)

    if (is.null(bucket) || bucket != bucket_for_file) {
      message("File is uploaded to '", bucket_for_file, "'")
      bucket <- bucket_for_file
    }
  }


  ## THIS SAME CODE CAN BE USED FOR UNLOADING TO EXTERNAL LOCATION, JUST SWAP THE FROM/INTO
  if (using_qry)
  {
    if (!missing(bucket_relative_to_stage) && !is.null(bucket_relative_to_stage))
      warning ("bucket_relative_to_stage is ignored when unloading a query")

    ## If location is not given explicitly with qry, create it

    if (!is.null(tbl)) {
      dstbl <- dbschematbl(dbname=dbname, schema=schema, tbl=tbl)
      qTableExists(dbname=dbname, schema=schema, tbl=tbl, msg_sf=FALSE, snowflake_inuse=TRUE) %>% ifelse(yes="Destination table already exists. Values will be appended.", no="Destination does not exist and will be created at unload of qry.") %>% message()
      use_credentials <- FALSE
    } else {
      if (missing(location)) {
        if (is.null(bucket) && is.null(stage_name))
          stop("When unloading a qry, if location and tbl are missing, then either bucket or stage_name must be non-NULL")

        ## calculate location
        location <- bucket %>% valueIfNull(sprintf("@%s/%s", stage_name, timeStamp("query_unload", seconds=TRUE))) %>% gsub("//+$", "/", .)
      }

      ## confirm a single string
      is.char_of_length1(location, fail=TRUE)
      use_credentials <- !grepl("@", location)

      dstbl <- location
    }
        
    ## 'FROM' is the qry where we are unloading from;  
    ##  'TO' (aka 'dstbl') is the location where we are unloading to
    FROM_CLAUSE <- qry %>% trim %>% gsub("\n", "\n\t", .) %>% sprintf("\nFROM (\n\t%s\n)\n", .)
    ## for non-stages we need to add a CREDENTIALS line
    if (use_credentials)
      FROM_CLAUSE %<>% sprintf("%s \n    CREDENTIALS=(AWS_KEY_ID='%s' AWS_SECRET_KEY='%s')", ., aws_aki, aws_sak)
    if (overwrite)
      FROM_CLAUSE %<>% sprintf("%s \n    OVERWRITE=TRUE", .)

    if (is.null(format_name))
      format_name <- "prod.public.BI_TABLES_DUMP_FROM_SNOWFLAKE_NOHEADER"

    ## verbose.rows only applies for when writing to table
    verbose.rows <- FALSE

  ## Otherwise, populate from some already existing source (DT, csv, stage, bucket)
  } else {

    ## tbl defaults to NULL so that force(tbl) above will not fail
    if (is.null(tbl))
      stop ("tbl is missing; it cannot be NULL")

    ## FROM_CLAUSE will either be 
    ##  EITHER  'FROM @stage/subfolder'    when using stage
    ##      OR  'FROM s3://bucket/subfolder   CREDENTIALS = (...)'  when using bucket
    ##
    if (!missing(stage_name) && !is.null(stage_name)) {
      ## We want bucket_relative_to_stage;  However, generally we have the full bucket.
      ## Thus, we can calculate the bucket_relative_to_stage by finding bucket_for_stage
      ## determine bucket_relative_to_stage, based on given bucket and stage's bucket
      if (missing(bucket_relative_to_stage)) {
        if (missing(bucket))
          stop ("at least one of bucket or bucket_relative_to_stage must be used with stage_name")
        bucket_for_stage <- sfGetBucketFromStage(stage_name=stage_name, wh=wh, dbname=dbname, schema=schema, verbose=FALSE)
        bucket_relative_to_stage <- bucket %>% gsub(bucket_for_stage, "", .)
      }

# 2015-09-20 removed      ## 2015-08-25 I believe format_name should not be used when stage is used;  I could be wrong here
# 2015-09-20 removed      if (!is.null(format_name) && (missing(DT) && missing(file_name))) {
# 2015-09-20 removed        warning("Rick is not sure if format_name should be used when stage is used -- EXCEPT WHEN CREATING THE TABLE;  I believe it should NOT be used and therefore, will not be -- setting it to NULL")
# 2015-09-20 removed        format_name <- NULL
# 2015-09-20 removed      }

      ## remove any proceeding '/' to avoid '//'
      bucket_relative_to_stage %<>% gsub("^/", "", .)
      FROM_CLAUSE <- sprintf("    FROM @%s/%s", stage_name, valueIfNull(bucket_relative_to_stage, ""))
    } else if (!missing(bucket) && !is.null(bucket)) {
      FROM_CLAUSE <- sprintf("    FROM '%s' \n    CREDENTIALS=(AWS_KEY_ID='%s' AWS_SECRET_KEY='%s')", bucket, aws_aki, aws_sak)
    } else {
      stop ("One of either bucket or stage_name must be given explicitly and be non-NULL")
    }

    dstbl <- dbschematbl(dbname=dbname, schema=schema, tbl=tbl)

    if (verbose.rows)
      row_count.before_load <- qRowCount(tbl=tbl, schema=schema, dbname=dbname, wh=wh, msg_sf=FALSE, snowflake_inuse=TRUE)
  }

  on_error <- match.arg(on_error)

  browser(expr=inDebugMode(c("sfPopulate", "sfPopulateTable")), text="in sfPopulateTable() before composing query")

  if (!is.null(validation_mode) && !identical(validation_mode, FALSE)) {
    if (isTRUE(validation_mode))
      validation_mode <- "RETURN_ERRORS"
    if (grepl("^'|'$", validation_mode)) {
      warning ("do not quote validation_mode")
      validation_mode <- gsub("^'|'$", "", validation_mode)
    }
    part.validation_mode <- sprintf("VALIDATION_MODE = '%s'", validation_mode)
  } else {
    part.validation_mode <- ""
  }

  part.file_format = ifelseNULL(format_name, yes="", no=sprintf("FILE_FORMAT = (FORMAT_NAME = '%s')", format_name))

  QRY <- sprintf("
  %s INTO %s
    %s
    ON_ERROR = %s
    PURGE = %s
    %s
    %s
    "
    , cmd
    , dstbl
    , FROM_CLAUSE
    , toupper(on_error)
    , isTRUE(delete_files_after_load)
    , part.file_format
    , part.validation_mode
    )
  
  QRY %<>% trim %>% paste0(";") %>% setQry

  ## This is a horribly sloppy way of addressing this need, but....
  ## ON_ERROR and PURGE do not apply for unloading queries
  if (!missing(qry))
    QRY %<>% removeText("\\n\\s+ON_ERROR = [a-zA-Z]+ *", .) %>% removeText("\\n\\s+PURGE = [a-zA-Z]+ *", .)

  if (just_qry)
    return(QRY)

  ## EXECUTE
  ret <- sfQry(QRY, verbose=verbose, wh=wh, dbname=dbname, schema=schema)

  if (verbose.rows) {
    row_count.after_load <- qRowCount(tbl=tbl, schema=schema, dbname=dbname, wh=wh, msg_sf=FALSE, snowflake_inuse=TRUE)
    (row_count.after_load - row_count.before_load) %>% formnumb(round=FALSE) %>% 
      sprintf("%s rows loaded", .) %>% cat("\n")
  }

  if (all(c("rows_parsed", "rows_loaded") %in% names(ret)) && nrow(ret)) {
    parsed <- sumn(ret$rows_parsed)
    loaded <- sumn(ret$rows_loaded)
    fr <- . %>% formnumb(round=FALSE)
    msg.rows_not_loaded <- ifelse(loaded, sprintf("%s of %s rows were NOT loaded", fr(parsed - loaded), fr(parsed)), sprintf("NONE of the %s rows were loaded", fr(parsed)))
    if (any(parsed != loaded)) {
      warning ("It looks like there were errors loading the data via sfPopulateTable()\n", msg.rows_not_loaded, "\n", call.=FALSE)
      print(ret)

      if (!missing(DT)) {
        tmp_file <- tempfile(pattern=paste0(tbl, "_to_load"), fileext=".csv")
        file.copy(from=file_name, to=tmp_file)
        catn("Backup temp file where DT was outputted:\n    ", tmp_file)
      }
      return(invisible(ret))
    }
  }

  return(ret)
}
} # outer band

sfShowFileFormat <- function(justnames=FALSE, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeSchema(), verbose=FALSE) {
  if (!is.null(wh) || !is.null(dbname) || !is.null(schema))
    setSnowflake(wh=wh, dbname=dbname, schema=schema)

  ret <- sfQry("SHOW FILE FORMATS", verbose=verbose)
  if (justnames)
    return(ret[["name"]])
  return(ret)
}

sfShowStages <- function(justnames=FALSE, wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeSchema(), set=TRUE, verbose=FALSE) {

  if (set && (!is.null(wh) || !is.null(dbname) || !is.null(schema)))
    setSnowflake(wh=wh, dbname=dbname, schema=schema)
  
  Qry <- sprintf("SHOW STAGES %s;", ifelse(isNULLorBlank(schema), "", paste0(" in ", schema))) %>% setQry

  ret <- sfQry(Qry, wh=wh, dbname=dbname, schema=schema, verbose=verbose)

  if (justnames)
    return(ret[["name"]])
  return(ret)
}


sfCreateFormatAndStage <- function(
  stage_name
, comment_stage=""

, format_name
, comment_format=""
, header
, delimiter=","
, gzip=FALSE
, quote_string=c("'", '"', "NONE")
, trim_space=FALSE
, overwrite_if_exists=FALSE

, create_schema_if_not_exist=TRUE
, wh=getSnowflakeWH()
, dbname=getSnowflakeDB()
, schema=getSnowflakeDB()
, verbose=TRUE
) {

  ## Since all of the sfQry calls take time;  force all parameters now
  force(header); force(gzip); force(format_name); force(stage_name)

  bucket_for_stage <- s3_p("snowflake_stages", stage_name)

  setSnowflake(wh=wh, dbname=dbname, schema=schema)

  if (create_schema_if_not_exist && !sfSchemaExists(schema, dbname=dbname, wh=wh))
    schema %>% sprintf("CREATE SCHEMA %s", .) %>% sfQry

  if (!sfSchemaExists(schema, dbname=dbname, wh=wh))
    stop (sprintf("schema '%s' does not exist in dbname='%' -- cannot create stage using said schema", schema, dbname))

  # if (overwrite) {
  #   message("overwrite is sloppy in its coding")
  #   try({sprintf("DROP FILE FORMAT %s.%s.%s", dbname, schema, format_name) %>% sfQry(wh=wh, verbose=FALSE)})
  #   try({sprintf("DROP stage_name %s.%s.%s", dbname, schema, stage_name) %>% sfQry(wh=wh, verbose=FALSE)})
  # }

  if (missing(gzip) || missing(delimiter))
    sprintf("Creating file format '%s' using delim = '%s' and %s", format_name, delimiter, ifelse(gzip, "using compression GZIP", "NOT using compression")) %>% message

  ## THESE ARE WRAPPED IN try() BECAUSE OF A BUG IN SNOWFLAKE ODBC -- ONCE BUG IS FIXED, REMOVE THE TRY COMMANDS
  try(sfCreateFileFormat.CSV(format_name=format_name, comment=comment_format, header=header, delimiter=delimiter, quote_string=quote_string, trim_space=trim_space, gzip=gzip, overwrite_if_exists=overwrite_if_exists, wh=wh, dbname=dbname, schema=schema, verbose=verbose))
  try(sfCreateStage(stage_name=stage_name, bucket=bucket_for_stage, format_name=format_name, comment=comment_stage, overwrite_if_exists=overwrite_if_exists, wh=wh, dbname=dbname, schema=schema, verbose=verbose))

  confirm.format <- (format_name %>% toupper %in% sfShowFileFormat(justnames=TRUE, wh=wh, dbname=dbname, schema=schema))
  confirm.stage <- (stage_name %>% toupper %in% sfShowStages(justnames=TRUE, wh=wh, dbname=dbname, schema=schema, set=FALSE))

  if (!confirm.format)
    warning(sprintf("format_name '%s' could not be confirmed as created in %s.%s", format_name, dbname, schema))
  if (!confirm.stage)
    warning(sprintf("stage_name '%s' could not be confirmed as created in %s.%s", stage_name, dbname, schema))

  return(list(bucket_for_stage=bucket_for_stage, stage_name=stage_name, format_name=format_name))
}


sfCreateSchema <- function(schema=schema
  , comment=NULL
  , comment_footer=getSFCommentFooter()
  , check_if_exists=TRUE
  , overwrite_if_exists=FALSE
  , wh=getSnowflakeWH()
  , dbname=getSnowflakeDB()
  , showWarnings=TRUE
  , verbose=TRUE
) {

  if (overwrite_if_exists || check_if_exists) {
    ## IF SCHEMA ALREADY EXISTS
    if (sfSchemaExists(schema=schema, dbname=dbname, verbose=FALSE, wh=wh)) {
      if (overwrite_if_exists) {
        verboseMsg(verbose, "SCHEMA ", schema, " ALREADY EXISTS --- even though 'overwrite_if_exists' is TRUE, will not overwrite.\nYou can do so manually via the following query\n\t", time=FALSE)
        sprintf("DROP SCHEMA %s", dbschematbl(db=dbname, schema=schema, tbl=schema)) %>% catn("\n\t", ., "\n") # sfQry(wh=wh, dbname=dbname, schema=schema, verbose=FALSE)
        stop("Schema '", schema, "' already exists")
      } else {
        DT.tbls <- sfShowTables(schema=schema, dbname=dbname, showWarnings=FALSE, verbose=FALSE)
        verboseMsg(showWarnings, "\tSCHEMA ", schema, " ALREADY EXISTS. It's contents are as follows: \n\n", pasteC(capture.output(DT.tbls), C="\n"), time=FALSE)
        return(invisible(DT.tbls))
      }
    ## ELSE, if not exist, create it
    } else {
      verboseMsg(verbose, "Creating schema '", schema, "' in database '", dbname, "'", time=FALSE, sep="")
    }
  }

  ## OTHERWISE CREATE
  sprintf("
  CREATE SCHEMA %s
    COMMENT = '%s'
  ;"
  , dbschematbl(dbname=dbname, schema=schema, tbl=NULL)
  , pasteC(c(comment, comment_footer), C=" ")
  ) %>%
  sfQry(wh=wh, verbose=verbose)
}


sfCreateStage <- function(stage_name=name, bucket, format_name=NULL
  , comment=NULL
  , comment_footer=getSFCommentFooter()
  , aws_aki=getOption("aws_aki", default=stop("aws_aki not found"))
  , aws_sak=getOption("aws_sak", default=stop("aws_sak not found"))
  , check_if_exists=TRUE
  , overwrite_if_exists=FALSE
  , wh=getSnowflakeWH()
  , dbname=getSnowflakeDB()
  , schema=getSnowflakeSchema()
  , name
  , showWarnings=TRUE
  , verbose=TRUE
) {

  if (!missing(name))
    warning ("name is deprecated -- use stage_name instead")

  ## OPTIONS NOT USED, BEFORE COMMENT
  #     ON_ERROR = <string>
  #     PURGE = TRUE | FALSE
  #     RETURN_FAILED_ONLY = TRUE | FALSE
  #     SIZE_LIMIT = <number>

  if (overwrite_if_exists || check_if_exists) {
    current_vals <- sprintf("SHOW STAGES LIKE '%s' %s", stage_name, ifelseNULL(schema, "", paste("IN", schema))) %>% sfQry(wh=wh, dbname=dbname, schema=schema, verbose=FALSE)
    exists_already <- current_vals %>% nrow %>% as.logical
    if (overwrite_if_exists && exists_already) {
      verboseMsg(verbose, "STAGE ", stage_name, " ALREADY EXISTS. Will overwrite it.")
      sprintf("DROP STAGE %s", dbschematbl(db=dbname, schema=schema, tbl=stage_name)) %>% sfQry(wh=wh, dbname=dbname, schema=schema, verbose=FALSE)
    } else if (exists_already) {
      verboseMsg(verbose, "STAGE ", stage_name, " ALREADY EXISTS. It's contents are as follows: ")
      print(current_vals)
      return(invisible(current_vals))
    }
  }

  sprintf("
  CREATE STAGE %s
    URL = '%s'
    CREDENTIALS=(AWS_KEY_ID='%s' AWS_SECRET_KEY='%s')
    %s
    COMMENT = '%s'
  ;"
  , dbschematbl(dbname=dbname, schema=schema, tbl=stage_name)
  , bucket
  , aws_aki
  , aws_sak
  , ifelseNULL(format_name, "", sprintf("FILE_FORMAT = (FORMAT_NAME = '%s')", format_name))
  , pasteC(c(comment, comment_footer), C=" ")
  ) %>%
  sfQry(wh=wh, verbose=verbose)

}

if (FALSE) {
  sfCreateFileFormat.CSV("TEST", TRUE, TRUE, quote_string="None")
  sfCreateFileFormat.CSV("TEST", TRUE, TRUE, quote_string="'")
  sfCreateFileFormat.CSV("TEST", TRUE, TRUE, quote_string='"')
  sfCreateFileFormat.CSV("TEST", TRUE, TRUE, quote_string=NULL)
  sfCreateFileFormat.CSV("TEST", TRUE, TRUE)
}

sfCreateFileFormat.CSV <- function(format_name=name, header, gzip=FALSE, delimiter="\t", quote_string=c("NONE", "'", '"'), escape_string=c("'\\'", "NULL", "'NONE'"), trim_space=FALSE, NULL_IF=c(''), error_on_column_count_mismatch=TRUE, comment=NULL, comment_footer=getSFCommentFooter(), wh=getSnowflakeWH(), dbname=getSnowflakeDB(), schema=getSnowflakeSchema(), name, check_if_exists=TRUE, overwrite_if_exists=FALSE, verbose=TRUE) {
## I am not sure of NULL_IF
# Q.file_format

## TODO 2:  Incorportate these into sfCreateFileFormat.CSV
## OPTIONS 
# https://documentation.snowflakecomputing.com/manuals/sql-reference/sql/create-file-format.html
   # FIELD_DELIMITER = '<character>' | 'NONE'
   # RECORD_DELIMITER = '<character>' | 'NONE'
   # SKIP_HEADER = <integer>
   # DATE_FORMAT = '<string>' | 'AUTO'
   # TIMESTAMP_FORMAT = '<string>' | 'AUTO'
   # ESCAPE = '<character>' | 'NONE'
   # ESCAPE_UNENCLOSED_FIELD = '<character>' | 'NONE'
   # TRIM_SPACE = TRUE | FALSE
   # FIELD_OPTIONALLY_ENCLOSED_BY = '<character>' | 'NONE'
   # NULL_IF = ( '<string>' [, '<string>'] ... )
   # ERROR_ON_COLUMN_COUNT_MISMATCH = TRUE | FALSE
# 


  if (!missing(name))
    warning ("name is deprecated -- use format_name instead")

  if (missing(header))
    stop("'header' must be specified explicitly.  Set to either T/F or an integer of how many rows to skip")

  if (missing(gzip))
    stop ("gzip needs to be set explicitly to either TRUE or FALSE -- it used to default to TRUE now defaults to FALSE")

  if (!is.logical(trim_space))
    stop ("trim_space must be either TRUE or FALSE")

  ## Clean up and test escape_string
  {
    if (length(escape_string) > 1)
      escape_string <- escape_string[[1]]
    if (is.null(escape_string))
      escape_string <- "'NONE'"
    if (escape_string == "'")
      stop("escape_string is a single quote -- does that make sense?  If so, how do you pass it to snowflake?")
    escape_string %<>% removeText("'", .)
    if (nchar(escape_string) > 1 && !escape_string %in% c("NONE", "NULL", "\\\\"))
      stop("escape_string must be a single character or 'NONE' or NULL")
    if (escape_string == "\\")
      escape_string <- "\\\\"
    if (escape_string != "NULL")
      escape_string %<>% sprintf("'%s'", .)
    print(escape_string)
  }

  quote_string %<>% {ifelseNULL(., "NONE", match.arg(toupper(.), c("NONE", "'", '"')))} %>% {ifelse(. == "'", "0x27", .)}

  if (overwrite_if_exists || check_if_exists) {
    current_vals <- sprintf("SHOW FILE FORMATS LIKE '%s' %s", format_name, ifelseNULL(schema, "", paste("IN", schema))) %>% sfQry(wh=wh, dbname=dbname, schema=schema, verbose=FALSE)
    exists_already <- current_vals %>% nrow %>% as.logical
    if (overwrite_if_exists && exists_already) {
      verboseMsg(verbose, "FILE FORMAT ", format_name, " ALREADY EXISTS. Will overwrite it.")
      sprintf("DROP FILE FORMAT %s", dbschematbl(db=dbname, schema=schema, tbl=format_name)) %>% sfQry(wh=wh, dbname=dbname, schema=schema, verbose=FALSE)
    } else if (exists_already) {
      verboseMsg(verbose, "FILE FORMAT ", format_name, " ALREADY EXISTS. It's contents are as follows: ")
      print(current_vals)
      return(invisible(current_vals))
    }
  }

  if (quote_string == "NONE")
    message(sprintf("NOTE that if this format ('%s') will be used for exporting, a quote_string MUST be used", format_name))

  Q.create <- 
  sprintf("
  CREATE FILE FORMAT %s
    TYPE = 'CSV'
    SKIP_HEADER = %i
    FIELD_DELIMITER = '%s'
    COMPRESSION = '%s'
    FIELD_OPTIONALLY_ENCLOSED_BY = '%s'
    ESCAPE = %s
    TRIM_SPACE = %s
    %s
    ERROR_ON_COLUMN_COUNT_MISMATCH = %s
    COMMENT = '%s'
  ;"
  , dbschematbl(dbname=dbname, schema=schema, tbl=format_name)
  , as.integer(header)
  , delimiter
  , ifelse(isTRUE(gzip), "GZIP", "AUTO")
  , quote_string
  , escape_string
  , as.character(isTRUE(trim_space))
  , ifelseNULL(NULL_IF, "", paste0('NULL_IF = ', pasteQ(NULL_IF)))
  , as.character(isTRUE(error_on_column_count_mismatch))
  , pasteC(c(comment, comment_footer), C=" ")
  )

  ## set as Query
  setQry(Q.create)

  ## Execute
  sfQry(Q.create, wh=wh, verbose=verbose)
}