badalprasadsingh commented on code in PR #1359:
URL: https://github.com/apache/iceberg-go/pull/1359#discussion_r3596333066
##########
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:
Done.
1. This now returns `error` propagated by `unionUpdateColumn`.
2. Rejects writes against a deleted field.
3. Rejects non-primitive columns.
4. For your last concern, here is a limitation, `NestedField.WriteDefault`
is `any`, and `iceberg.NewLiteral` is generic / compile-time only - there is no
runtime `any` to `Literal` converter to check compatibility. <br>
Here, `updateColumn` has existing default validation which only fires when
the caller supplies an `iceberg.Literal`, which is unavailable here because
we've received an already-unwrapped `any` from the incoming schema. <br>
Have added a `TODO` for it. Happy to open a PR for it!
--
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]
