Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

materialize-iceberg: column type migrations #2540

Open
wants to merge 4 commits into
base: main
Choose a base branch
from
Open

materialize-iceberg: column type migrations #2540

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
97ba330
materializer: support for column migrations
williamhbaker Mar 16, 2025
55f9144
materialize-iceberg: make integration tests work with concurrent CI runs
williamhbaker Mar 16, 2025
c1ea82c
materialize-iceberg: get rid of `getHighestFieldID`
williamhbaker Mar 16, 2025
ded4a69
materialize-iceberg: column type migrations
williamhbaker Mar 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
188 changes: 123 additions & 65 deletions materialize-boilerplate/materializer.go
View file
Open in desktop

Large diffs are not rendered by default.

32 changes: 32 additions & 0 deletions materialize-boilerplate/type_mapping.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,20 @@ func (FlatTypeString) isFlatType() {}
func (FlatTypeStringFormatInteger) isFlatType() {}
func (FlatTypeStringFormatNumber) isFlatType() {}

type MappedTyper interface {
// String produces a human readable description of the mapped type, which
// will be included validation constraint descriptions.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
// will be included validation constraint descriptions.
// will be included in validation constraint descriptions.

String() string

// Compatible determines if an existing materialized field as reported in
// the InfoSchema is compatible with a proposed mapped type.
Compatible(ExistingField) bool

// CanMigrate determines if an existing materialized field can be migrated
// to be compatible with the updated mapped type.
CanMigrate(ExistingField) bool
}

