This is an automated email from the ASF dual-hosted git repository.
laskoviymishka 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 2987cc761 Schema compatibility functions (#2073)
2987cc761 is described below
commit 2987cc761b30a32331d4483c800af5d84ad4f37c
Author: Alex Stephen <[email protected]>
AuthorDate: Thu Oct 1 00:59:38 2026 -0700
Schema compatibility functions (#2073)
* Schema compatibility functions
* quick changes
* PR comments
---
schema_compatibility.go | 255 +++++++++++++++++++++
schema_compatibility_test.go | 531 +++++++++++++++++++++++++++++++++++++++++++
table/update_schema.go | 29 +--
3 files changed, 788 insertions(+), 27 deletions(-)
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<string, int> 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<int> 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
