iceberg: fix schema evolution producing non-lowercase column names

Author: josephwoodward

internal/impl/iceberg/writer.go

@@ -359,6 +359,9 @@ func (s *parquetSink) EmitValue(sv shredder.ShreddedValue) error {
 }
 
 func (s *parquetSink) OnNewField(parentPath icebergx.Path, name string, value any) {
+	if !s.caseSensitive {
+		name = strings.ToLower(name)
+	}
 	fe := NewUnknownFieldError(parentPath, name, value)
 	key := dedupKey(fe.FullPath().String(), s.caseSensitive)
 	if _, ok := s.seenFields[key]; ok {
@@ -443,6 +446,9 @@ func (s *bufferingSink) EmitValue(sv shredder.ShreddedValue) error {
 }
 
 func (s *bufferingSink) OnNewField(parentPath icebergx.Path, name string, value any) {
+	if !s.caseSensitive {
+		name = strings.ToLower(name)
+	}
 	fe := NewUnknownFieldError(parentPath, name, value)
 	key := dedupKey(fe.FullPath().String(), s.caseSensitive)
 	if _, ok := s.seenFields[key]; ok {

internal/impl/iceberg/writer_test.go

@@ -55,6 +55,40 @@ func TestBufferingSinkNewFieldDedup(t *testing.T) {
 	})
 }
 
+// TestSchemaEvolutionColumnNameNormalization verifies that when
+// case_sensitive_columns=false, new column names are normalised to lowercase
+// before being stored in the UnknownFieldError that drives AddColumn. Without
+// this, a message keyed "EMAIL" would add an uppercase column to an otherwise
+// lowercase schema, violating iceberg's recommended convention.
+func TestSchemaEvolutionColumnNameNormalization(t *testing.T) {
+	t.Run("buffering sink normalises to lowercase", func(t *testing.T) {
+		sink := newBufferingSink(nil, 0, false)
+		sink.OnNewField(icebergx.Path{}, "EMAIL", "a@x.z")
+
+		errs := sink.newFieldErrors()
+		require.Len(t, errs, 1)
+		assert.Equal(t, "email", errs[0].FieldName())
+	})
+
+	t.Run("parquet sink normalises to lowercase", func(t *testing.T) {
+		sink := &parquetSink{caseSensitive: false}
+		sink.OnNewField(icebergx.Path{}, "EMAIL", "a@x.z")
+
+		errs := sink.newFieldErrors()
+		require.Len(t, errs, 1)
+		assert.Equal(t, "email", errs[0].FieldName())
+	})
+
+	t.Run("case-sensitive mode preserves original capitalisation", func(t *testing.T) {
+		sink := newBufferingSink(nil, 0, true)
+		sink.OnNewField(icebergx.Path{}, "EMAIL", "a@x.z")
+
+		errs := sink.newFieldErrors()
+		require.Len(t, errs, 1)
+		assert.Equal(t, "EMAIL", errs[0].FieldName())
+	})
+}
+
 // TestParquetSinkNewFieldDedup mirrors TestBufferingSinkNewFieldDedup for the
 // non-buffering parquet sink, which is used for unpartitioned writes.
 func TestParquetSinkNewFieldDedup(t *testing.T) {