diff --git a/schema_compatibility.go b/schema_compatibility.go new file mode 100644 index 000000000..fe4468848 --- /dev/null +++ b/schema_compatibility.go @@ -0,0 +1,255 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package iceberg + +import ( + "fmt" + "strings" +) + +// IsPromotionAllowed reports whether schema evolution may change a column +// from one type to another: int to long, float to double, or widening a +// decimal's precision with the same scale. +func IsPromotionAllowed(from, to PrimitiveType) bool { + if from.Equals(to) { + return true + } + + switch f := from.(type) { + case Int32Type: + _, ok := to.(Int64Type) + + return ok + case Float32Type: + _, ok := to.(Float64Type) + + return ok + case DecimalType: + t, ok := to.(DecimalType) + if !ok { + return false + } + + return f.Scale() == t.Scale() && f.Precision() <= t.Precision() + } + + return false +} + +// ReadCompatibilityErrors returns the problems with reading data written +// with writeSchema using readSchema. Fields are matched by ID and field +// order is not checked. +func ReadCompatibilityErrors(readSchema, writeSchema *Schema) ([]string, error) { + return checkCompatibility(readSchema, writeSchema, false, true) +} + +// WriteCompatibilityErrors returns the problems with writing data in +// writeSchema to a table whose schema is readSchema. If checkOrdering is +// set, reordered fields are also reported. +func WriteCompatibilityErrors(readSchema, writeSchema *Schema, checkOrdering bool) ([]string, error) { + return checkCompatibility(readSchema, writeSchema, checkOrdering, true) +} + +// TypeCompatibilityErrors is WriteCompatibilityErrors without the struct +// field nullability check. Required list elements and map values are still +// checked. +func TypeCompatibilityErrors(readSchema, writeSchema *Schema, checkOrdering bool) ([]string, error) { + return checkCompatibility(readSchema, writeSchema, checkOrdering, false) +} + +func checkCompatibility(readSchema, writeSchema *Schema, checkOrdering, checkNullability bool) ([]string, error) { + if readSchema == nil || writeSchema == nil { + return nil, fmt.Errorf("%w: cannot check compatibility against nil schema", ErrInvalidArgument) + } + + return PreOrderVisit(readSchema, &compatibilityChecker{ + schema: writeSchema, + checkOrdering: checkOrdering, + checkNullability: checkNullability, + }) +} + +// compatibilityChecker walks the read schema, tracking the matching type in +// the write schema. Errors starting with ":" get the enclosing field's name +// prepended; others are joined to it with ".". +type compatibilityChecker struct { + schema *Schema + checkOrdering bool + checkNullability bool + + current Type + // PreOrderVisit sends list elements and map keys/values through Field, + // but only struct fields should be looked up by ID. + inContainer bool +} + +func (c *compatibilityChecker) Schema(_ *Schema, structErrors func() []string) []string { + st := c.schema.asStructRef() + c.current = &st + defer func() { c.current = nil }() + + return structErrors() +} + +func (c *compatibilityChecker) Struct(readStruct StructType, fieldErrors []func() []string) []string { + st, ok := c.current.(*StructType) + if !ok { + return []string{fmt.Sprintf(": %s cannot be read as a struct", c.current)} + } + + var errs []string + for _, fieldErrs := range fieldErrors { + errs = append(errs, fieldErrs()...) + } + + if c.checkOrdering { + ordinals := make(map[int]int, len(st.FieldList)) + for i, f := range st.FieldList { + ordinals[f.ID] = i + } + + lastOrdinal, lastName := -1, "" + for _, readField := range readStruct.FieldList { + ordinal, ok := ordinals[readField.ID] + if !ok { + continue + } + if lastOrdinal >= ordinal { + errs = append(errs, fmt.Sprintf("%s is out of order, before %s", readField.Name, lastName)) + } + lastOrdinal, lastName = ordinal, readField.Name + } + } + + return errs +} + +func (c *compatibilityChecker) Field(readField NestedField, fieldErrors func() []string) []string { + if c.inContainer { + c.inContainer = false + + return fieldErrors() + } + + st, ok := c.current.(*StructType) + if !ok { + return nil // Struct() already reported the mismatch + } + + var ( + writeField NestedField + found bool + ) + for _, f := range st.FieldList { + if f.ID == readField.ID { + writeField, found = f, true + + break + } + } + + if !found { + if readField.Required { + return []string{readField.Name + " is required, but is missing"} + } + + // an optional field is read as nulls + return nil + } + + c.current = writeField.Type + defer func() { c.current = st }() + + var errs []string + if c.checkNullability && readField.Required && !writeField.Required { + errs = append(errs, readField.Name+" should be required, but is optional") + } + + for _, err := range fieldErrors() { + if strings.HasPrefix(err, ":") { + errs = append(errs, readField.Name+err) + } else { + errs = append(errs, readField.Name+"."+err) + } + } + + return errs +} + +func (c *compatibilityChecker) List(readList ListType, elementErrors func() []string) []string { + list, ok := c.current.(*ListType) + if !ok { + return []string{fmt.Sprintf(": %s cannot be read as a list", c.current)} + } + + var errs []string + if readList.ElementRequired && !list.ElementRequired { + errs = append(errs, ": elements should be required, but are optional") + } + + c.current, c.inContainer = list.Element, true + defer func() { c.current = list }() + + return append(errs, elementErrors()...) +} + +func (c *compatibilityChecker) Map(readMap MapType, keyErrors, valueErrors func() []string) []string { + m, ok := c.current.(*MapType) + if !ok { + return []string{fmt.Sprintf(": %s cannot be read as a map", c.current)} + } + defer func() { c.current = m }() + + var errs []string + if readMap.ValueRequired && !m.ValueRequired { + errs = append(errs, ": values should be required, but are optional") + } + + c.current, c.inContainer = m.KeyType, true + errs = append(errs, keyErrors()...) + + c.current, c.inContainer = m.ValueType, true + + return append(errs, valueErrors()...) +} + +func (c *compatibilityChecker) Primitive(readPrimitive PrimitiveType) []string { + if c.current.Equals(readPrimitive) { + return nil + } + + writePrimitive, ok := c.current.(PrimitiveType) + if !ok { + return []string{fmt.Sprintf(": %s cannot be read as a %s", c.current, readPrimitive)} + } + + if !IsPromotionAllowed(writePrimitive, readPrimitive) { + return []string{fmt.Sprintf(": %s cannot be promoted to %s", writePrimitive, readPrimitive)} + } + + return nil +} + +func (c *compatibilityChecker) Variant(readVariant VariantType) []string { + if _, ok := c.current.(VariantType); ok { + return nil + } + + // promotion to variant is not allowed + return []string{fmt.Sprintf(": %s cannot be read as a %s", c.current, readVariant)} +} diff --git a/schema_compatibility_test.go b/schema_compatibility_test.go new file mode 100644 index 000000000..6fbde7983 --- /dev/null +++ b/schema_compatibility_test.go @@ -0,0 +1,531 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package iceberg_test + +import ( + "testing" + + "github.com/apache/iceberg-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Ported from Java's TestReadabilityChecks. + +func compatPrimitives(t *testing.T) []iceberg.PrimitiveType { + t.Helper() + + geomCRS84, err := iceberg.GeometryTypeOf(iceberg.DefaultGeoCRS) + require.NoError(t, err) + geom3857, err := iceberg.GeometryTypeOf("srid:3857") + require.NoError(t, err) + geogCRS84, err := iceberg.GeographyTypeOf(iceberg.DefaultGeoCRS, "spherical") + require.NoError(t, err) + geog4269, err := iceberg.GeographyTypeOf("srid:4269", "spherical") + require.NoError(t, err) + geog4269Karney, err := iceberg.GeographyTypeOf("srid:4269", "karney") + require.NoError(t, err) + + return []iceberg.PrimitiveType{ + iceberg.PrimitiveTypes.Bool, + iceberg.PrimitiveTypes.Int32, + iceberg.PrimitiveTypes.Int64, + iceberg.PrimitiveTypes.Float32, + iceberg.PrimitiveTypes.Float64, + iceberg.PrimitiveTypes.Date, + iceberg.PrimitiveTypes.Time, + iceberg.PrimitiveTypes.Timestamp, + iceberg.PrimitiveTypes.TimestampTz, + iceberg.PrimitiveTypes.TimestampNs, + iceberg.PrimitiveTypes.TimestampTzNs, + iceberg.PrimitiveTypes.String, + iceberg.PrimitiveTypes.UUID, + iceberg.FixedTypeOf(3), + iceberg.FixedTypeOf(4), + iceberg.PrimitiveTypes.Binary, + iceberg.DecimalTypeOf(9, 2), + iceberg.DecimalTypeOf(11, 2), + iceberg.DecimalTypeOf(9, 3), + geomCRS84, + geom3857, + geogCRS84, + geog4269, + geog4269Karney, + } +} + +func required(id int, name string, typ iceberg.Type) iceberg.NestedField { + return iceberg.NestedField{ID: id, Name: name, Type: typ, Required: true} +} + +func optional(id int, name string, typ iceberg.Type) iceberg.NestedField { + return iceberg.NestedField{ID: id, Name: name, Type: typ} +} + +func schemaOf(fields ...iceberg.NestedField) *iceberg.Schema { + return iceberg.NewSchema(0, fields...) +} + +func writeErrors(t *testing.T, read, write *iceberg.Schema) []string { + t.Helper() + + errs, err := iceberg.WriteCompatibilityErrors(read, write, true) + require.NoError(t, err) + + return errs +} + +func TestCompatibilityPrimitiveTypes(t *testing.T) { + primitives := compatPrimitives(t) + for _, from := range primitives { + fromSchema := schemaOf(required(1, "from_field", from)) + for _, to := range primitives { + errs := writeErrors(t, schemaOf(required(1, "to_field", to)), fromSchema) + + if iceberg.IsPromotionAllowed(from, to) { + assert.Empty(t, errs, "%s -> %s", from, to) + } else { + require.Len(t, errs, 1, "%s -> %s", from, to) + assert.Contains(t, errs[0], "cannot be promoted to") + } + } + + structSchema := schemaOf(required(1, "struct_field", &iceberg.StructType{ + FieldList: []iceberg.NestedField{required(2, "from", from)}, + })) + errs := writeErrors(t, structSchema, fromSchema) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "cannot be read as a struct") + + listSchema := schemaOf(required(1, "list_field", &iceberg.ListType{ + ElementID: 2, Element: from, ElementRequired: true, + })) + errs = writeErrors(t, listSchema, fromSchema) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "cannot be read as a list") + + mapSchema := schemaOf(required(1, "map_field", &iceberg.MapType{ + KeyID: 2, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 3, ValueType: from, ValueRequired: true, + })) + errs = writeErrors(t, mapSchema, fromSchema) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "cannot be read as a map") + } +} + +func TestIsPromotionAllowed(t *testing.T) { + tests := []struct { + from, to iceberg.PrimitiveType + allowed bool + }{ + {iceberg.PrimitiveTypes.Int32, iceberg.PrimitiveTypes.Int32, true}, + {iceberg.PrimitiveTypes.Int32, iceberg.PrimitiveTypes.Int64, true}, + {iceberg.PrimitiveTypes.Int64, iceberg.PrimitiveTypes.Int32, false}, + {iceberg.PrimitiveTypes.Float32, iceberg.PrimitiveTypes.Float64, true}, + {iceberg.PrimitiveTypes.Float64, iceberg.PrimitiveTypes.Float32, false}, + {iceberg.DecimalTypeOf(9, 2), iceberg.DecimalTypeOf(11, 2), true}, + {iceberg.DecimalTypeOf(11, 2), iceberg.DecimalTypeOf(9, 2), false}, + {iceberg.DecimalTypeOf(9, 2), iceberg.DecimalTypeOf(9, 3), false}, + {iceberg.PrimitiveTypes.String, iceberg.PrimitiveTypes.Binary, false}, + {iceberg.FixedTypeOf(16), iceberg.PrimitiveTypes.UUID, false}, + {iceberg.PrimitiveTypes.Date, iceberg.PrimitiveTypes.Timestamp, false}, + } + + for _, tt := range tests { + assert.Equal(t, tt.allowed, iceberg.IsPromotionAllowed(tt.from, tt.to), "%s -> %s", tt.from, tt.to) + } +} + +func TestCompatibilityVariantToVariant(t *testing.T) { + errs := writeErrors(t, + schemaOf(required(1, "to_field", iceberg.VariantType{})), + schemaOf(required(1, "from_field", iceberg.VariantType{}))) + assert.Empty(t, errs) +} + +func TestCompatibilityIncompatibleTypesToVariant(t *testing.T) { + from := []iceberg.Type{ + &iceberg.StructType{FieldList: []iceberg.NestedField{required(1, "from", iceberg.PrimitiveTypes.Int32)}}, + &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 2, ValueType: iceberg.PrimitiveTypes.Int32, ValueRequired: true, + }, + &iceberg.ListType{ElementID: 1, Element: iceberg.PrimitiveTypes.String, ElementRequired: true}, + } + for _, p := range compatPrimitives(t) { + from = append(from, p) + } + + for _, typ := range from { + errs := writeErrors(t, + schemaOf(required(3, "to_field", iceberg.VariantType{})), + schemaOf(required(3, "from_field", typ))) + require.Len(t, errs, 1, "%s", typ) + assert.Contains(t, errs[0], "cannot be read as a variant") + } +} + +func TestCompatibilityRequiredSchemaField(t *testing.T) { + write := schemaOf(optional(1, "from_field", iceberg.PrimitiveTypes.Int32)) + read := schemaOf(required(1, "to_field", iceberg.PrimitiveTypes.Int32)) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "should be required, but is optional") +} + +func TestCompatibilityMissingSchemaField(t *testing.T) { + write := schemaOf(required(0, "other_field", iceberg.PrimitiveTypes.Int32)) + read := schemaOf(required(1, "to_field", iceberg.PrimitiveTypes.Int32)) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "is required, but is missing") +} + +func nestedStruct(fields ...iceberg.NestedField) *iceberg.StructType { + return &iceberg.StructType{FieldList: fields} +} + +func TestCompatibilityRequiredStructField(t *testing.T) { + write := schemaOf(required(0, "nested", nestedStruct(optional(1, "from_field", iceberg.PrimitiveTypes.Int32)))) + read := schemaOf(required(0, "nested", nestedStruct(required(1, "to_field", iceberg.PrimitiveTypes.Int32)))) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "should be required, but is optional") +} + +func TestCompatibilityMissingRequiredStructField(t *testing.T) { + write := schemaOf(required(0, "nested", nestedStruct(optional(2, "from_field", iceberg.PrimitiveTypes.Int32)))) + read := schemaOf(required(0, "nested", nestedStruct(required(1, "to_field", iceberg.PrimitiveTypes.Int32)))) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "is required, but is missing") +} + +func TestCompatibilityMissingOptionalStructField(t *testing.T) { + write := schemaOf(required(0, "nested", nestedStruct(required(2, "from_field", iceberg.PrimitiveTypes.Int32)))) + read := schemaOf(required(0, "nested", nestedStruct(optional(1, "to_field", iceberg.PrimitiveTypes.Int32)))) + + assert.Empty(t, writeErrors(t, read, write)) +} + +func TestCompatibilityIncompatibleStructField(t *testing.T) { + write := schemaOf(required(0, "nested", nestedStruct(required(1, "from_field", iceberg.PrimitiveTypes.Int32)))) + read := schemaOf(required(0, "nested", nestedStruct(required(1, "to_field", iceberg.PrimitiveTypes.Float32)))) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "cannot be promoted to float") +} + +func TestCompatibilityIncompatibleStructAndPrimitive(t *testing.T) { + write := schemaOf(required(0, "nested", nestedStruct(required(1, "from_field", iceberg.PrimitiveTypes.String)))) + read := schemaOf(required(0, "nested", iceberg.PrimitiveTypes.String)) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Equal(t, "nested: struct<1: from_field: required string> cannot be read as a string", errs[0]) +} + +func TestCompatibilityMultipleErrors(t *testing.T) { + // required field is optional and cannot be promoted to the read type + write := schemaOf(required(0, "nested", nestedStruct(optional(1, "from_field", iceberg.PrimitiveTypes.Int32)))) + read := schemaOf(required(0, "nested", nestedStruct(required(1, "to_field", iceberg.PrimitiveTypes.Float32)))) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 2) + assert.Contains(t, errs[0], "should be required, but is optional") + assert.Contains(t, errs[1], "cannot be promoted to float") +} + +func TestCompatibilityRequiredMapValue(t *testing.T) { + write := schemaOf(required(0, "map_field", &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 2, ValueType: iceberg.PrimitiveTypes.Int32, + })) + read := schemaOf(required(0, "map_field", &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 2, ValueType: iceberg.PrimitiveTypes.Int32, ValueRequired: true, + })) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "values should be required, but are optional") +} + +func TestCompatibilityIncompatibleMapKey(t *testing.T) { + write := schemaOf(required(0, "map_field", &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.Int32, + ValueID: 2, ValueType: iceberg.PrimitiveTypes.String, + })) + read := schemaOf(required(0, "map_field", &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.Float64, + ValueID: 2, ValueType: iceberg.PrimitiveTypes.String, + })) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "cannot be promoted to double") +} + +func TestCompatibilityIncompatibleMapValue(t *testing.T) { + write := schemaOf(required(0, "map_field", &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 2, ValueType: iceberg.PrimitiveTypes.Int32, + })) + read := schemaOf(required(0, "map_field", &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 2, ValueType: iceberg.PrimitiveTypes.Float64, + })) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "cannot be promoted to double") +} + +func TestCompatibilityIncompatibleMapAndPrimitive(t *testing.T) { + write := schemaOf(required(0, "map_field", &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 2, ValueType: iceberg.PrimitiveTypes.Int32, + })) + read := schemaOf(required(0, "map_field", iceberg.PrimitiveTypes.String)) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Equal(t, "map_field: map cannot be read as a string", errs[0]) +} + +func TestCompatibilityRequiredListElement(t *testing.T) { + write := schemaOf(required(0, "list_field", &iceberg.ListType{ElementID: 1, Element: iceberg.PrimitiveTypes.Int32})) + read := schemaOf(required(0, "list_field", &iceberg.ListType{ElementID: 1, Element: iceberg.PrimitiveTypes.Int32, ElementRequired: true})) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "elements should be required, but are optional") +} + +func TestCompatibilityIncompatibleListElement(t *testing.T) { + write := schemaOf(required(0, "list_field", &iceberg.ListType{ElementID: 1, Element: iceberg.PrimitiveTypes.Int32})) + read := schemaOf(required(0, "list_field", &iceberg.ListType{ElementID: 1, Element: iceberg.PrimitiveTypes.String})) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Contains(t, errs[0], "cannot be promoted to string") +} + +func TestCompatibilityIncompatibleListAndPrimitive(t *testing.T) { + write := schemaOf(required(0, "list_field", &iceberg.ListType{ElementID: 1, Element: iceberg.PrimitiveTypes.Int32})) + read := schemaOf(required(0, "list_field", iceberg.PrimitiveTypes.String)) + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Equal(t, "list_field: list cannot be read as a string", errs[0]) +} + +func TestCompatibilityIncompatibleNestedListOfMap(t *testing.T) { + write := schemaOf(required(0, "list_field", &iceberg.ListType{ElementID: 1, Element: &iceberg.MapType{ + KeyID: 2, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 3, ValueType: iceberg.PrimitiveTypes.Int32, + }})) + read := schemaOf(required(0, "list_field", &iceberg.ListType{ElementID: 1, Element: &iceberg.MapType{ + KeyID: 2, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 3, ValueType: iceberg.PrimitiveTypes.String, + }})) + + errs := writeErrors(t, read, write) + assert.Equal(t, []string{"list_field: int cannot be promoted to string"}, errs) +} + +func TestCompatibilityIncompatibleMapOfStruct(t *testing.T) { + write := schemaOf(required(0, "map_field", &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 2, ValueType: nestedStruct(required(3, "x", iceberg.PrimitiveTypes.Int32)), + })) + read := schemaOf(required(0, "map_field", &iceberg.MapType{ + KeyID: 1, KeyType: iceberg.PrimitiveTypes.String, + ValueID: 2, ValueType: nestedStruct(required(3, "x", iceberg.PrimitiveTypes.Float64)), + })) + + errs := writeErrors(t, read, write) + assert.Equal(t, []string{"map_field.x: int cannot be promoted to double"}, errs) +} + +func reorderedSchemas() (read, write *iceberg.Schema) { + read = schemaOf(required(0, "nested", nestedStruct( + required(1, "field_a", iceberg.PrimitiveTypes.Int32), + required(2, "field_b", iceberg.PrimitiveTypes.Int32)))) + write = schemaOf(required(0, "nested", nestedStruct( + required(2, "field_b", iceberg.PrimitiveTypes.Int32), + required(1, "field_a", iceberg.PrimitiveTypes.Int32)))) + + return read, write +} + +func TestCompatibilityDifferentFieldOrdering(t *testing.T) { + read, write := reorderedSchemas() + + errs, err := iceberg.WriteCompatibilityErrors(read, write, false) + require.NoError(t, err) + assert.Empty(t, errs) +} + +func TestCompatibilityStructWriteReordering(t *testing.T) { + // writes should not reorder fields + read, write := reorderedSchemas() + + errs := writeErrors(t, read, write) + require.Len(t, errs, 1) + assert.Equal(t, "nested.field_b is out of order, before field_a", errs[0]) +} + +func TestCompatibilityStructWriteReorderingRenamed(t *testing.T) { + // the message should use read-side names only + read, _ := reorderedSchemas() + write := schemaOf(required(0, "nested", nestedStruct( + required(2, "old_b", iceberg.PrimitiveTypes.Int32), + required(1, "old_a", iceberg.PrimitiveTypes.Int32)))) + + errs := writeErrors(t, read, write) + assert.Equal(t, []string{"nested.field_b is out of order, before field_a"}, errs) +} + +func TestCompatibilityStructReadReordering(t *testing.T) { + // reads should allow reordering + read, write := reorderedSchemas() + + errs, err := iceberg.ReadCompatibilityErrors(read, write) + require.NoError(t, err) + assert.Empty(t, errs) +} + +func TestCompatibilityCheckNullabilityRequiredSchemaField(t *testing.T) { + write := schemaOf(optional(1, "from_field", iceberg.PrimitiveTypes.Int32)) + read := schemaOf(required(1, "to_field", iceberg.PrimitiveTypes.Int32)) + + errs, err := iceberg.TypeCompatibilityErrors(read, write, true) + require.NoError(t, err) + assert.Empty(t, errs) +} + +func TestCompatibilityCheckNullabilityRequiredStructField(t *testing.T) { + write := schemaOf(required(0, "nested", nestedStruct(optional(1, "from_field", iceberg.PrimitiveTypes.Int32)))) + read := schemaOf(required(0, "nested", nestedStruct(required(1, "to_field", iceberg.PrimitiveTypes.Int32)))) + + errs, err := iceberg.TypeCompatibilityErrors(read, write, true) + require.NoError(t, err) + assert.Empty(t, errs) +} + +func TestCompatibilityTypeErrorsStillReportPromotion(t *testing.T) { + write := schemaOf(optional(1, "f", iceberg.PrimitiveTypes.Int32)) + read := schemaOf(required(1, "f", iceberg.PrimitiveTypes.String)) + + errs, err := iceberg.TypeCompatibilityErrors(read, write, true) + require.NoError(t, err) + assert.Equal(t, []string{"f: int cannot be promoted to string"}, errs) +} + +func TestReadCompatibilitySchemaEvolution(t *testing.T) { + current := schemaOf( + required(1, "id", iceberg.PrimitiveTypes.Int32), + optional(2, "name", iceberg.PrimitiveTypes.String), + optional(3, "tags", &iceberg.ListType{ElementID: 4, Element: iceberg.PrimitiveTypes.String}), + ) + + tests := []struct { + name string + proposed *iceberg.Schema + errs []string + }{ + { + name: "promote, rename, drop and add optional", + proposed: schemaOf( + required(1, "id", iceberg.PrimitiveTypes.Int64), + optional(2, "full_name", iceberg.PrimitiveTypes.String), + optional(5, "added", iceberg.PrimitiveTypes.Bool), + ), + }, + { + name: "add required column", + proposed: schemaOf( + required(1, "id", iceberg.PrimitiveTypes.Int32), + required(5, "added", iceberg.PrimitiveTypes.Bool), + ), + errs: []string{"added is required, but is missing"}, + }, + { + name: "make optional column required", + proposed: schemaOf( + required(1, "id", iceberg.PrimitiveTypes.Int32), + required(2, "name", iceberg.PrimitiveTypes.String), + ), + errs: []string{"name should be required, but is optional"}, + }, + { + name: "narrow type", + proposed: schemaOf( + required(1, "id", iceberg.PrimitiveTypes.String), + ), + errs: []string{"id: int cannot be promoted to string"}, + }, + { + name: "nested list element error is prefixed with the field name", + proposed: schemaOf( + required(1, "id", iceberg.PrimitiveTypes.Int32), + optional(3, "tags", &iceberg.ListType{ElementID: 4, Element: iceberg.PrimitiveTypes.Int32, ElementRequired: true}), + ), + errs: []string{ + "tags: elements should be required, but are optional", + "tags: string cannot be promoted to int", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + errs, err := iceberg.ReadCompatibilityErrors(tt.proposed, current) + require.NoError(t, err) + assert.Equal(t, tt.errs, errs) + }) + } +} + +func TestCompatibilityNestedFieldErrorPath(t *testing.T) { + write := schemaOf(required(0, "outer", nestedStruct( + required(1, "inner", nestedStruct(required(2, "leaf", iceberg.PrimitiveTypes.Int64)))))) + read := schemaOf(required(0, "outer", nestedStruct( + required(1, "inner", nestedStruct(required(2, "leaf", iceberg.PrimitiveTypes.Int32)))))) + + errs, err := iceberg.ReadCompatibilityErrors(read, write) + require.NoError(t, err) + assert.Equal(t, []string{"outer.inner.leaf: long cannot be promoted to int"}, errs) +} + +func TestCompatibilityNilSchema(t *testing.T) { + sc := schemaOf(required(1, "id", iceberg.PrimitiveTypes.Int32)) + + for _, tc := range []struct{ read, write *iceberg.Schema }{{nil, sc}, {sc, nil}} { + _, err := iceberg.ReadCompatibilityErrors(tc.read, tc.write) + assert.ErrorIs(t, err, iceberg.ErrInvalidArgument) + assert.ErrorContains(t, err, "cannot check compatibility against nil schema") + } +} diff --git a/table/update_schema.go b/table/update_schema.go index fead23a4c..dea77812f 100644 --- a/table/update_schema.go +++ b/table/update_schema.go @@ -834,7 +834,7 @@ func (u *UpdateSchema) unionUpdateColumn(path []string, existing, newField icebe // 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) { + if !existingIsPrim || !newIsPrim || !iceberg.IsPromotionAllowed(existingPrim, newPrim) { return fmt.Errorf("cannot change column type: %s: %s -> %s", strings.Join(path, "."), existing.Type, newField.Type) } @@ -1010,7 +1010,7 @@ func isIgnorableTypeUpdate(existingType, newType iceberg.Type) bool { return false } - return isPromotionAllowed(newPrimitive, existingPrim) + return iceberg.IsPromotionAllowed(newPrimitive, existingPrim) } switch existingType.(type) { @@ -1031,31 +1031,6 @@ func isIgnorableTypeUpdate(existingType, newType iceberg.Type) bool { 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