laskoviymishka commented on code in PR #1359:
URL: https://github.com/apache/iceberg-go/pull/1359#discussion_r3552779486
##########
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:
This is the one spot in the file that calls `PromoteType(new, existing)` —
every other call, including `updateColumn`, is `PromoteType(existing, new)`. It
happens to work here only because the reversal flips the widening test in the
direction we want, so it's a latent trap: a future maintainer "fixing" the
order to match the convention introduces a silent correctness bug.
I'd either invert the test to match the rest of the file, or drop a comment
right here spelling out that the argument order is intentionally reversed and
why.
##########
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)
+
+ return err == nil
+ }
+
+ _, ok := newType.(iceberg.PrimitiveType)
Review Comment:
I think there's a real bug in this fallthrough. Once we're past the
primitive branch, we return `true` (ignorable) whenever `newType` isn't
primitive — so `list<int>`→`map<string,int>`, `struct`→`list`, `map`→`struct`
all read as no-ops regardless of whether the kinds actually match.
The problem is what happens next: `unionField` then switches on the *new*
type and recurses its children against the *existing* field. For `list`→`map`
we look up `path.value` in a list, miss, and `unionAddColumn` grafts the
map-value subtree onto the list element. The field IDs won't line up with the
physical columns, so Java/PyIceberg/rust can't read the result — and for other
inputs it surfaces later as a misleading "cannot add field to non-struct type"
at the wrong path.
I'd gate this on same-kind: only return `true` when `newType` is the same
complex kind as `existingType`, otherwise `false` so `unionUpdateColumn` emits
a real (and rejectable) type change. That also transitively fixes the map-key
guard below — right now `existing.Type.(*iceberg.MapType)` silently skips the
key-immutability check whenever the existing field isn't already a map, so a
`list`→`map` never validates the key at all. Worth an explicit error when
`newField.Type` is a map but `existing.Type` isn't. Could we add cross-kind
cases (`list↔map`, `struct↔list`, `struct↔map`) to lock it in? wdyt?
##########
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) {
Review Comment:
This writes straight into `u.updates` and skips everything `updateColumn`
guards. Two things worry me.
There's no `u.isDeleted(existing.ID)` check — a transaction that deletes a
field and then unions a schema with a differing write-default inserts a stale
entry for a deleted field. It's harmless today only because delete wins in
`applyChanges`, but that's an ordering coincidence, not an invariant.
And taking `writeDefault any` instead of an `iceberg.Literal` means an
incompatible default (an `int64` for an `int32` field) is accepted with no type
check and can blow up downstream at encode time. I'd route this through the
same `ColumnUpdate`/`updateColumn` path the rest of the defaults use, or at
minimum add the `isDeleted` guard plus a type-compat check.
##########
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)
+
+ return err == nil
Review Comment:
Treating anything `PromoteType` accepts as "ignorable" pulls in promotions
Java's `isPromotionAllowed` explicitly rejects. `PromoteType` allows
string↔binary, fixed[16]→uuid, and decimal scale *widening* (`scale <= scale`),
whereas Java permits only int→long, float→double, and decimal with exact scale
equality.
Concretely: existing `string` + incoming `binary` returns ignorable and
silently keeps `string`; incoming `uuid` on existing `fixed[16]` stages a
`uuid` field over `fixed[16]` physical data. Both write schemas Java/PyIceberg
can't resolve. I'd add a narrow `isPromotionAllowed`-style helper mirroring
Java's exact switch rather than delegating the whole spec-allowed set to
`PromoteType`. wdyt?
##########
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) {
Review Comment:
This switch omits `PrimitiveType` and `VariantType`. The implicit no-op is
correct (neither has nested fields), but the exhaustive linter CI runs will
flag a type switch with no `default`, and `applyChanges.Variant` already
handles variant explicitly elsewhere. I'd add a `default:` with a one-line
comment noting there's nothing to recurse into, so CI stays green.
##########
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 {
Review Comment:
Every other exported `UpdateSchema` method carries a godoc comment, and this
one has enough load-bearing semantics — forced-optional on new fields,
doc/write-default updates, case-sensitivity — that a short doc block would help
callers a lot. Worth adding before this goes in.
##########
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,
Review Comment:
`addColumn` guards initial defaults with a `defaultValue.Type().Equals(t)`
check; here we copy `InitialDefault` verbatim with no type-compatibility check.
Not blocking, but it'd be nice to keep the two add paths consistent so a
mistyped default can't slip through the union route.
##########
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] {
Review Comment:
`addColumn` also scans `u.updates[parentID]` for a name collision here, and
`unionAddColumn` drops that scan. Since `findField` only sees the current
schema, not pending renames: if a prior op renamed `old`→`new` and the incoming
schema has `new`, `findField("new")` misses, we fall in here, and append a
second `new` to `u.adds` — a duplicate name in the applied schema. I'd mirror
the `u.updates` scan before the `u.adds` check.
--
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]
