zeroshade commented on code in PR #1359:
URL: https://github.com/apache/iceberg-go/pull/1359#discussion_r3553634572
##########
table/update_schema.go:
##########
@@ -616,6 +617,185 @@ func (u *UpdateSchema) SetIdentifierField(paths
[][]string) *UpdateSchema {
return u
}
+func (u *UpdateSchema) UnionByNameWith(newSchema *iceberg.Schema)
*UpdateSchema {
+ u.ops = append(u.ops, func() error {
+ return u.unionByName(newSchema)
+ })
+
+ return u
+}
+
+func (u *UpdateSchema) unionByName(newSchema *iceberg.Schema) error {
+ if newSchema == nil {
+ return errors.New("cannot union with nil schema")
+ }
+
+ for _, field := range newSchema.Fields() {
+ if err := u.unionField([]string{field.Name}, field); err != nil
{
+ return err
+ }
+ }
+
+ return nil
+}
+
+func (u *UpdateSchema) unionField(path []string, newField iceberg.NestedField)
error {
+ existing, ok := u.findField(strings.Join(path, "."))
+ if !ok {
+ return u.unionAddColumn(path, newField)
+ }
+
+ if err := u.unionUpdateColumn(path, existing, newField); err != nil {
+ return err
+ }
+
+ switch t := newField.Type.(type) {
+ case *iceberg.StructType:
+ for _, child := range t.FieldList {
+ if err := u.unionField(childPath(path, child.Name),
child); err != nil {
+ return err
+ }
+ }
+ case *iceberg.ListType:
+ if err := u.unionField(childPath(path, "element"),
t.ElementField()); err != nil {
+ return err
+ }
+ case *iceberg.MapType:
+ // Map keys are immutable; reject any key change up front
+ if existingMap, ok := existing.Type.(*iceberg.MapType); ok {
+ if !isIgnorableTypeUpdate(existingMap.KeyType,
t.KeyType) {
+ return fmt.Errorf("cannot update map key: %s",
strings.Join(childPath(path, "key"), "."))
+ }
+ }
+ if err := u.unionField(childPath(path, "value"),
t.ValueField()); err != nil {
+ return err
+ }
+ }
+
+ return nil
+}
+
+func childPath(parent []string, name string) []string {
+ child := make([]string, 0, len(parent)+1)
+ child = append(child, parent...)
+
+ return append(child, name)
+}
+
+func (u *UpdateSchema) unionAddColumn(path []string, newField
iceberg.NestedField) error {
+ fullName := strings.Join(path, ".")
+
+ parent := path[:len(path)-1]
+ parentID := TableRootID
+
+ if len(parent) > 0 {
+ parentFullPath := strings.Join(parent, ".")
+ parentField, ok := u.findField(parentFullPath)
+ if !ok {
+ return fmt.Errorf("parent field not found: %s",
parentFullPath)
+ }
+
+ switch parentType := parentField.Type.(type) {
+ case *iceberg.ListType:
+ parentField = parentType.ElementField()
+ case *iceberg.MapType:
+ parentField = parentType.ValueField()
+ }
+
+ if _, ok := parentField.Type.(*iceberg.StructType); !ok {
+ return fmt.Errorf("cannot add field to non-struct type:
%s", parentFullPath)
+ }
+
+ parentID = parentField.ID
+ }
+
+ name := path[len(path)-1]
+ for _, add := range u.adds[parentID] {
+ if add.Name == name {
+ return fmt.Errorf("field already exists in adds: %s",
fullName)
+ }
+ }
+
+ field := iceberg.NestedField{
+ Name: name,
+ Type: newField.Type,
+ Required: false,
+ Doc: newField.Doc,
+ InitialDefault: newField.InitialDefault,
+ WriteDefault: newField.WriteDefault,
+ }
+
+ sch, err := iceberg.AssignFreshSchemaIDs(iceberg.NewSchema(0, field),
u.assignNewColumnID)
+ if err != nil {
+ return fmt.Errorf("failed to assign field id: %w", err)
+ }
+ u.adds[parentID] = append(u.adds[parentID], sch.Field(0))
+ u.addedNameToID[fullName] = sch.Field(0).ID
+
+ return nil
+}
+
+func (u *UpdateSchema) unionUpdateColumn(path []string, existing, newField
iceberg.NestedField) error {
+ update := ColumnUpdate{}
+
+ if !newField.Required && existing.Required {
+ update.Required = iceberg.Optional[bool]{Valid: true, Val:
false}
+ }
+
+ if !isIgnorableTypeUpdate(existing.Type, newField.Type) {
+ update.FieldType = iceberg.Optional[iceberg.Type]{Valid: true,
Val: newField.Type}
+ }
+
+ if newField.Doc != "" && newField.Doc != existing.Doc {
+ update.Doc = iceberg.Optional[string]{Valid: true, Val:
newField.Doc}
+ }
+
+ if update.Required.Valid || update.FieldType.Valid || update.Doc.Valid {
+ if err := u.updateColumn(path, update); err != nil {
+ return err
+ }
+ }
+
+ if newField.WriteDefault != nil &&
!reflect.DeepEqual(newField.WriteDefault, existing.WriteDefault) {
+ u.setWriteDefault(existing, newField.WriteDefault)
+ }
+
+ return nil
+}
+
+func (u *UpdateSchema) setWriteDefault(existing iceberg.NestedField,
writeDefault any) {
+ parentID := u.findParentID(existing.ID)
+ if u.updates[parentID] == nil {
+ u.updates[parentID] = make(map[int]iceberg.NestedField)
+ }
+
+ updatedField, ok := u.updates[parentID][existing.ID]
+ if !ok {
+ updatedField = existing
+ }
+ updatedField.WriteDefault = writeDefault
+ u.updates[parentID][existing.ID] = updatedField
Review Comment:
`setWriteDefault` writes directly into `u.updates`, bypassing
`updateColumn`'s deleted-field guard and default validation, and accepts a raw
`any` — so an incompatible write default can enter schema metadata unchecked.
##########
table/update_schema.go:
##########
@@ -616,6 +617,185 @@ func (u *UpdateSchema) SetIdentifierField(paths
[][]string) *UpdateSchema {
return u
}
+func (u *UpdateSchema) UnionByNameWith(newSchema *iceberg.Schema)
*UpdateSchema {
+ u.ops = append(u.ops, func() error {
+ return u.unionByName(newSchema)
+ })
+
+ return u
+}
+
+func (u *UpdateSchema) unionByName(newSchema *iceberg.Schema) error {
+ if newSchema == nil {
+ return errors.New("cannot union with nil schema")
+ }
+
+ for _, field := range newSchema.Fields() {
+ if err := u.unionField([]string{field.Name}, field); err != nil
{
+ return err
+ }
+ }
+
+ return nil
+}
+
+func (u *UpdateSchema) unionField(path []string, newField iceberg.NestedField)
error {
+ existing, ok := u.findField(strings.Join(path, "."))
+ if !ok {
+ return u.unionAddColumn(path, newField)
+ }
+
+ if err := u.unionUpdateColumn(path, existing, newField); err != nil {
+ return err
+ }
+
+ switch t := newField.Type.(type) {
+ case *iceberg.StructType:
+ for _, child := range t.FieldList {
+ if err := u.unionField(childPath(path, child.Name),
child); err != nil {
+ return err
+ }
+ }
+ case *iceberg.ListType:
+ if err := u.unionField(childPath(path, "element"),
t.ElementField()); err != nil {
+ return err
+ }
+ case *iceberg.MapType:
+ // Map keys are immutable; reject any key change up front
+ if existingMap, ok := existing.Type.(*iceberg.MapType); ok {
+ if !isIgnorableTypeUpdate(existingMap.KeyType,
t.KeyType) {
+ return fmt.Errorf("cannot update map key: %s",
strings.Join(childPath(path, "key"), "."))
+ }
+ }
+ if err := u.unionField(childPath(path, "value"),
t.ValueField()); err != nil {
+ return err
+ }
+ }
+
+ return nil
+}
+
+func childPath(parent []string, name string) []string {
+ child := make([]string, 0, len(parent)+1)
+ child = append(child, parent...)
+
+ return append(child, name)
+}
+
+func (u *UpdateSchema) unionAddColumn(path []string, newField
iceberg.NestedField) error {
+ fullName := strings.Join(path, ".")
+
+ parent := path[:len(path)-1]
+ parentID := TableRootID
+
+ if len(parent) > 0 {
+ parentFullPath := strings.Join(parent, ".")
+ parentField, ok := u.findField(parentFullPath)
+ if !ok {
+ return fmt.Errorf("parent field not found: %s",
parentFullPath)
+ }
+
+ switch parentType := parentField.Type.(type) {
+ case *iceberg.ListType:
+ parentField = parentType.ElementField()
+ case *iceberg.MapType:
+ parentField = parentType.ValueField()
+ }
+
+ if _, ok := parentField.Type.(*iceberg.StructType); !ok {
+ return fmt.Errorf("cannot add field to non-struct type:
%s", parentFullPath)
+ }
+
+ parentID = parentField.ID
+ }
+
+ name := path[len(path)-1]
+ for _, add := range u.adds[parentID] {
+ if add.Name == name {
+ return fmt.Errorf("field already exists in adds: %s",
fullName)
+ }
+ }
+
+ field := iceberg.NestedField{
+ Name: name,
+ Type: newField.Type,
+ Required: false,
+ Doc: newField.Doc,
+ InitialDefault: newField.InitialDefault,
+ WriteDefault: newField.WriteDefault,
+ }
+
+ sch, err := iceberg.AssignFreshSchemaIDs(iceberg.NewSchema(0, field),
u.assignNewColumnID)
+ if err != nil {
+ return fmt.Errorf("failed to assign field id: %w", err)
+ }
+ u.adds[parentID] = append(u.adds[parentID], sch.Field(0))
Review Comment:
`unionAddColumn` scans `u.adds` for name collisions but not `u.renames`
(which `addColumn` does). A chained rename followed by a union can add a
duplicate name, because `findField` only sees the original schema.
##########
table/update_schema.go:
##########
@@ -616,6 +617,185 @@ func (u *UpdateSchema) SetIdentifierField(paths
[][]string) *UpdateSchema {
return u
}
+func (u *UpdateSchema) UnionByNameWith(newSchema *iceberg.Schema)
*UpdateSchema {
+ u.ops = append(u.ops, func() error {
+ return u.unionByName(newSchema)
+ })
+
+ return u
+}
+
+func (u *UpdateSchema) unionByName(newSchema *iceberg.Schema) error {
+ if newSchema == nil {
+ return errors.New("cannot union with nil schema")
+ }
+
+ for _, field := range newSchema.Fields() {
+ if err := u.unionField([]string{field.Name}, field); err != nil
{
+ return err
+ }
+ }
+
+ return nil
+}
+
+func (u *UpdateSchema) unionField(path []string, newField iceberg.NestedField)
error {
+ existing, ok := u.findField(strings.Join(path, "."))
+ if !ok {
+ return u.unionAddColumn(path, newField)
+ }
+
+ if err := u.unionUpdateColumn(path, existing, newField); err != nil {
+ return err
+ }
+
+ switch t := newField.Type.(type) {
+ case *iceberg.StructType:
+ for _, child := range t.FieldList {
+ if err := u.unionField(childPath(path, child.Name),
child); err != nil {
+ return err
+ }
+ }
+ case *iceberg.ListType:
+ if err := u.unionField(childPath(path, "element"),
t.ElementField()); err != nil {
+ return err
+ }
+ case *iceberg.MapType:
+ // Map keys are immutable; reject any key change up front
+ if existingMap, ok := existing.Type.(*iceberg.MapType); ok {
+ if !isIgnorableTypeUpdate(existingMap.KeyType,
t.KeyType) {
+ return fmt.Errorf("cannot update map key: %s",
strings.Join(childPath(path, "key"), "."))
+ }
+ }
+ if err := u.unionField(childPath(path, "value"),
t.ValueField()); err != nil {
+ return err
+ }
+ }
+
+ return nil
+}
+
+func childPath(parent []string, name string) []string {
+ child := make([]string, 0, len(parent)+1)
+ child = append(child, parent...)
+
+ return append(child, name)
+}
+
+func (u *UpdateSchema) unionAddColumn(path []string, newField
iceberg.NestedField) error {
+ fullName := strings.Join(path, ".")
+
+ parent := path[:len(path)-1]
+ parentID := TableRootID
+
+ if len(parent) > 0 {
+ parentFullPath := strings.Join(parent, ".")
+ parentField, ok := u.findField(parentFullPath)
+ if !ok {
+ return fmt.Errorf("parent field not found: %s",
parentFullPath)
+ }
+
+ switch parentType := parentField.Type.(type) {
+ case *iceberg.ListType:
+ parentField = parentType.ElementField()
+ case *iceberg.MapType:
+ parentField = parentType.ValueField()
+ }
+
+ if _, ok := parentField.Type.(*iceberg.StructType); !ok {
+ return fmt.Errorf("cannot add field to non-struct type:
%s", parentFullPath)
+ }
+
+ parentID = parentField.ID
+ }
+
+ name := path[len(path)-1]
+ for _, add := range u.adds[parentID] {
+ if add.Name == name {
+ return fmt.Errorf("field already exists in adds: %s",
fullName)
+ }
+ }
+
+ field := iceberg.NestedField{
+ Name: name,
+ Type: newField.Type,
+ Required: false,
+ Doc: newField.Doc,
+ InitialDefault: newField.InitialDefault,
+ WriteDefault: newField.WriteDefault,
+ }
+
+ sch, err := iceberg.AssignFreshSchemaIDs(iceberg.NewSchema(0, field),
u.assignNewColumnID)
+ if err != nil {
+ return fmt.Errorf("failed to assign field id: %w", err)
+ }
+ u.adds[parentID] = append(u.adds[parentID], sch.Field(0))
+ u.addedNameToID[fullName] = sch.Field(0).ID
+
+ return nil
+}
+
+func (u *UpdateSchema) unionUpdateColumn(path []string, existing, newField
iceberg.NestedField) error {
+ update := ColumnUpdate{}
+
+ if !newField.Required && existing.Required {
+ update.Required = iceberg.Optional[bool]{Valid: true, Val:
false}
+ }
+
+ if !isIgnorableTypeUpdate(existing.Type, newField.Type) {
+ update.FieldType = iceberg.Optional[iceberg.Type]{Valid: true,
Val: newField.Type}
+ }
+
+ if newField.Doc != "" && newField.Doc != existing.Doc {
+ update.Doc = iceberg.Optional[string]{Valid: true, Val:
newField.Doc}
+ }
+
+ if update.Required.Valid || update.FieldType.Valid || update.Doc.Valid {
+ if err := u.updateColumn(path, update); err != nil {
+ return err
+ }
+ }
+
+ if newField.WriteDefault != nil &&
!reflect.DeepEqual(newField.WriteDefault, existing.WriteDefault) {
+ u.setWriteDefault(existing, newField.WriteDefault)
+ }
+
+ return nil
+}
+
+func (u *UpdateSchema) setWriteDefault(existing iceberg.NestedField,
writeDefault any) {
+ parentID := u.findParentID(existing.ID)
+ if u.updates[parentID] == nil {
+ u.updates[parentID] = make(map[int]iceberg.NestedField)
+ }
+
+ updatedField, ok := u.updates[parentID][existing.ID]
+ if !ok {
+ updatedField = existing
+ }
+ updatedField.WriteDefault = writeDefault
+ u.updates[parentID][existing.ID] = updatedField
+}
+
+func isIgnorableTypeUpdate(existingType, newType iceberg.Type) bool {
+ if _, ok := existingType.(iceberg.PrimitiveType); ok {
+ newPrimitive, ok := newType.(iceberg.PrimitiveType)
+ if !ok {
+ return false
+ }
+ if existingType.Equals(newType) {
+ return true
+ }
+ _, err := iceberg.PromoteType(newPrimitive, existingType)
Review Comment:
Two issues in `isIgnorableTypeUpdate`: (1) the non-primitive branch
effectively `return !ok` (around line 796) treats any
non-primitive→non-primitive pair as ignorable, so cross-kind changes (list→map,
struct→list) become no-ops and then recurse into the wrong child layout —
risking corrupted field IDs / invalid schemas. (2) `PromoteType(newPrimitive,
existingType)` has its arguments reversed vs. the `updateColumn` convention
used elsewhere; even if it works today it's fragile and leans on the broader
`PromoteType` allowed set rather than the narrower promotion rules.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
