-
Notifications
You must be signed in to change notification settings - Fork 8
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
2cbe98e
commit 416a3dc
Showing
18 changed files
with
383 additions
and
144 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
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
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
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 |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| package provider | ||
|
|
||
| // Interface implements the required functionality for a Provider. | ||
| type Interface interface { | ||
| GetSelectQueryForTable(table string, params DumpParams) (string, error) | ||
| } | ||
|
|
||
| // DumpParams is used to pass parameters to the Dump function. | ||
| type DumpParams struct { | ||
| SelectMap map[string]map[string]string | ||
| WhereMap map[string]string | ||
| FilterMap map[string]string | ||
| UseTableLock bool | ||
| ExtendedInsertRows int | ||
| } |
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 |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| package rds | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "fmt" | ||
| "log" | ||
| "strings" | ||
|
|
||
| "github.com/skpr/mtk/internal/mysql/provider" | ||
| providerutils "github.com/skpr/mtk/internal/mysql/provider/utils" | ||
| ) | ||
|
|
||
| // Client used for dumping a database and/or table. | ||
| type Client struct { | ||
| provider.Interface | ||
| DB *sql.DB | ||
| Logger *log.Logger | ||
|
|
||
| Region string // Region configuration | ||
| URI string // S3 URI configuration | ||
| } | ||
|
|
||
| // NewClient for dumping a full or single table from a database. | ||
| func NewClient(db *sql.DB, logger *log.Logger, region, uri string) *Client { | ||
| return &Client{ | ||
| DB: db, | ||
| Logger: logger, | ||
| Region: region, | ||
| URI: uri, | ||
| } | ||
| } | ||
|
|
||
| // GetSelectQueryForTable will return a complete SELECT query to export data from a table. | ||
| func (d *Client) GetSelectQueryForTable(table string, params provider.DumpParams) (string, error) { | ||
| cols, err := providerutils.QueryColumnsForTable(d.DB, table, params) | ||
| if err != nil { | ||
| return "", err | ||
| } | ||
|
|
||
| query := fmt.Sprintf("SELECT %s", strings.Join(cols, ", ")) | ||
| query = fmt.Sprintf("%s FROM `%s`", query, table) | ||
|
|
||
| if where, ok := params.WhereMap[strings.ToLower(table)]; ok { | ||
| query = fmt.Sprintf("%s WHERE %s", query, where) | ||
| } | ||
|
|
||
| query = fmt.Sprintf("%s INTO OUTFILE S3 '%s/%s.csv'", query, d.URI, table) | ||
| query = fmt.Sprintf("%s FIELDS TERMINATED BY ',' ENCLOSED BY '\"' LINES TERMINATED BY '\\n'", query) | ||
| query = fmt.Sprintf("%s MANIFEST ON", query) | ||
| query = fmt.Sprintf("%s OVERWRITE ON", query) | ||
|
|
||
| importQuery, err := d.GetLoadQueryForTable(table) | ||
| if err != nil { | ||
| return "", err | ||
| } | ||
|
|
||
| fmt.Println(importQuery) | ||
| return query, nil | ||
| } | ||
|
|
||
| // GetLoadQueryForTable will return a complete SELECT query to fetch data from a table. | ||
| func (d *Client) GetLoadQueryForTable(table string) (string, error) { | ||
| if table == "" { | ||
| return "", fmt.Errorf("error: no table specified") | ||
| } | ||
| if d.Region == "" || len(strings.Split(d.Region, "-")) != 3 { | ||
| return "", fmt.Errorf("error: region is not configured correctly") | ||
| } | ||
| path := strings.TrimPrefix(d.URI, "s3://") | ||
| query := fmt.Sprintf("LOAD DATA FROM S3 MANIFEST 'S3-%s://%s/%s.csv.manifest' INTO TABLE `%s`", d.Region, path, table, table) | ||
| query = fmt.Sprintf("%s FIELDS TERMINATED BY ',' ENCLOSED BY '\"' LINES TERMINATED BY '\\n'", query) | ||
|
|
||
| return query, 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,35 @@ | ||
| package rds | ||
|
|
||
| import ( | ||
| "log" | ||
| "os" | ||
| "testing" | ||
|
|
||
| "github.com/DATA-DOG/go-sqlmock" | ||
| "github.com/stretchr/testify/assert" | ||
|
|
||
| "github.com/skpr/mtk/internal/mysql/mock" | ||
| "github.com/skpr/mtk/internal/mysql/provider" | ||
| ) | ||
|
|
||
| func TestMySQLGetExportSelectQueryFor(t *testing.T) { | ||
| db, mock := mock.GetDB(t) | ||
| dumper := NewClient(db, log.New(os.Stdout, "", 0), "ap-southheast-2", "s3://path/to/bucket") | ||
| mock.ExpectQuery("SELECT \\* FROM `table` LIMIT 1").WillReturnRows( | ||
| sqlmock.NewRows([]string{"c1", "c2"}).AddRow("a", "b")) | ||
| query, err := dumper.GetSelectQueryForTable("table", provider.DumpParams{ | ||
| SelectMap: map[string]map[string]string{"table": {"c2": "NOW()"}}, | ||
| WhereMap: map[string]string{"table": "c1 > 0"}, | ||
| }) | ||
| assert.Nil(t, err) | ||
| assert.Equal(t, "SELECT `c1`, NOW() AS `c2` FROM `table` WHERE c1 > 0 INTO OUTFILE S3 's3://path/to/bucket/table.csv' FIELDS TERMINATED BY ',' ENCLOSED BY '\"' LINES TERMINATED BY '\\n' MANIFEST ON OVERWRITE ON", query) | ||
| } | ||
|
|
||
| func TestMySQLGetLoadQueryFor(t *testing.T) { | ||
| db, _ := mock.GetDB(t) | ||
| dumper := NewClient(db, log.New(os.Stdout, "", 0), "ap-southeast-4", "s3://path/to/bucket") | ||
| query, err := dumper.GetLoadQueryForTable("table_name") | ||
| assert.Nil(t, err) | ||
| assert.Equal(t, "LOAD DATA FROM S3 MANIFEST 'S3-ap-southeast-4://path/to/bucket/table_name.csv.manifest' INTO TABLE `table_name` FIELDS TERMINATED BY ',' ENCLOSED BY '\"' LINES TERMINATED BY '\\n'", query) | ||
|
|
||
| } |
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 |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| package stdout | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "fmt" | ||
| "log" | ||
| "strings" | ||
|
|
||
| "github.com/skpr/mtk/internal/mysql/provider" | ||
| providerutils "github.com/skpr/mtk/internal/mysql/provider/utils" | ||
| ) | ||
|
|
||
| // Client used for dumping a database and/or table. | ||
| type Client struct { | ||
| provider.Interface | ||
| DB *sql.DB | ||
| Logger *log.Logger | ||
| } | ||
|
|
||
| // NewClient for dumping a full or single table from a database. | ||
| func NewClient(db *sql.DB, logger *log.Logger) *Client { | ||
| return &Client{ | ||
| DB: db, | ||
| Logger: logger, | ||
| } | ||
| } | ||
|
|
||
| // GetSelectQueryForTable will return a complete SELECT query to fetch data from a table. | ||
| func (d *Client) GetSelectQueryForTable(table string, params provider.DumpParams) (string, error) { | ||
| cols, err := providerutils.QueryColumnsForTable(d.DB, table, params) | ||
| if err != nil { | ||
| return "", err | ||
| } | ||
|
|
||
| query := fmt.Sprintf("SELECT %s FROM `%s`", strings.Join(cols, ", "), table) | ||
|
|
||
| if where, ok := params.WhereMap[strings.ToLower(table)]; ok { | ||
| query = fmt.Sprintf("%s WHERE %s", query, where) | ||
| } | ||
|
|
||
| return query, nil | ||
| } |
Oops, something went wrong.