feat(stdlib/sql): add BigQuery support #2925
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,174 @@ | ||
| package sql | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "math/big" | ||
| "time" | ||
|
|
||
| "cloud.google.com/go/civil" | ||
| _ "github.com/bonitoo-io/go-sql-bigquery" | ||
| "github.com/influxdata/flux" | ||
| "github.com/influxdata/flux/codes" | ||
| "github.com/influxdata/flux/execute" | ||
| "github.com/influxdata/flux/internal/errors" | ||
| "github.com/influxdata/flux/values" | ||
| ) | ||
|
|
||
| // Google BigQuery support. | ||
| // Notes: | ||
| // * data types | ||
| // https://cloud.google.com/bigquery/docs/reference/standard-sql/data-types | ||
| // * connection string | ||
| // https://github.com/bonitoo-io/go-sql-bigquery#connection-string | ||
|
|
||
| type BigQueryRowReader struct { | ||
| Cursor *sql.Rows | ||
| columns []interface{} | ||
| columnTypes []flux.ColType | ||
| columnNames []string | ||
| sqlTypes []*sql.ColumnType | ||
| NextFunc func() bool | ||
| CloseFunc func() error | ||
| } | ||
|
|
||
| // Next prepares BigQueryRowReader to return rows | ||
| func (m *BigQueryRowReader) Next() bool { | ||
| if m.NextFunc != nil { | ||
| return m.NextFunc() | ||
| } | ||
| next := m.Cursor.Next() | ||
| if next { | ||
| columnNames, err := m.Cursor.Columns() | ||
| if err != nil { | ||
| return false | ||
| } | ||
| m.columns = make([]interface{}, len(columnNames)) | ||
| columnPointers := make([]interface{}, len(columnNames)) | ||
| for i := 0; i < len(columnNames); i++ { | ||
| columnPointers[i] = &m.columns[i] | ||
| } | ||
| if err := m.Cursor.Scan(columnPointers...); err != nil { | ||
| return false | ||
| } | ||
| } | ||
| return next | ||
| } | ||
|
|
||
| func (m *BigQueryRowReader) GetNextRow() ([]values.Value, error) { | ||
| row := make([]values.Value, len(m.columns)) | ||
| for i, column := range m.columns { | ||
| switch value := column.(type) { | ||
| case bool, int64, float64, string: | ||
| row[i] = values.New(value) | ||
| case civil.Date: | ||
| row[i] = values.NewString(value.String()) | ||
| case civil.DateTime: | ||
alespour marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| row[i] = values.NewString(value.String()) | ||
| case civil.Time: | ||
| row[i] = values.NewString(value.String()) | ||
| case time.Time: | ||
| row[i] = values.NewTime(values.ConvertTime(value)) | ||
| case *big.Int: | ||
| row[i] = values.NewInt(value.Int64()) | ||
| case *big.Float: | ||
| f, _ := value.Float64() | ||
| row[i] = values.NewFloat(f) | ||
| case *big.Rat: | ||
| f, _ := value.Float64() | ||
alespour marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| row[i] = values.NewFloat(f) | ||
| case nil: | ||
| row[i] = values.NewNull(flux.SemanticType(m.columnTypes[i])) | ||
| default: | ||
| execute.PanicUnknownType(flux.TInvalid) | ||
| } | ||
| } | ||
| return row, nil | ||
| } | ||
|
|
||
| func (m *BigQueryRowReader) InitColumnNames(names []string) { | ||
| m.columnNames = names | ||
| } | ||
|
|
||
| func (m *BigQueryRowReader) InitColumnTypes(types []*sql.ColumnType) { | ||
| fluxTypes := make([]flux.ColType, len(types)) | ||
| for i := 0; i < len(types); i++ { | ||
| switch types[i].DatabaseTypeName() { | ||
| case "INTEGER": | ||
| fluxTypes[i] = flux.TInt | ||
| case "FLOAT", "NUMERIC": | ||
| fluxTypes[i] = flux.TFloat | ||
| case "BOOLEAN": | ||
| fluxTypes[i] = flux.TBool | ||
| case "TIMESTAMP": // "DATE", "TIME" and "DATETIME" will be represented as string because TZ is unknown | ||
| fluxTypes[i] = flux.TTime | ||
| default: | ||
| fluxTypes[i] = flux.TString | ||
| } | ||
| } | ||
| m.columnTypes = fluxTypes | ||
| m.sqlTypes = types | ||
| } | ||
|
|
||
| func (m *BigQueryRowReader) ColumnNames() []string { | ||
| return m.columnNames | ||
| } | ||
|
|
||
| func (m *BigQueryRowReader) ColumnTypes() []flux.ColType { | ||
| return m.columnTypes | ||
| } | ||
|
|
||
| func (m *BigQueryRowReader) SetColumnTypes(types []flux.ColType) { | ||
| m.columnTypes = types | ||
| } | ||
|
|
||
| func (m *BigQueryRowReader) SetColumns(i []interface{}) { | ||
| m.columns = i | ||
| } | ||
|
|
||
| func (m *BigQueryRowReader) Close() error { | ||
| if m.CloseFunc != nil { | ||
| return m.CloseFunc() | ||
| } | ||
| if err := m.Cursor.Err(); err != nil { | ||
| return err | ||
| } | ||
| return m.Cursor.Close() | ||
| } | ||
|
|
||
| func NewBigQueryRowReader(r *sql.Rows) (execute.RowReader, error) { | ||
| reader := &BigQueryRowReader{ | ||
| Cursor: r, | ||
| } | ||
| cols, err := r.Columns() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| reader.InitColumnNames(cols) | ||
|
|
||
| types, err := r.ColumnTypes() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| reader.InitColumnTypes(types) | ||
|
|
||
| return reader, nil | ||
| } | ||
|
|
||
| var fluxToBigQuery = map[flux.ColType]string{ | ||
| flux.TFloat: "FLOAT64", | ||
| flux.TInt: "INT64", | ||
| flux.TString: "STRING", | ||
| flux.TBool: "BOOL", | ||
| flux.TTime: "TIMESTAMP", | ||
| } | ||
|
|
||
| // BigQueryTranslateColumn translates flux colTypes into their corresponding BigQuery column type | ||
| func BigQueryColumnTranslateFunc() translationFunc { | ||
| return func(f flux.ColType, colName string) (string, error) { | ||
| s, found := fluxToBigQuery[f] | ||
| if !found { | ||
| return "", errors.Newf(codes.Internal, "BigQuery does not support column type %s", f.String()) | ||
| } | ||
| return colName + " " + s, 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
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
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I see that you forked
solcatesrepository, but also merged some changes to there, does that mean we can use the originalgithub.com/solcates/go-sql-bigquerynow?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Since I wasn't getting a response from the maintainer who seem to have abandoned the project, I gave up on hoping on the merge and continued with modifications in the
bonitoo-iofork. The owner then commented later that he _ had totally forgotten about this project_ and merged the PR, but the fork has diverted more from the original before that, into more pure SQL golang driver unlike the original which also contained some ORM stuff etc.The original project does not seem to be active (no commits since Oct 27, 2019) and maintained (eg. PR was merge despite failing check due to some missing 3rd party lib).