snowflake: schema evolution #2982
Merged
Conversation
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 is a mechanism that can be used to evolve the schema when extra data is coming through
Will be needed for the column name we use for schema evolution
Enable schema evolution using a custom mapping function, to allow customers to customize the evolution method. Schema evolution needs to be a global pause so that we can reopen channels to get new schemas. I'm not in love with how errors are being hijacked here but I have not came up with a better idea in the short time working on.
rockwotj
force-pushed
the
pikachu-is-evolving
branch
from
eb6c7d8 to
2582de9
Compare
rockwotj
force-pushed
the
pikachu-is-evolving
branch
from
f97e467 to
f08d9f8
Compare
mihaitodor
approved these changes
Nov 6, 2024
There was a problem hiding this comment.
Thank you @rockwotj! LGTM, just left a few small nitpicks. Feel free to 🐑 🚀
| import "fmt" | ||
|
|
||
| // SchemaMismatchError occurs when the user provided data has data that | ||
| // doesn't match the schema *and* the table can be evolved to accomidate |
| func quoteColumnName(name string) string { | ||
| var quoted strings.Builder | ||
| // Default to assume we're just going to add quotes and there won't | ||
| // be any double quotes inside the string that need escaped. |
Comment on lines
562
to
566
| return fmt.Errorf("unable to compute new column type for %s: %w", col.ColumnName(), err) | ||
| } | ||
| v, err := out.AsBytes() | ||
| if err != nil { | ||
| return fmt.Errorf("unable to compute new column type for %s: %w", col.ColumnName(), err) |
There was a problem hiding this comment.
Can't distinguish between these errors
Comment on lines
573
to
593
| _, err = o.restClient.RunSQL(ctx, streaming.RunSQLRequest{ | ||
| // This looks very scary and it *should*. This is prone to SQL injection attacks. The column name is | ||
| // quoted according to the rules in Snowflake's documentation. This is also why we need to | ||
| // validate the data type, so that you can't sneak an injection attack in there. | ||
| Statement: fmt.Sprintf(`ALTER TABLE IDENTIFIER(?) | ||
| ADD COLUMN IF NOT EXISTS %s %s | ||
| COMMENT 'column created by schema evolution from Redpanda Connect'`, | ||
| col.ColumnName(), | ||
| columnType, | ||
| ), | ||
| // Currently we set of timeout of 30 seconds so that we don't have to handle async operations | ||
| // that need polling to wait until they finish (results are made async when execution is longer | ||
| // than 45 seconds). | ||
| Timeout: 30, | ||
| Database: o.db, | ||
| Schema: o.schema, | ||
| Role: o.role, | ||
| Bindings: map[string]streaming.BindingValue{ | ||
| "1": {Type: "TEXT", Value: o.table}, | ||
| }, | ||
| }) |
There was a problem hiding this comment.
Could we have a wrapper func so we don't duplicate this code & comments in MigrateNotNullColumn()?
There was a problem hiding this comment.
Yes - thanks it's much easier to read now :)
Good comments as always!
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
Support schema evolution for Redpanda Connect's new Snowflake output.