Skip to content

Commit

Permalink
Factor
Browse files Browse the repository at this point in the history
  • Loading branch information
nathan-artie committed May 10, 2024
1 parent 10efccf commit ae95ca8
Show file tree
Hide file tree
Showing 2 changed files with 12 additions and 6 deletions.
11 changes: 8 additions & 3 deletions lib/sql/bigquery.go
Original file line number Diff line number Diff line change
Expand Up @@ -118,11 +118,16 @@ func (BigQueryDialect) IsColumnAlreadyExistsErr(err error) bool {
}

func (BigQueryDialect) BuildCreateTableQuery(fqTableName string, temporary bool, colSQLParts []string) string {
query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s)", fqTableName, strings.Join(colSQLParts, ","))

if temporary {
return fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (%s) OPTIONS (expiration_timestamp = TIMESTAMP("%s"))`,
fqTableName, strings.Join(colSQLParts, ","), BQExpiresDate(time.Now().UTC().Add(constants.TemporaryTableTTL)))
return fmt.Sprintf(
`%s OPTIONS (expiration_timestamp = TIMESTAMP("%s"))`,
query,
BQExpiresDate(time.Now().UTC().Add(constants.TemporaryTableTTL)),
)
} else {
return fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s)", fqTableName, strings.Join(colSQLParts, ","))
return query
}
}

Expand Down
7 changes: 4 additions & 3 deletions lib/sql/snowflake.go
Original file line number Diff line number Diff line change
Expand Up @@ -115,14 +115,15 @@ func (SnowflakeDialect) IsColumnAlreadyExistsErr(err error) bool {
}

func (SnowflakeDialect) BuildCreateTableQuery(fqTableName string, temporary bool, colSQLParts []string) string {
query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s)", fqTableName, strings.Join(colSQLParts, ","))

if temporary {
// TEMPORARY Table syntax - https://docs.snowflake.com/en/sql-reference/sql/create-table
// PURGE syntax - https://docs.snowflake.com/en/sql-reference/sql/copy-into-table#purging-files-after-loading
// FIELD_OPTIONALLY_ENCLOSED_BY - is needed because CSV will try to escape any values that have `"`
return fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (%s) STAGE_COPY_OPTIONS = ( PURGE = TRUE ) STAGE_FILE_FORMAT = ( TYPE = 'csv' FIELD_DELIMITER= '\t' FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF='\\N' EMPTY_FIELD_AS_NULL=FALSE)`,
fqTableName, strings.Join(colSQLParts, ","))
return query + ` STAGE_COPY_OPTIONS = ( PURGE = TRUE ) STAGE_FILE_FORMAT = ( TYPE = 'csv' FIELD_DELIMITER= '\t' FIELD_OPTIONALLY_ENCLOSED_BY='"' NULL_IF='\\N' EMPTY_FIELD_AS_NULL=FALSE)`
} else {
return fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s)", fqTableName, strings.Join(colSQLParts, ","))
return query
}
}

Expand Down

0 comments on commit ae95ca8

Please sign in to comment.