-
Notifications
You must be signed in to change notification settings - Fork 30
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[DDL] Moving DDL queries to
ddl.go (#1053)
- Loading branch information
Showing
10 changed files
with
157 additions
and
136 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,14 +1,47 @@ | ||
| package dialect | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
| "time" | ||
|
|
||
| "github.com/artie-labs/transfer/lib/config/constants" | ||
| "github.com/artie-labs/transfer/lib/sql" | ||
| "github.com/artie-labs/transfer/lib/typing" | ||
| ) | ||
|
|
||
| func (BigQueryDialect) BuildCreateTableQuery(tableID sql.TableIdentifier, temporary bool, colSQLParts []string) string { | ||
| query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s)", tableID.FullyQualifiedName(), strings.Join(colSQLParts, ",")) | ||
|
|
||
| if temporary { | ||
| return fmt.Sprintf( | ||
| `%s OPTIONS (expiration_timestamp = TIMESTAMP("%s"))`, | ||
| query, | ||
| BQExpiresDate(time.Now().UTC().Add(constants.TemporaryTableTTL)), | ||
| ) | ||
| } else { | ||
| return query | ||
| } | ||
| } | ||
|
|
||
| func (bd BigQueryDialect) BuildAddColumnQuery(tableID sql.TableIdentifier, sqlPart string) string { | ||
| return bd.buildAlterColumnQuery(tableID, constants.Add, sqlPart) | ||
| } | ||
|
|
||
| func (bd BigQueryDialect) BuildDropColumnQuery(tableID sql.TableIdentifier, colName string) string { | ||
| return bd.buildAlterColumnQuery(tableID, constants.Delete, colName) | ||
| } | ||
|
|
||
| func (BigQueryDialect) buildAlterColumnQuery(tableID sql.TableIdentifier, columnOp constants.ColumnOperation, colSQLPart string) string { | ||
| return fmt.Sprintf("ALTER TABLE %s %s COLUMN %s", tableID.FullyQualifiedName(), columnOp, colSQLPart) | ||
| } | ||
|
|
||
| func (BigQueryDialect) BuildDescribeTableQuery(tableID sql.TableIdentifier) (string, []interface{}, error) { | ||
| bqTableID, err := typing.AssertType[TableIdentifier](tableID) | ||
| if err != nil { | ||
| return "", nil, err | ||
| } | ||
|
|
||
| query := fmt.Sprintf("SELECT column_name, data_type, description FROM `%s.INFORMATION_SCHEMA.COLUMN_FIELD_PATHS` WHERE table_name = ?;", bqTableID.Dataset()) | ||
| return query, []any{bqTableID.Table()}, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,14 +1,30 @@ | ||
| package dialect | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
|
|
||
| "github.com/artie-labs/transfer/lib/config/constants" | ||
| "github.com/artie-labs/transfer/lib/sql" | ||
| ) | ||
|
|
||
| func (DatabricksDialect) BuildCreateTableQuery(tableID sql.TableIdentifier, _ bool, colSQLParts []string) string { | ||
| // Databricks doesn't have a concept of temporary tables. | ||
| return fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s)", tableID.FullyQualifiedName(), strings.Join(colSQLParts, ", ")) | ||
| } | ||
|
|
||
| func (d DatabricksDialect) BuildAddColumnQuery(tableID sql.TableIdentifier, sqlPart string) string { | ||
| return d.buildAlterColumnQuery(tableID, constants.Add, sqlPart) | ||
| } | ||
|
|
||
| func (d DatabricksDialect) BuildDropColumnQuery(tableID sql.TableIdentifier, colName string) string { | ||
| return d.buildAlterColumnQuery(tableID, constants.Delete, colName) | ||
| } | ||
|
|
||
| func (DatabricksDialect) BuildDescribeTableQuery(tableID sql.TableIdentifier) (string, []any, error) { | ||
| return fmt.Sprintf("DESCRIBE TABLE %s", tableID.FullyQualifiedName()), nil, nil | ||
| } | ||
|
|
||
| func (DatabricksDialect) buildAlterColumnQuery(tableID sql.TableIdentifier, columnOp constants.ColumnOperation, colSQLPart string) string { | ||
| return fmt.Sprintf("ALTER TABLE %s %s COLUMN %s", tableID.FullyQualifiedName(), columnOp, colSQLPart) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,14 +1,54 @@ | ||
| package dialect | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
|
|
||
| mssql "github.com/microsoft/go-mssqldb" | ||
|
|
||
| "github.com/artie-labs/transfer/lib/config/constants" | ||
| "github.com/artie-labs/transfer/lib/sql" | ||
| "github.com/artie-labs/transfer/lib/typing" | ||
| ) | ||
|
|
||
| func (MSSQLDialect) BuildDescribeTableQuery(tableID sql.TableIdentifier) (string, []any, error) { | ||
| mssqlTableID, err := typing.AssertType[TableIdentifier](tableID) | ||
| if err != nil { | ||
| return "", nil, err | ||
| } | ||
|
|
||
| return ` | ||
| SELECT | ||
| COLUMN_NAME, | ||
| CASE | ||
| WHEN DATA_TYPE = 'numeric' THEN | ||
| 'numeric(' + COALESCE(CAST(NUMERIC_PRECISION AS VARCHAR), '') + ',' + COALESCE(CAST(NUMERIC_SCALE AS VARCHAR), '') + ')' | ||
| ELSE | ||
| DATA_TYPE | ||
| END AS DATA_TYPE, | ||
| CHARACTER_MAXIMUM_LENGTH, | ||
| COLUMN_DEFAULT AS DEFAULT_VALUE | ||
| FROM | ||
| INFORMATION_SCHEMA.COLUMNS | ||
| WHERE | ||
| LOWER(TABLE_NAME) = LOWER(?) AND LOWER(TABLE_SCHEMA) = LOWER(?);`, []any{mssql.VarChar(mssqlTableID.Table()), mssql.VarChar(mssqlTableID.Schema())}, nil | ||
| } | ||
|
|
||
| func (md MSSQLDialect) BuildAddColumnQuery(tableID sql.TableIdentifier, sqlPart string) string { | ||
| return md.buildAlterColumnQuery(tableID, constants.Add, sqlPart) | ||
| } | ||
|
|
||
| func (md MSSQLDialect) BuildDropColumnQuery(tableID sql.TableIdentifier, colName string) string { | ||
| return md.buildAlterColumnQuery(tableID, constants.Delete, colName) | ||
| } | ||
|
|
||
| func (MSSQLDialect) BuildCreateTableQuery(tableID sql.TableIdentifier, _ bool, colSQLParts []string) string { | ||
| // Microsoft SQL Server uses the same syntax for temporary and permanent tables. | ||
| // Microsoft SQL Server doesn't support IF NOT EXISTS | ||
| return fmt.Sprintf("CREATE TABLE %s (%s);", tableID.FullyQualifiedName(), strings.Join(colSQLParts, ",")) | ||
| } | ||
|
|
||
| func (MSSQLDialect) buildAlterColumnQuery(tableID sql.TableIdentifier, columnOp constants.ColumnOperation, colSQLPart string) string { | ||
| // Microsoft SQL Server doesn't support the COLUMN keyword | ||
| return fmt.Sprintf("ALTER TABLE %s %s %s", tableID.FullyQualifiedName(), columnOp, colSQLPart) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,14 +1,58 @@ | ||
| package dialect | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
|
|
||
| "github.com/artie-labs/transfer/lib/config/constants" | ||
| "github.com/artie-labs/transfer/lib/sql" | ||
| "github.com/artie-labs/transfer/lib/typing" | ||
| ) | ||
|
|
||
| func (RedshiftDialect) BuildDescribeTableQuery(tableID sql.TableIdentifier) (string, []any, error) { | ||
| redshiftTableID, err := typing.AssertType[TableIdentifier](tableID) | ||
| if err != nil { | ||
| return "", nil, err | ||
| } | ||
|
|
||
| // This query is a modified fork from: https://gist.github.com/alexanderlz/7302623 | ||
| return fmt.Sprintf(` | ||
| SELECT | ||
| c.column_name, | ||
| CASE | ||
| WHEN c.data_type = 'numeric' THEN | ||
| 'numeric(' || COALESCE(CAST(c.numeric_precision AS VARCHAR), '') || ',' || COALESCE(CAST(c.numeric_scale AS VARCHAR), '') || ')' | ||
| ELSE | ||
| c.data_type | ||
| END AS data_type, | ||
| c.%s, | ||
| d.description | ||
| FROM | ||
| INFORMATION_SCHEMA.COLUMNS c | ||
| LEFT JOIN | ||
| PG_CLASS c1 ON c.table_name = c1.relname | ||
| LEFT JOIN | ||
| PG_CATALOG.PG_NAMESPACE n ON c.table_schema = n.nspname AND c1.relnamespace = n.oid | ||
| LEFT JOIN | ||
| PG_CATALOG.PG_DESCRIPTION d ON d.objsubid = c.ordinal_position AND d.objoid = c1.oid | ||
| WHERE | ||
| LOWER(c.table_schema) = LOWER($1) AND LOWER(c.table_name) = LOWER($2); | ||
| `, constants.StrPrecisionCol), []any{redshiftTableID.Schema(), redshiftTableID.Table()}, nil | ||
| } | ||
|
|
||
| func (rd RedshiftDialect) BuildAddColumnQuery(tableID sql.TableIdentifier, sqlPart string) string { | ||
| return rd.buildAlterColumnQuery(tableID, constants.Add, sqlPart) | ||
| } | ||
|
|
||
| func (rd RedshiftDialect) BuildDropColumnQuery(tableID sql.TableIdentifier, colName string) string { | ||
| return rd.buildAlterColumnQuery(tableID, constants.Delete, colName) | ||
| } | ||
|
|
||
| func (RedshiftDialect) BuildCreateTableQuery(tableID sql.TableIdentifier, _ bool, colSQLParts []string) string { | ||
| // Redshift uses the same syntax for temporary and permanent tables. | ||
| return fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s);", tableID.FullyQualifiedName(), strings.Join(colSQLParts, ",")) | ||
| } | ||
|
|
||
| func (RedshiftDialect) buildAlterColumnQuery(tableID sql.TableIdentifier, columnOp constants.ColumnOperation, colSQLPart string) string { | ||
| return fmt.Sprintf("ALTER TABLE %s %s COLUMN %s", tableID.FullyQualifiedName(), columnOp, colSQLPart) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,14 +1,38 @@ | ||
| package dialect | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "strings" | ||
|
|
||
| "github.com/artie-labs/transfer/lib/config/constants" | ||
| "github.com/artie-labs/transfer/lib/sql" | ||
| ) | ||
|
|
||
| func (SnowflakeDialect) BuildCreateTableQuery(tableID sql.TableIdentifier, temporary bool, colSQLParts []string) string { | ||
| query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (%s)", tableID.FullyQualifiedName(), 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 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 query | ||
| } | ||
| } | ||
|
|
||
| func (sd SnowflakeDialect) BuildAddColumnQuery(tableID sql.TableIdentifier, sqlPart string) string { | ||
| return sd.buildAlterColumnQuery(tableID, constants.Add, sqlPart) | ||
| } | ||
|
|
||
| func (sd SnowflakeDialect) BuildDropColumnQuery(tableID sql.TableIdentifier, colName string) string { | ||
| return sd.buildAlterColumnQuery(tableID, constants.Delete, colName) | ||
| } | ||
|
|
||
| func (SnowflakeDialect) BuildDescribeTableQuery(tableID sql.TableIdentifier) (string, []any, error) { | ||
| return fmt.Sprintf("DESC TABLE %s", tableID.FullyQualifiedName()), nil, nil | ||
| } | ||
|
|
||
| func (SnowflakeDialect) buildAlterColumnQuery(tableID sql.TableIdentifier, columnOp constants.ColumnOperation, colSQLPart string) string { | ||
| return fmt.Sprintf("ALTER TABLE %s %s COLUMN %s", tableID.FullyQualifiedName(), columnOp, colSQLPart) | ||
| } |
Oops, something went wrong.