You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ze...@apache.org on 2026/07/24 18:27:32 UTC
(iceberg-go) branch main updated: feat(table): union schema evolution (#1359)
This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 0df1d8b7e feat(table): union schema evolution (#1359)
0df1d8b7e is described below
commit 0df1d8b7ebf951d4e3501f717fcad69e257c0cf1
Author: Badal Prasad Singh <ba...@datazip.io>
AuthorDate: Fri Jul 24 23:57:27 2026 +0530
feat(table): union schema evolution (#1359)
### Description
In Iceberg-go, `table.UpdateSchema`, right now, does not provide a way
to evolve a table's schema to the union of its current schema and an
incoming schema matched by field name.
Iceberg-java's `org.apache.iceberg.UpdateSchema` exposes this as
`unionByNameWith(Schema)`. This PR solves the same for Iceberg-Go.
Fixes #1187
---------
Signed-off-by: badalprasadsingh <ba...@datazip.io>
---
table/update_schema.go | 392 +++++++++++++++++++
table/update_schema_test.go | 901 ++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 1293 insertions(+)
diff --git a/table/update_schema.go b/table/update_schema.go
index 577d768ef..6709437e5 100644
--- a/table/update_schema.go
+++ b/table/update_schema.go
@@ -22,10 +22,12 @@ import (
"errors"
"fmt"
"maps"
+ "reflect"
"slices"
"strings"
"github.com/apache/iceberg-go"
+ "github.com/google/uuid"
)
const TableRootID = -1
@@ -631,6 +633,396 @@ func (u *UpdateSchema) SetIdentifierField(paths [][]string) *UpdateSchema {
return u
}
+// UnionByNameWith stages changes to evolve the current schema into the union of
+// itself and newSchema, matching fields by name (honoring caseSensitive):
+//
+// - New fields are added as optional (regardless of their incoming required
+// flag), keeping their Doc/InitialDefault/WriteDefault and getting fresh ids.
+// - Existing fields may be evolved: required->optional is applied,
+// optional->required is skipped. A primitive type change is applied only if
+// it is a valid promotion (int->long, float->double, decimal same-scale
+// wider-precision, or exact match); narrowing is ignored and any other
+// change errors.
+// - A Doc is updated only when the incoming Doc is non-empty and differs (an
+// empty Doc never clears an existing one).
+// - A WriteDefault is updated when the incoming value differs; an existing
+// column's InitialDefault is never modified.
+// - Map keys are immutable and cross-kind changes (list->map, struct->list,
+// etc.) are rejected rather than silently grafted onto the existing column.
+//
+// Matching consults only the current schema, not columns staged by an earlier
+// op in the same UpdateSchema; pre-merge such schemas before a single call.
+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
+ existingMap, ok := existing.Type.(*iceberg.MapType)
+ if !ok {
+ return fmt.Errorf("cannot change field to map: %s", strings.Join(path, "."))
+ }
+ 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
+ }
+ default:
+ // Primitive and Variant types have no nested children to recurse into.
+ }
+
+ 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)
+ }
+ }
+
+ // guard against colliding with a pending rename
+ if field, ok := u.findField(fullName); ok {
+ if !u.isDeleted(field.ID) {
+ for _, upd := range u.updates[parentID] {
+ if upd.Name == name {
+ return fmt.Errorf("field already exists: %s", fullName)
+ }
+ }
+ }
+ }
+
+ if err := validateFieldDefaults(newField); err != nil {
+ return err
+ }
+
+ 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) {
+ // Only a valid primitive promotion may change a column's type. Anything
+ // else (non-primitive, cross-kind, or a disallowed promotion) is an error.
+ existingPrim, existingIsPrim := existing.Type.(iceberg.PrimitiveType)
+ newPrim, newIsPrim := newField.Type.(iceberg.PrimitiveType)
+ if !existingIsPrim || !newIsPrim || !isPromotionAllowed(existingPrim, newPrim) {
+ return fmt.Errorf("cannot change column type: %s: %s -> %s",
+ strings.Join(path, "."), 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) {
+ // Validate against the promoted type when this same op changes the
+ // column's type
+ effectiveType := existing.Type
+ if update.FieldType.Valid {
+ effectiveType = update.FieldType.Val
+ }
+ if err := u.setWriteDefault(existing, effectiveType, newField.WriteDefault); err != nil {
+ return err
+ }
+ }
+
+ return nil
+}
+
+func (u *UpdateSchema) setWriteDefault(existing iceberg.NestedField, fieldType iceberg.Type, writeDefault any) error {
+ if u.isDeleted(existing.ID) {
+ return fmt.Errorf("field that has been deleted cannot be updated: %s", existing.Name)
+ }
+ if _, ok := fieldType.(iceberg.PrimitiveType); !ok {
+ return fmt.Errorf("default values are not supported for %s", fieldType)
+ }
+ if err := validateDefaultValue(fieldType, writeDefault); err != nil {
+ return fmt.Errorf("invalid write-default for %s: %w", existing.Name, err)
+ }
+
+ 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
+
+ return nil
+}
+
+func validateNestedDefaults(t iceberg.Type) error {
+ switch typ := t.(type) {
+ case *iceberg.StructType:
+ for _, f := range typ.FieldList {
+ if err := validateFieldDefaults(f); err != nil {
+ return err
+ }
+ }
+ case *iceberg.ListType:
+ return validateFieldDefaults(typ.ElementField())
+ case *iceberg.MapType:
+ if err := validateFieldDefaults(typ.KeyField()); err != nil {
+ return err
+ }
+
+ return validateFieldDefaults(typ.ValueField())
+ }
+
+ return nil
+}
+
+func validateFieldDefaults(f iceberg.NestedField) error {
+ switch f.Type.(type) {
+ case *iceberg.StructType, *iceberg.ListType, *iceberg.MapType:
+ if err := validateComplexDefault(f); err != nil {
+ return err
+ }
+
+ return validateNestedDefaults(f.Type)
+ case iceberg.PrimitiveType:
+ if err := validateDefaultValue(f.Type, f.InitialDefault); err != nil {
+ return fmt.Errorf("invalid initial-default for %s: %w", f.Name, err)
+ }
+ if err := validateDefaultValue(f.Type, f.WriteDefault); err != nil {
+ return fmt.Errorf("invalid write-default for %s: %w", f.Name, err)
+ }
+ }
+
+ return nil
+}
+
+// nativeDefaultLiteral wraps a default already in a native iceberg Go type so
+// its Literal type can be checked against the column type. Values whose Go
+// encoding is shared across iceberg types (e.g. string, []byte, float64,
+// Decimal) are excluded, deferring to the materialization check below.
+func nativeDefaultLiteral(v any) (iceberg.Literal, bool) {
+ switch val := v.(type) {
+ case bool:
+ return iceberg.NewLiteral(val), true
+ case int32:
+ return iceberg.NewLiteral(val), true
+ case int64:
+ return iceberg.NewLiteral(val), true
+ case float32:
+ return iceberg.NewLiteral(val), true
+ case iceberg.Date:
+ return iceberg.NewLiteral(val), true
+ case iceberg.Time:
+ return iceberg.NewLiteral(val), true
+ case iceberg.Timestamp:
+ return iceberg.NewLiteral(val), true
+ case iceberg.TimestampNano:
+ return iceberg.NewLiteral(val), true
+ case uuid.UUID:
+ return iceberg.NewLiteral(val), true
+ }
+
+ return nil, false
+}
+
+func validateDefaultValue(t iceberg.Type, v any) (err error) {
+ if v == nil {
+ return nil
+ }
+ if _, ok := t.(iceberg.PrimitiveType); !ok {
+ return fmt.Errorf("default values are not supported for %s", t)
+ }
+
+ if lit, ok := nativeDefaultLiteral(v); ok && !lit.Type().Equals(t) {
+ return fmt.Errorf("value of type %s does not match column type %s", lit.Type(), t)
+ }
+
+ dt, err := TypeToArrowType(t, false, false)
+ if err != nil {
+ return err
+ }
+
+ defer func() {
+ if r := recover(); r != nil {
+ err = fmt.Errorf("value %#v is not valid for type %s: %v", v, t, r)
+ }
+ }()
+ _ = defaultToScalar(v, t, dt)
+
+ return nil
+}
+
+// isIgnorableTypeUpdate reports whether changing existingType to newType is a
+// no-op for union-by-name: either the incoming primitive is narrower-or-equal
+// (so the existing, wider type already covers it), or both types are the same
+// complex kind. Cross-kind changes return false so the caller can reject them.
+//
+// Note: the promotion check is called with the arguments reversed relative to
+// the rest of the file. Here we ask whether newType can be promoted *up* to
+// existingType; a true result means the change is narrowing and ignorable.
+func isIgnorableTypeUpdate(existingType, newType iceberg.Type) bool {
+ // An unchanged type is always a no-op. This also covers parameterless
+ // non-primitive types such as variant.
+ if existingType.Equals(newType) {
+ return true
+ }
+
+ if existingPrim, ok := existingType.(iceberg.PrimitiveType); ok {
+ newPrimitive, ok := newType.(iceberg.PrimitiveType)
+ if !ok {
+ return false
+ }
+
+ return isPromotionAllowed(newPrimitive, existingPrim)
+ }
+
+ switch existingType.(type) {
+ case *iceberg.StructType:
+ _, ok := newType.(*iceberg.StructType)
+
+ return ok
+ case *iceberg.ListType:
+ _, ok := newType.(*iceberg.ListType)
+
+ return ok
+ case *iceberg.MapType:
+ _, ok := newType.(*iceberg.MapType)
+
+ return ok
+ }
+
+ return false
+}
+
+func isPromotionAllowed(from, to iceberg.PrimitiveType) bool {
+ if from.Equals(to) {
+ return true
+ }
+ switch f := from.(type) {
+ case iceberg.Int32Type:
+ _, ok := to.(iceberg.Int64Type)
+
+ return ok
+ case iceberg.Float32Type:
+ _, ok := to.(iceberg.Float64Type)
+
+ return ok
+ case iceberg.DecimalType:
+ t, ok := to.(iceberg.DecimalType)
+ if !ok {
+ return false
+ }
+
+ return f.Scale() == t.Scale() && f.Precision() <= t.Precision()
+ }
+
+ return false
+}
+
func (u *UpdateSchema) BuildUpdates() ([]Update, []Requirement, error) {
if u.err != nil {
return nil, nil, u.err
diff --git a/table/update_schema_test.go b/table/update_schema_test.go
index c469d679f..f0f5ec338 100644
--- a/table/update_schema_test.go
+++ b/table/update_schema_test.go
@@ -1414,3 +1414,904 @@ func TestAddColumnAfterDropHighestID(t *testing.T) {
"new column id must be last-column-id+1 (13), not HighestFieldID+1 (12) — "+
"reusing 12 would collide with the dropped column still in historical schemas")
}
+
+func unionTxn(t *testing.T, schema *iceberg.Schema) *Transaction {
+ t.Helper()
+ meta, err := NewMetadata(schema, nil, UnsortedSortOrder, "", nil)
+ require.NoError(t, err)
+
+ return New([]string{"tbl"}, meta, "", nil, nil).NewTransaction()
+}
+
+func TestUnionByNameAddTopLevelColumns(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
+ iceberg.NestedField{ID: 3, Name: "age", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ name, ok := applied.FindFieldByName("name")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.String, name.Type)
+ assert.False(t, name.Required)
+
+ age, ok := applied.FindFieldByName("age")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.Int32, age.Type)
+}
+
+func TestUnionByNameAddedColumnsAreOptional(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "data", Type: iceberg.PrimitiveTypes.String, Required: true},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ data, ok := applied.FindFieldByName("data")
+ require.True(t, ok)
+ assert.False(t, data.Required)
+}
+
+func TestUnionByNameAddNestedStructField(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "address", Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 3, Name: "city", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ }, Required: false},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "address", Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 3, Name: "city", Type: iceberg.PrimitiveTypes.String, Required: false},
+ {ID: 4, Name: "zip", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ }, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ zip, ok := applied.FindFieldByName("address.zip")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.String, zip.Type)
+
+ _, ok = applied.FindFieldByName("address.city")
+ assert.True(t, ok)
+}
+
+func TestUnionByNamePromoteType(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "amount", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "amount", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ amount, ok := applied.FindFieldByName("amount")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.Int64, amount.Type)
+}
+
+func TestUnionByNamePromoteFloatToDouble(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "amount", Type: iceberg.PrimitiveTypes.Float32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "amount", Type: iceberg.PrimitiveTypes.Float64, Required: true},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ amount, ok := applied.FindFieldByName("amount")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.Float64, amount.Type)
+}
+
+func TestUnionByNamePromoteDecimalPrecision(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "amount", Type: iceberg.DecimalTypeOf(9, 2), Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "amount", Type: iceberg.DecimalTypeOf(18, 2), Required: true},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ amount, ok := applied.FindFieldByName("amount")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.DecimalTypeOf(18, 2), amount.Type)
+}
+
+func TestUnionByNameIgnoreNarrowingType(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "amount", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "amount", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ amount, ok := applied.FindFieldByName("amount")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.Int64, amount.Type)
+}
+
+// An invalid (non-promotable) map value type change must error.
+func TestUnionByNameInvalidMapValueTypeChange(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.String, ValueRequired: false,
+ }, Required: false},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.Int64, ValueRequired: false,
+ }, Required: false},
+ )
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.Error(t, err, "map value string -> long is not a valid promotion")
+}
+
+func TestUnionByNamePromoteListElement(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "vals", Type: &iceberg.ListType{
+ ElementID: 2, Element: iceberg.PrimitiveTypes.Int32, ElementRequired: false,
+ }, Required: false},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "vals", Type: &iceberg.ListType{
+ ElementID: 2, Element: iceberg.PrimitiveTypes.Int64, ElementRequired: false,
+ }, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ elem, ok := applied.FindFieldByName("vals.element")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.Int64, elem.Type)
+}
+
+func TestUnionByNameMakeColumnOptional(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ id, ok := applied.FindFieldByName("id")
+ require.True(t, ok)
+ assert.False(t, id.Required)
+}
+
+func TestUnionByNameUpdateDoc(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true, Doc: "identifier"},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ id, ok := applied.FindFieldByName("id")
+ require.True(t, ok)
+ assert.Equal(t, "identifier", id.Doc)
+}
+
+func TestUnionByNameUpdateWriteDefaultOnly(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true,
+ InitialDefault: int32(34), WriteDefault: int32(35),
+ },
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ id, ok := applied.FindFieldByName("id")
+ require.True(t, ok)
+ assert.Equal(t, int32(35), id.WriteDefault)
+ assert.Nil(t, id.InitialDefault)
+}
+
+func TestUnionByNameCaseSensitivity(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "ID", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ )
+
+ t.Run("case sensitive adds a new column", func(t *testing.T) {
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ _, ok := applied.FindFieldByName("id")
+ require.True(t, ok)
+ _, ok = applied.FindFieldByName("ID")
+ require.True(t, ok)
+ })
+
+ t.Run("case insensitive matches the existing column", func(t *testing.T) {
+ applied, err := NewUpdateSchema(unionTxn(t, current), false, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ assert.Len(t, applied.Fields(), 1)
+ id, ok := applied.FindFieldByName("id")
+ require.True(t, ok)
+ assert.False(t, id.Required)
+ })
+}
+
+func TestUnionByNameIdenticalSchemaIsNoop(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(current).Apply()
+ require.NoError(t, err)
+ assert.Equal(t, current.AsStruct(), applied.AsStruct())
+}
+
+// Variant is not a primitive and has no nested children, so,
+// it must be recognized as an equal-type no-op.
+func TestUnionByNameUnchangedVariantIsNoop(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "payload", Type: iceberg.VariantType{}, Required: false},
+ )
+ meta, err := NewMetadata(current, iceberg.UnpartitionedSpec, UnsortedSortOrder, "",
+ iceberg.Properties{PropertyFormatVersion: "3"})
+ require.NoError(t, err)
+ txn := New([]string{"tbl"}, meta, "", nil, nil).NewTransaction()
+
+ applied, err := NewUpdateSchema(txn, true, false).UnionByNameWith(current).Apply()
+ require.NoError(t, err)
+ assert.Equal(t, current.AsStruct(), applied.AsStruct())
+}
+
+func TestUnionByNameNilSchema(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(nil).Apply()
+ assert.Error(t, err)
+}
+
+func TestUnionByNameRejectsMapKeyChange(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.Int32,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.String, ValueRequired: false,
+ }, Required: false},
+ )
+
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.Int64,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.String, ValueRequired: false,
+ }, Required: false},
+ )
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.Error(t, err, "widening map key promotion int -> long must be rejected")
+}
+
+func TestUnionByNameIgnoresNarrowingMapKey(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.Int64,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.String, ValueRequired: false,
+ }, Required: false},
+ )
+
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.Int32,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.String, ValueRequired: false,
+ }, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ key, ok := applied.FindFieldByName("m.key")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.Int64, key.Type)
+}
+
+func TestUnionByNamePromoteMapValue(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.Int32, ValueRequired: false,
+ }, Required: false},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.Int64, ValueRequired: false,
+ }, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ value, ok := applied.FindFieldByName("m.value")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.Int64, value.Type)
+}
+
+func TestUnionByNameAddNewNestedStructWithChildren(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "address", Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 3, Name: "city", Type: iceberg.PrimitiveTypes.String, Required: false},
+ {ID: 4, Name: "zip", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ }, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ address, ok := applied.FindFieldByName("address")
+ require.True(t, ok)
+ assert.False(t, address.Required, "added struct must be optional")
+
+ city, ok := applied.FindFieldByName("address.city")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.String, city.Type)
+
+ zip, ok := applied.FindFieldByName("address.zip")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.String, zip.Type)
+}
+
+func TestUnionByNameAddFieldToListOfStruct(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "events", Type: &iceberg.ListType{
+ ElementID: 2, ElementRequired: false,
+ Element: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 3, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ },
+ },
+ }, Required: false},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "events", Type: &iceberg.ListType{
+ ElementID: 2, ElementRequired: false,
+ Element: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 3, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ {ID: 4, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ },
+ }, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ name, ok := applied.FindFieldByName("events.element.name")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.String, name.Type)
+
+ _, ok = applied.FindFieldByName("events.element.id")
+ require.True(t, ok)
+}
+
+func TestUnionByNameAddFieldToMapValueStruct(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 3, ValueRequired: false,
+ ValueType: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 4, Name: "a", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ },
+ },
+ }, Required: false},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "m", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 3, ValueRequired: false,
+ ValueType: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 4, Name: "a", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ {ID: 5, Name: "b", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ },
+ }, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ b, ok := applied.FindFieldByName("m.value.b")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.PrimitiveTypes.String, b.Type)
+
+ _, ok = applied.FindFieldByName("m.value.a")
+ require.True(t, ok)
+}
+
+func TestUnionByNameEmptyDocDoesNotClearExisting(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true, Doc: "original"},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true, Doc: ""},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ id, ok := applied.FindFieldByName("id")
+ require.True(t, ok)
+ assert.Equal(t, "original", id.Doc)
+}
+
+func TestUnionByNameAddedColumnPreservesInitialDefault(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{
+ ID: 2, Name: "count", Type: iceberg.PrimitiveTypes.Int32,
+ Required: false, InitialDefault: int32(42), WriteDefault: int32(42),
+ },
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ count, ok := applied.FindFieldByName("count")
+ require.True(t, ok)
+ assert.Equal(t, int32(42), count.InitialDefault)
+ assert.Equal(t, int32(42), count.WriteDefault)
+}
+
+func TestUnionByNameRejectsInvalidAddedColumnDefault(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+
+ incomingInitial := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{
+ ID: 2, Name: "count", Type: iceberg.PrimitiveTypes.Int32,
+ Required: false, InitialDefault: "not-an-int",
+ },
+ )
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incomingInitial).Apply()
+ require.ErrorContains(t, err, "invalid initial-default for count")
+
+ incomingWrite := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{
+ ID: 2, Name: "when", Type: iceberg.PrimitiveTypes.Date,
+ Required: false, WriteDefault: "not-a-date",
+ },
+ )
+ _, err = NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incomingWrite).Apply()
+ require.ErrorContains(t, err, "invalid write-default for when")
+}
+
+func TestUnionByNameRejectsInvalidWriteDefaultUpdate(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true,
+ WriteDefault: "not-an-int",
+ },
+ )
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.ErrorContains(t, err, "invalid write-default for id")
+}
+
+func TestUnionByNameRejectsInvalidNestedAddedColumnDefault(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+
+ incomingStruct := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "addr", Required: false, Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {
+ ID: 3, Name: "zip", Type: iceberg.PrimitiveTypes.Int32,
+ Required: false, InitialDefault: "not-an-int",
+ },
+ },
+ }},
+ )
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incomingStruct).Apply()
+ require.ErrorContains(t, err, "invalid initial-default for zip")
+
+ incomingMap := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "events", Required: false, Type: &iceberg.MapType{
+ KeyID: 3, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 4, ValueRequired: false, ValueType: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {
+ ID: 5, Name: "when", Type: iceberg.PrimitiveTypes.Date,
+ Required: false, WriteDefault: "not-a-date",
+ },
+ },
+ },
+ }},
+ )
+ _, err = NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incomingMap).Apply()
+ require.ErrorContains(t, err, "invalid write-default for when")
+
+ incomingMapKey := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "lookup", Required: false, Type: &iceberg.MapType{
+ KeyID: 3, KeyType: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {
+ ID: 4, Name: "code", Type: iceberg.PrimitiveTypes.Int32,
+ Required: false, InitialDefault: "not-an-int",
+ },
+ },
+ },
+ ValueID: 5, ValueRequired: false, ValueType: iceberg.PrimitiveTypes.String,
+ }},
+ )
+ _, err = NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incomingMapKey).Apply()
+ require.ErrorContains(t, err, "invalid initial-default for code")
+}
+
+func TestUnionByNameRejectsMistypedNativeDefault(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{
+ ID: 2, Name: "count", Type: iceberg.PrimitiveTypes.Int32,
+ Required: false, InitialDefault: int64(5),
+ },
+ )
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.ErrorContains(t, err, "invalid initial-default for count")
+}
+
+func TestUnionByNamePromoteDecimalKeepsWiderWriteDefault(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "amount", Type: iceberg.DecimalTypeOf(9, 2), Required: true},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{
+ ID: 1, Name: "amount", Type: iceberg.DecimalTypeOf(18, 2), Required: true,
+ WriteDefault: "123456789.12",
+ },
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.NoError(t, err)
+
+ amount, ok := applied.FindFieldByName("amount")
+ require.True(t, ok)
+ assert.Equal(t, iceberg.DecimalTypeOf(18, 2), amount.Type)
+ assert.Equal(t, "123456789.12", amount.WriteDefault)
+}
+
+func TestUnionByNameAddedComplexColumnDefault(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+
+ structType := func() *iceberg.StructType {
+ return &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 3, Name: "zip", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ }
+ }
+
+ valid := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{
+ ID: 2, Name: "addr", Required: false, Type: structType(),
+ InitialDefault: map[string]any{"zip": "00000"},
+ },
+ )
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(valid).Apply()
+ require.NoError(t, err)
+ addr, ok := applied.FindFieldByName("addr")
+ require.True(t, ok)
+ assert.Equal(t, map[string]any{"zip": "00000"}, addr.InitialDefault)
+
+ invalid := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{
+ ID: 2, Name: "addr", Required: false, Type: structType(),
+ InitialDefault: "not-json",
+ },
+ )
+ _, err = NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(invalid).Apply()
+ require.ErrorContains(t, err, "must have null or JSON object initial-default")
+}
+
+func TestUnionByNameCannotExtendStructAddedByPriorOp(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+
+ extend := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "addr", Required: false, Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 3, Name: "zip", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ }},
+ )
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, true).
+ AddColumn([]string{"addr"}, &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 5, Name: "city", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ }, "", false, nil).
+ UnionByNameWith(extend).
+ Apply()
+ require.ErrorContains(t, err, "field already exists in adds: addr")
+}
+
+func TestUnionByNameMirroredSchemas(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 9, Name: "s", Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 8, Name: "inner", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ }, Required: false},
+ iceberg.NestedField{ID: 5, Name: "a", Type: iceberg.PrimitiveTypes.String, Required: false},
+ iceberg.NestedField{ID: 4, Name: "b", Type: iceberg.PrimitiveTypes.String, Required: false},
+ )
+ mirrored := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "s", Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 2, Name: "inner", Type: iceberg.PrimitiveTypes.String, Required: false},
+ },
+ }, Required: false},
+ iceberg.NestedField{ID: 3, Name: "a", Type: iceberg.PrimitiveTypes.String, Required: false},
+ iceberg.NestedField{ID: 4, Name: "b", Type: iceberg.PrimitiveTypes.String, Required: false},
+ )
+
+ txn := unionTxn(t, current)
+ baseline := txn.meta.CurrentSchema().AsStruct()
+
+ applied, err := NewUpdateSchema(txn, true, false).UnionByNameWith(mirrored).Apply()
+ require.NoError(t, err)
+ assert.Equal(t, baseline, applied.AsStruct(),
+ "union operates by name; differing incoming field IDs must not cause changes")
+}
+
+func TestUnionByNameInvalidListElementTypeChange(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "vals", Type: &iceberg.ListType{
+ ElementID: 2, Element: iceberg.PrimitiveTypes.String, ElementRequired: false,
+ }, Required: false},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "vals", Type: &iceberg.ListType{
+ ElementID: 2, Element: iceberg.PrimitiveTypes.Int64, ElementRequired: false,
+ }, Required: false},
+ )
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ assert.Error(t, err, "list<string> -> list<long> is not a valid promotion")
+}
+
+// Cross-kind changes must be rejected outright
+func TestUnionByNameRejectsCrossKindChanges(t *testing.T) {
+ cases := []struct {
+ name string
+ existing iceberg.NestedField
+ incoming iceberg.NestedField
+ errContains string
+ }{
+ {
+ name: "list -> map",
+ existing: iceberg.NestedField{ID: 1, Name: "c", Type: &iceberg.ListType{
+ ElementID: 2, Element: iceberg.PrimitiveTypes.Int32, ElementRequired: false,
+ }, Required: false},
+ incoming: iceberg.NestedField{ID: 1, Name: "c", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.Int32, ValueRequired: false,
+ }, Required: false},
+ errContains: "cannot change column type",
+ },
+ {
+ name: "struct -> list",
+ existing: iceberg.NestedField{ID: 1, Name: "c", Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 2, Name: "x", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ },
+ }, Required: false},
+ incoming: iceberg.NestedField{ID: 1, Name: "c", Type: &iceberg.ListType{
+ ElementID: 2, Element: iceberg.PrimitiveTypes.Int32, ElementRequired: false,
+ }, Required: false},
+ errContains: "cannot change column type",
+ },
+ {
+ name: "map -> struct",
+ existing: iceberg.NestedField{ID: 1, Name: "c", Type: &iceberg.MapType{
+ KeyID: 2, KeyType: iceberg.PrimitiveTypes.String,
+ ValueID: 3, ValueType: iceberg.PrimitiveTypes.Int32, ValueRequired: false,
+ }, Required: false},
+ incoming: iceberg.NestedField{ID: 1, Name: "c", Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 2, Name: "x", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ },
+ }, Required: false},
+ errContains: "cannot change column type",
+ },
+ {
+ name: "primitive -> struct",
+ existing: iceberg.NestedField{ID: 1, Name: "c", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ incoming: iceberg.NestedField{ID: 1, Name: "c", Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 2, Name: "x", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ },
+ }, Required: false},
+ errContains: "cannot change column type",
+ },
+ {
+ name: "struct -> primitive",
+ existing: iceberg.NestedField{ID: 1, Name: "c", Type: &iceberg.StructType{
+ FieldList: []iceberg.NestedField{
+ {ID: 2, Name: "x", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ },
+ }, Required: false},
+ incoming: iceberg.NestedField{ID: 1, Name: "c", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ errContains: "cannot change column type",
+ },
+ {
+ name: "list -> primitive",
+ existing: iceberg.NestedField{ID: 1, Name: "c", Type: &iceberg.ListType{
+ ElementID: 2, Element: iceberg.PrimitiveTypes.String, ElementRequired: false,
+ }, Required: false},
+ incoming: iceberg.NestedField{ID: 1, Name: "c", Type: iceberg.PrimitiveTypes.String, Required: false},
+ errContains: "cannot change column type",
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ current := iceberg.NewSchema(1, tc.existing)
+ incoming := iceberg.NewSchema(1, tc.incoming)
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), tc.errContains)
+ })
+ }
+}
+
+func TestUnionByNameRejectsInvalidPromotions(t *testing.T) {
+ cases := []struct {
+ name string
+ existing iceberg.Type
+ incoming iceberg.Type
+ }{
+ {"string -> binary", iceberg.PrimitiveTypes.String, iceberg.PrimitiveTypes.Binary},
+ {"binary -> string", iceberg.PrimitiveTypes.Binary, iceberg.PrimitiveTypes.String},
+ {"fixed[16] -> uuid", iceberg.FixedTypeOf(16), iceberg.PrimitiveTypes.UUID},
+ {"decimal scale widening", iceberg.DecimalTypeOf(10, 1), iceberg.DecimalTypeOf(10, 2)},
+ {"string -> int", iceberg.PrimitiveTypes.String, iceberg.PrimitiveTypes.Int32},
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "c", Type: tc.existing, Required: false},
+ )
+ incoming := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "c", Type: tc.incoming, Required: false},
+ )
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).UnionByNameWith(incoming).Apply()
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "cannot change column type")
+ })
+ }
+}
+
+// Two sequential UnionByNameWith calls on the same UpdateSchema must both
+// take effect when they contribute disjoint additions.
+func TestUnionByNameSequentialCalls(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ first := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
+ )
+ // Note: second schema intentionally omits "name" so we don't retry an add
+ // that is already staged.
+ second := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "age", Type: iceberg.PrimitiveTypes.Int32, Required: false},
+ )
+
+ applied, err := NewUpdateSchema(unionTxn(t, current), true, false).
+ UnionByNameWith(first).
+ UnionByNameWith(second).
+ Apply()
+ require.NoError(t, err)
+
+ _, ok := applied.FindFieldByName("name")
+ require.True(t, ok, "field added by first union must survive the second call")
+ _, ok = applied.FindFieldByName("age")
+ require.True(t, ok, "field added by second union must be present")
+}
+
+// A second UnionByNameWith that re-adds a field already staged by the first
+// must error rather than silently double-add.
+func TestUnionByNameRejectsOverlappingAdds(t *testing.T) {
+ current := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ )
+ overlap := iceberg.NewSchema(1,
+ iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+ iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
+ )
+
+ _, err := NewUpdateSchema(unionTxn(t, current), true, false).
+ UnionByNameWith(overlap).
+ UnionByNameWith(overlap).
+ Apply()
+ require.Error(t, err, "re-adding a field already pending must be rejected")
+ assert.Contains(t, err.Error(), "name")
+}