From 1741d745e2582a165bacdaed32de50e4712c4392 Mon Sep 17 00:00:00 2001 From: Alex Stephen Date: Tue, 29 Sep 2026 00:51:13 +0000 Subject: [PATCH 1/3] Schema compatibility functions --- schema_compatibility.go | 268 +++++++++++++++++++ schema_compatibility_test.go | 486 +++++++++++++++++++++++++++++++++++ table/update_schema.go | 29 +-- 3 files changed, 756 insertions(+), 27 deletions(-) create mode 100644 schema_compatibility.go create mode 100644 schema_compatibility_test.go diff --git a/schema_compatibility.go b/schema_compatibility.go new file mode 100644 index 000000000..f0f9706b8 --- /dev/null +++ b/schema_compatibility.go @@ -0,0 +1,268 @@ +// 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 a column of type from may be changed to +// type to by schema evolution: int to long, float to double, or widening a +// decimal's precision while keeping its scale. Mirrors Java's +// TypeUtil.isPromotionAllowed. +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, or an empty list if there are none. +// Fields are matched by ID. A required read field must exist in writeSchema +// and be required there, and every type must equal or be a valid promotion +// of the written type. Field order is not checked. +// +// Passing a table's current schema as writeSchema and a proposed schema as +// readSchema checks that the proposed schema is a valid evolution. +// +// Mirrors Java's CheckCompatibility.readCompatibilityErrors. +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, or an empty list if +// there are none. It applies the same checks as ReadCompatibilityErrors and, +// when checkOrdering is true, also rejects fields that appear in a different +// order than in readSchema. +// +// Mirrors Java's CheckCompatibility.writeCompatibilityErrors. +func WriteCompatibilityErrors(readSchema, writeSchema *Schema, checkOrdering bool) ([]string, error) { + return checkCompatibility(readSchema, writeSchema, checkOrdering, true) +} + +// TypeCompatibilityErrors is WriteCompatibilityErrors without the +// nullability checks: writing optional values to a required field is not +// reported. +// +// Mirrors Java's CheckCompatibility.typeCompatibilityErrors. +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 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. Error messages starting with ":" belong to the enclosing +// field, which prefixes its name; others are nested and are joined with ".". +type compatibilityChecker struct { + schema *Schema + checkOrdering bool + checkNullability bool + + // current is the write-side type matching the read-side node being visited. + current Type + // inContainer is set by List and Map before visiting an element, key or + // value. PreOrderVisit routes those through Field, but only struct members + // are matched by ID, so Field passes them straight through. + 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 := -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, st.FieldList[lastOrdinal].Name)) + } + lastOrdinal = ordinal + } + } + + return errs +} + +func (c *compatibilityChecker) Field(readField NestedField, fieldErrors func() []string) []string { + if c.inContainer { + c.inContainer = false + + return fieldErrors() + } + + st := c.current.(*StructType) + 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.Type(), 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..430d4a510 --- /dev/null +++ b/schema_compatibility_test.go @@ -0,0 +1,486 @@ +// 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.Contains(t, errs[0], "struct cannot be read as a string") +} + +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.Contains(t, errs[0], "map cannot be read as a string") +} + +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.Contains(t, errs[0], "list cannot be read as a string") +} + +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.Contains(t, errs[0], "field_b is out of order, before field_a") +} + +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) +} + +// The remaining tests are not in Java's suite. They cover using +// ReadCompatibilityErrors to validate a schema evolution, and error paths. + +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)) + + _, err := iceberg.ReadCompatibilityErrors(nil, sc) + assert.ErrorIs(t, err, iceberg.ErrInvalidArgument) + + _, err = iceberg.ReadCompatibilityErrors(sc, nil) + assert.ErrorIs(t, err, iceberg.ErrInvalidArgument) +} 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 From 99fbd63b57413d6b4bd0b923b293697c25776d6f Mon Sep 17 00:00:00 2001 From: Alex Stephen Date: Tue, 29 Sep 2026 21:17:54 +0000 Subject: [PATCH 2/3] quick changes --- schema_compatibility.go | 41 +++++++++++------------------------- schema_compatibility_test.go | 3 --- 2 files changed, 12 insertions(+), 32 deletions(-) diff --git a/schema_compatibility.go b/schema_compatibility.go index f0f9706b8..63929f3fd 100644 --- a/schema_compatibility.go +++ b/schema_compatibility.go @@ -22,10 +22,9 @@ import ( "strings" ) -// IsPromotionAllowed reports whether a column of type from may be changed to -// type to by schema evolution: int to long, float to double, or widening a -// decimal's precision while keeping its scale. Mirrors Java's -// TypeUtil.isPromotionAllowed. +// 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 @@ -53,35 +52,21 @@ func IsPromotionAllowed(from, to PrimitiveType) bool { } // ReadCompatibilityErrors returns the problems with reading data written -// with writeSchema using readSchema, or an empty list if there are none. -// Fields are matched by ID. A required read field must exist in writeSchema -// and be required there, and every type must equal or be a valid promotion -// of the written type. Field order is not checked. -// -// Passing a table's current schema as writeSchema and a proposed schema as -// readSchema checks that the proposed schema is a valid evolution. -// -// Mirrors Java's CheckCompatibility.readCompatibilityErrors. +// 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, or an empty list if -// there are none. It applies the same checks as ReadCompatibilityErrors and, -// when checkOrdering is true, also rejects fields that appear in a different -// order than in readSchema. -// -// Mirrors Java's CheckCompatibility.writeCompatibilityErrors. +// 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 -// nullability checks: writing optional values to a required field is not -// reported. -// -// Mirrors Java's CheckCompatibility.typeCompatibilityErrors. +// nullability checks. func TypeCompatibilityErrors(readSchema, writeSchema *Schema, checkOrdering bool) ([]string, error) { return checkCompatibility(readSchema, writeSchema, checkOrdering, false) } @@ -99,18 +84,16 @@ func checkCompatibility(readSchema, writeSchema *Schema, checkOrdering, checkNul } // compatibilityChecker walks the read schema, tracking the matching type in -// the write schema. Error messages starting with ":" belong to the enclosing -// field, which prefixes its name; others are nested and are joined with ".". +// 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 is the write-side type matching the read-side node being visited. current Type - // inContainer is set by List and Map before visiting an element, key or - // value. PreOrderVisit routes those through Field, but only struct members - // are matched by ID, so Field passes them straight through. + // PreOrderVisit sends list elements and map keys/values through Field, + // but only struct fields should be looked up by ID. inContainer bool } diff --git a/schema_compatibility_test.go b/schema_compatibility_test.go index 430d4a510..37f37b795 100644 --- a/schema_compatibility_test.go +++ b/schema_compatibility_test.go @@ -396,9 +396,6 @@ func TestCompatibilityCheckNullabilityRequiredStructField(t *testing.T) { assert.Empty(t, errs) } -// The remaining tests are not in Java's suite. They cover using -// ReadCompatibilityErrors to validate a schema evolution, and error paths. - func TestReadCompatibilitySchemaEvolution(t *testing.T) { current := schemaOf( required(1, "id", iceberg.PrimitiveTypes.Int32), From dc925942b89f7fe976e36a0641fa8c6f26ba04b8 Mon Sep 17 00:00:00 2001 From: Alex Stephen Date: Wed, 30 Sep 2026 20:16:00 +0000 Subject: [PATCH 3/3] PR comments --- schema_compatibility.go | 22 +++++++----- schema_compatibility_test.go | 66 +++++++++++++++++++++++++++++++----- 2 files changed, 70 insertions(+), 18 deletions(-) diff --git a/schema_compatibility.go b/schema_compatibility.go index 63929f3fd..fe4468848 100644 --- a/schema_compatibility.go +++ b/schema_compatibility.go @@ -65,14 +65,15 @@ func WriteCompatibilityErrors(readSchema, writeSchema *Schema, checkOrdering boo return checkCompatibility(readSchema, writeSchema, checkOrdering, true) } -// TypeCompatibilityErrors is WriteCompatibilityErrors without the -// nullability checks. +// 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 writeSchema == nil { + if readSchema == nil || writeSchema == nil { return nil, fmt.Errorf("%w: cannot check compatibility against nil schema", ErrInvalidArgument) } @@ -122,17 +123,16 @@ func (c *compatibilityChecker) Struct(readStruct StructType, fieldErrors []func( ordinals[f.ID] = i } - lastOrdinal := -1 + 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, st.FieldList[lastOrdinal].Name)) + errs = append(errs, fmt.Sprintf("%s is out of order, before %s", readField.Name, lastName)) } - lastOrdinal = ordinal + lastOrdinal, lastName = ordinal, readField.Name } } @@ -146,7 +146,11 @@ func (c *compatibilityChecker) Field(readField NestedField, fieldErrors func() [ return fieldErrors() } - st := c.current.(*StructType) + st, ok := c.current.(*StructType) + if !ok { + return nil // Struct() already reported the mismatch + } + var ( writeField NestedField found bool @@ -231,7 +235,7 @@ func (c *compatibilityChecker) Primitive(readPrimitive PrimitiveType) []string { writePrimitive, ok := c.current.(PrimitiveType) if !ok { - return []string{fmt.Sprintf(": %s cannot be read as a %s", c.current.Type(), readPrimitive)} + return []string{fmt.Sprintf(": %s cannot be read as a %s", c.current, readPrimitive)} } if !IsPromotionAllowed(writePrimitive, readPrimitive) { diff --git a/schema_compatibility_test.go b/schema_compatibility_test.go index 37f37b795..6fbde7983 100644 --- a/schema_compatibility_test.go +++ b/schema_compatibility_test.go @@ -243,7 +243,7 @@ func TestCompatibilityIncompatibleStructAndPrimitive(t *testing.T) { errs := writeErrors(t, read, write) require.Len(t, errs, 1) - assert.Contains(t, errs[0], "struct cannot be read as a string") + assert.Equal(t, "nested: struct<1: from_field: required string> cannot be read as a string", errs[0]) } func TestCompatibilityMultipleErrors(t *testing.T) { @@ -311,7 +311,7 @@ func TestCompatibilityIncompatibleMapAndPrimitive(t *testing.T) { errs := writeErrors(t, read, write) require.Len(t, errs, 1) - assert.Contains(t, errs[0], "map cannot be read as a string") + assert.Equal(t, "map_field: map cannot be read as a string", errs[0]) } func TestCompatibilityRequiredListElement(t *testing.T) { @@ -338,7 +338,35 @@ func TestCompatibilityIncompatibleListAndPrimitive(t *testing.T) { errs := writeErrors(t, read, write) require.Len(t, errs, 1) - assert.Contains(t, errs[0], "list cannot be read as a string") + 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) { @@ -366,7 +394,18 @@ func TestCompatibilityStructWriteReordering(t *testing.T) { errs := writeErrors(t, read, write) require.Len(t, errs, 1) - assert.Contains(t, errs[0], "field_b is out of order, before field_a") + 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) { @@ -396,6 +435,15 @@ func TestCompatibilityCheckNullabilityRequiredStructField(t *testing.T) { 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), @@ -475,9 +523,9 @@ func TestCompatibilityNestedFieldErrorPath(t *testing.T) { func TestCompatibilityNilSchema(t *testing.T) { sc := schemaOf(required(1, "id", iceberg.PrimitiveTypes.Int32)) - _, err := iceberg.ReadCompatibilityErrors(nil, sc) - assert.ErrorIs(t, err, iceberg.ErrInvalidArgument) - - _, err = iceberg.ReadCompatibilityErrors(sc, nil) - assert.ErrorIs(t, err, iceberg.ErrInvalidArgument) + 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") + } }