badalprasadsingh commented on code in PR #1359:
URL: https://github.com/apache/iceberg-go/pull/1359#discussion_r3637185884
##########
table/update_schema.go:
##########
@@ -631,6 +632,368 @@ 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.
+//
+// The change is queued and applied with other pending updates on Apply/Commit,
+// so fields added by an earlier op in the same UpdateSchema are visible here.
+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 _, isPrimitive := newField.Type.(iceberg.PrimitiveType);
!isPrimitive {
+ if newField.InitialDefault != nil || newField.WriteDefault !=
nil {
+ return fmt.Errorf("default values are not supported for
%s", newField.Type)
+ }
+ if err := validateNestedDefaults(newField.Type); err != nil {
+ return err
+ }
+ } else {
+ if err := validateDefaultValue(newField.Type,
newField.InitialDefault); err != nil {
+ return fmt.Errorf("invalid initial-default for %s: %w",
fullName, err)
+ }
+ if err := validateDefaultValue(newField.Type,
newField.WriteDefault); err != nil {
+ return fmt.Errorf("invalid write-default for %s: %w",
fullName, err)
Review Comment:
Done.
Added `InitialDefault` validation on the add path as well, rejecting native
defaults with mismatched types (e.g. `int64` for `Int32Type`) and covered it
with a **test**.
--
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]