// Projection lifts a pf.Projection into a form that's more easily worked with
// for materialization-specific type mapping.
type Projection struct {
Expand All @@ -78,6 +92,24 @@ type Projection struct {
MustExist bool
}

const AnyExistingType string = "*"

// TypeMigrations is a utility struct for defining which existing fields can be
// migrated to be compatible with updated projections.
type TypeMigrations[T comparable] map[string][]T

func (m TypeMigrations[T]) CanMigrate(from string, to T) bool {
for f, ts := range m {
if f == from || f == AnyExistingType {
if slices.Contains(ts, to) {
return true
}
}
}

return false
}

func mapProjection(p pf.Projection, fc FieldConfiger) Projection {
mustExist := p.Inference.Exists == pf.Inference_MUST && !slices.Contains(p.Inference.Types, "null")
typesWithoutNull := getTypesWithoutNull(p.Inference.Types)
Expand Down
42 changes: 42 additions & 0 deletions materialize-iceberg/.snapshots/TestComputeSchemas
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,42 @@
--- Original Schema ---
table {
1: firstKey: required long
2: secondKey: required string
3: val1: required string
4: val2: optional boolean
5: dateToStr: required date
6: intToDecimal: optional long
7: decimalToFloat: required decimal(38, 0)
8: timestampToStr: required timestamptz
10: dateToStr_flow_tmp: required double
}

--- Next Schema ---
table {
1: firstKey: required long
2: secondKey: required string
3: val1: required string
4: val2: optional boolean
5: dateToStr: optional date
6: intToDecimal: optional long
7: decimalToFloat: required decimal(38, 0)
8: timestampToStr: required timestamptz
13: new: optional string
14: dateToStr_flow_tmp: optional string
15: intToDecimal_flow_tmp: optional decimal(38, 0)
16: decimalToFloat_flow_tmp: required double
17: timestampToStr_flow_tmp: optional string
}

--- After Migrate Schema ---
table {
1: firstKey: required long
2: secondKey: required string
3: val1: required string
4: val2: optional boolean
13: new: optional string
14: dateToStr: optional string
15: intToDecimal: optional decimal(38, 0)
16: decimalToFloat: required double
17: timestampToStr: optional string
}
16 changes: 12 additions & 4 deletions materialize-iceberg/.snapshots/TestTemplates
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ SELECT 0, l.`flow_document`
FROM `foo`.`bar` AS l
JOIN load_view_0 AS r
ON l.`first-key` = r.`first-key` AND l.`first-key` >= 1 AND l.`first-key` <= 10
AND l.`second-key` = unbase64(r.`second-key`)
AND l.`second-key` = UNBASE64(r.`second-key`)
AND l.`third-key` = r.`third-key` AND l.`third-key` >= 'aaaSomeString' AND l.`third-key` <= 'zzzSomeString'
--- End load query ---

Expand All @@ -12,9 +12,17 @@ MERGE INTO `foo`.`bar` AS l
USING merge_view_0 AS r
ON
l.`first-key` = r.`first-key` AND l.`first-key` >= 1 AND l.`first-key` <= 10
AND l.`second-key` = unbase64(r.`second-key`)
AND l.`second-key` = UNBASE64(r.`second-key`)
AND l.`third-key` = r.`third-key` AND l.`third-key` >= 'aaaSomeString' AND l.`third-key` <= 'zzzSomeString'
WHEN MATCHED AND r.`flow_document` = '"delete"' THEN DELETE
WHEN MATCHED THEN UPDATE SET l.`first-key` = r.`first-key`, l.`second-key` = unbase64(r.`second-key`), l.`third-key` = r.`third-key`, l.`first-val` = r.`first-val`, l.`second-val` = r.`second-val`, l.`third-val` = r.`third-val`, l.`fourth-val` = r.`fourth-val`, l.`flow_document` = r.`flow_document`
WHEN NOT MATCHED AND r.`flow_document` != '"delete"' THEN INSERT (`first-key`, `second-key`, `third-key`, `first-val`, `second-val`, `third-val`, `fourth-val`, `flow_document`) VALUES (r.`first-key`, unbase64(r.`second-key`), r.`third-key`, r.`first-val`, r.`second-val`, r.`third-val`, r.`fourth-val`, r.`flow_document`)
WHEN MATCHED THEN UPDATE SET l.`first-key` = r.`first-key`, l.`second-key` = UNBASE64(r.`second-key`), l.`third-key` = r.`third-key`, l.`first-val` = r.`first-val`, l.`second-val` = r.`second-val`, l.`third-val` = UNBASE64(r.`third-val`), l.`fourth-val` = r.`fourth-val`, l.`flow_document` = r.`flow_document`
WHEN NOT MATCHED AND r.`flow_document` != '"delete"' THEN INSERT (`first-key`, `second-key`, `third-key`, `first-val`, `second-val`, `third-val`, `fourth-val`, `flow_document`) VALUES (r.`first-key`, UNBASE64(r.`second-key`), r.`third-key`, r.`first-val`, r.`second-val`, UNBASE64(r.`third-val`), r.`fourth-val`, r.`flow_document`)
--- End merge query ---

--- Begin migrate query ---
UPDATE `some`.`table`
SET
`long_to_decimal_flow_tmp` = CAST(`long_to_decimal` AS decimal(38, 0)),
`datetime_to_string_flow_tmp` = DATE_FORMAT(`datetime_to_string`, 'yyyy-MM-dd\'T\'HH:mm:ss.SSSSSS\'Z\''),
`binary_to_string_flow_tmp` = BASE64(`binary_to_string`)
--- End migrate query ---
30 changes: 15 additions & 15 deletions materialize-iceberg/.snapshots/TestValidateAndApply
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,34 +82,34 @@ Big Schema Changed Types Constraints:
{"Field":"boolField","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'boolField' is already being materialized as endpoint type 'BOOLEAN' but endpoint type 'LONG' is required by its schema '{ type: [integer] }'"}
{"Field":"flow_document","Type":1,"TypeString":"FIELD_REQUIRED","Reason":"This field is the document in the current materialization"}
{"Field":"flow_published_at","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"intField","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'intField' is already being materialized as endpoint type 'LONG' but endpoint type 'STRING' is required by its schema '{ type: [string] }'"}
{"Field":"intField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"key","Type":1,"TypeString":"FIELD_REQUIRED","Reason":"This field is a key in the current materialization"}
{"Field":"multipleField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"nullField","Type":4,"TypeString":"FIELD_OPTIONAL","Reason":"Object fields may be materialized"}
{"Field":"numField","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'numField' is already being materialized as endpoint type 'DOUBLE' but endpoint type 'BOOLEAN' is required by its schema '{ type: [boolean] }'"}
{"Field":"objField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringDateField","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'stringDateField' is already being materialized as endpoint type 'DATE' but endpoint type 'STRING' is required by its schema '{ type: [string] }'"}
{"Field":"stringDateTimeField","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'stringDateTimeField' is already being materialized as endpoint type 'TIMESTAMPTZ' but endpoint type 'STRING' is required by its schema '{ type: [string] }'"}
{"Field":"stringDateField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringDateTimeField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringDurationField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringEmailField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringField","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'stringField' is already being materialized as endpoint type 'STRING' but endpoint type 'LONG' is required by its schema '{ type: [integer] }'"}
{"Field":"stringHostnameField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringIdnEmailField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringIdnHostnameField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringIntegerField","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'stringIntegerField' is already being materialized as endpoint type 'DECIMAL(38, 0)' but endpoint type 'STRING' is required by its schema '{ type: [string] }'"}
{"Field":"stringIntegerField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringIpv4Field","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringIpv6Field","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringIriField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringIriReferenceField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringJsonPointerField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringMacAddr8Field","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringMacAddrField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringNumberField","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'stringNumberField' is already being materialized as endpoint type 'DOUBLE' but endpoint type 'STRING' is required by its schema '{ type: [string] }'"}
{"Field":"stringNumberField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringRegexField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringRelativeJsonPointerField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringTimeField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringUint32Field","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'stringUint32Field' is already being materialized as endpoint type 'DECIMAL(38, 0)' but endpoint type 'STRING' is required by its schema '{ type: [string] }'"}
{"Field":"stringUint64Field","Type":6,"TypeString":"UNSATISFIABLE","Reason":"Field 'stringUint64Field' is already being materialized as endpoint type 'DECIMAL(38, 0)' but endpoint type 'STRING' is required by its schema '{ type: [string] }'"}
{"Field":"stringUint32Field","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringUint64Field","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringUriField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringUriReferenceField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
{"Field":"stringUriTemplateField","Type":3,"TypeString":"LOCATION_RECOMMENDED","Reason":"This location is part of the current materialization"}
Expand Down Expand Up @@ -152,7 +152,7 @@ table {
33: stringUriReferenceField: required string (auto-generated projection of JSON at: /stringUriReferenceField with inferred types: [string])
34: stringUriTemplateField: required string (auto-generated projection of JSON at: /stringUriTemplateField with inferred types: [string])
35: stringUuidField: required string (auto-generated projection of JSON at: /stringUuidField with inferred types: [string])
36: flow_document: required string (auto-generated projection of JSON at: with inferred types: [object])
36: flow_document: required string (auto-generated projection of the root document with inferred types: [object])
}

Big Schema Materialized Resource Schema With No Fields Required:
Expand Down Expand Up @@ -192,7 +192,7 @@ table {
33: stringUriReferenceField: optional string (auto-generated projection of JSON at: /stringUriReferenceField with inferred types: [string])
34: stringUriTemplateField: optional string (auto-generated projection of JSON at: /stringUriTemplateField with inferred types: [string])
35: stringUuidField: optional string (auto-generated projection of JSON at: /stringUuidField with inferred types: [string])
36: flow_document: required string (auto-generated projection of JSON at: with inferred types: [object])
36: flow_document: required string (auto-generated projection of the root document with inferred types: [object])
}

Big Schema Changed Types With Table Replacement Constraints:
Expand Down Expand Up @@ -272,7 +272,7 @@ table {
34: stringUriReferenceField: required string (auto-generated projection of JSON at: /stringUriReferenceField with inferred types: [string])
35: stringUriTemplateField: required string (auto-generated projection of JSON at: /stringUriTemplateField with inferred types: [string])
36: stringUuidField: required string (auto-generated projection of JSON at: /stringUuidField with inferred types: [string])
37: flow_document: required string (auto-generated projection of JSON at: with inferred types: [object])
37: flow_document: required string (auto-generated projection of the root document with inferred types: [object])
}

add a single field:
Expand All @@ -288,7 +288,7 @@ table {
9: requiredInteger: required long (auto-generated projection of JSON at: /requiredInteger with inferred types: [integer])
10: requiredObject: required string (auto-generated projection of JSON at: /requiredObject with inferred types: [object])
11: requiredString: required string (auto-generated projection of JSON at: /requiredString with inferred types: [string])
12: flow_document: required string (user-provided projection of JSON at: with inferred types: [object])
12: flow_document: required string (user-provided projection of the root document with inferred types: [object])
13: addedOptionalString: optional string (auto-generated projection of JSON at: /addedOptionalString with inferred types: [string])
}

Expand All @@ -305,7 +305,7 @@ table {
9: requiredInteger: required long (auto-generated projection of JSON at: /requiredInteger with inferred types: [integer])
10: requiredObject: required string (auto-generated projection of JSON at: /requiredObject with inferred types: [object])
11: requiredString: required string (auto-generated projection of JSON at: /requiredString with inferred types: [string])
12: flow_document: required string (user-provided projection of JSON at: with inferred types: [object])
12: flow_document: required string (user-provided projection of the root document with inferred types: [object])
}

remove a single required field:
Expand All @@ -321,7 +321,7 @@ table {
9: requiredInteger: required long (auto-generated projection of JSON at: /requiredInteger with inferred types: [integer])
10: requiredObject: required string (auto-generated projection of JSON at: /requiredObject with inferred types: [object])
11: requiredString: optional string (auto-generated projection of JSON at: /requiredString with inferred types: [string])
12: flow_document: required string (user-provided projection of JSON at: with inferred types: [object])
12: flow_document: required string (user-provided projection of the root document with inferred types: [object])
}

add and remove many fields:
Expand All @@ -337,7 +337,7 @@ table {
9: requiredInteger: required long (auto-generated projection of JSON at: /requiredInteger with inferred types: [integer])
10: requiredObject: optional string (auto-generated projection of JSON at: /requiredObject with inferred types: [object])
11: requiredString: optional string (auto-generated projection of JSON at: /requiredString with inferred types: [string])
12: flow_document: required string (user-provided projection of JSON at: with inferred types: [object])
12: flow_document: required string (user-provided projection of the root document with inferred types: [object])
13: addedOptionalString: optional string (auto-generated projection of JSON at: /addedOptionalString with inferred types: [string])
14: addedRequiredString: required string (auto-generated projection of JSON at: /addedRequiredString with inferred types: [string])
}
Expand All @@ -357,6 +357,6 @@ table {
11: value.with-separated_words: optional string (auto-generated projection of JSON at: /value.with-separated_words with inferred types: [string])
12: value.with.separated.words: optional string (auto-generated projection of JSON at: /value.with.separated.words with inferred types: [string])
13: value_with_separated_words: optional string (auto-generated projection of JSON at: /value_with_separated_words with inferred types: [string])
14: flow_document: required string (auto-generated projection of JSON at: with inferred types: [object])
14: flow_document: required string (auto-generated projection of the root document with inferred types: [object])
}

4 changes: 4 additions & 0 deletions materialize-iceberg/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,10 @@ func (c config) Validate() error {
return nil
}

func (c config) DefaultNamespace() string {
return sanitizePath(c.Namespace)[0]
}

type catalogAuthType string

const (
Expand Down
Loading
Loading