fix(iceberg): apply schema_metadata for nil fields during table creation (#4383)

Author: josephwoodward

internal/impl/iceberg/router.go

@@ -416,6 +416,9 @@ func (r *Router) buildSchemaWithResolver(record map[string]any, msg *service.Mes
 	for _, metaName := range orderedMetaNames {
 		recordKey, ok := matchRecordKey(record, metaName, r.caseSensitive)
 		if !ok {
+			// Field declared in schema metadata but absent from this record.
+			// Include it so resolveTypeForCreateTable can apply the metadata type.
+			finalOrder = append(finalOrder, orderedField{emitName: metaName, metadataOnly: true})
 			continue
 		}
 		if _, seen := used[recordKey]; seen {
@@ -432,7 +435,10 @@ func (r *Router) buildSchemaWithResolver(record map[string]any, msg *service.Mes
 	}
 
 	for _, f := range finalOrder {
-		value := record[f.recordKey]
+		var value any
+		if !f.metadataOnly {
+			value = record[f.recordKey]
+		}
 		fieldType, err := r.resolver.resolveTypeForCreateTable(f.emitName, value, msg, common, key.namespace, key.table, ti)
 		if err != nil {
 			return nil, fmt.Errorf("resolving type for field %q: %w", f.emitName, err)
@@ -455,10 +461,13 @@ func (r *Router) buildSchemaWithResolver(record map[string]any, msg *service.Mes
 // name that should land on the iceberg column. They differ in case-insensitive
 // mode when a metadata field matches a record key with different casing — the
 // metadata casing wins for the column, the record casing is needed to find the
-// value.
+// value. When metadataOnly is true the field was declared in schema metadata but
+// is absent from the record; value is treated as nil and the column type comes
+// entirely from the metadata.
 type orderedField struct {
-	recordKey string
-	emitName  string
+	recordKey    string
+	emitName     string
+	metadataOnly bool
 }
 
 // matchRecordKey returns the actual key from record matching name, preserving

internal/impl/iceberg/router_test.go

@@ -325,3 +325,104 @@ func TestBuildSchemaWithResolverNewColumnTypeMappingSeesMetadataCasing(t *testin
 	assert.Equal(t, "long", fields[0].Type.Type(),
 		"mapping returns long only when name+path match metadata casing; any other casing means the rename path is broken")
 }
+
+// TestBuildSchemaWithResolverUsesMetadataTypeForNilValue verifies that
+// table-creation type resolution still applies schema metadata when the
+// observed record value is nil. Without this, nil values would be skipped
+// before metadata lookup and columns would appear later only after a non-nil
+// value is seen.
+func TestBuildSchemaWithResolverUsesMetadataTypeForNilValue(t *testing.T) {
+	router := &Router{
+		caseSensitive: true,
+		resolver:      newTypeResolver("schema_key", nil, true, nil),
+	}
+
+	schemaMeta := schema.Common{
+		Type: schema.Object,
+		Children: []schema.Common{
+			{Name: "count", Type: schema.Int32},
+		},
+	}
+
+	msg := service.NewMessage(nil)
+	msg.MetaSetMut("schema_key", schemaMeta.ToAny())
+
+	record := map[string]any{
+		"count": nil,
+	}
+
+	icebergSchema, err := router.buildSchemaWithResolver(record, msg, tableKey{namespace: "ns", table: "t"})
+	require.NoError(t, err)
+
+	fields := icebergSchema.Fields()
+	require.Len(t, fields, 1)
+	assert.Equal(t, "count", fields[0].Name)
+	assert.Equal(t, "int", fields[0].Type.Type())
+}
+
+// TestBuildSchemaWithResolverUsesMetadataTypeForAbsentField verifies that a
+// field declared in schema metadata but entirely absent from the record (not
+// even present as nil) is still created as an Iceberg column using the metadata
+// type.
+func TestBuildSchemaWithResolverUsesMetadataTypeForAbsentField(t *testing.T) {
+	router := &Router{
+		caseSensitive: true,
+		resolver:      newTypeResolver("schema_key", nil, true, nil),
+	}
+
+	schemaMeta := schema.Common{
+		Type: schema.Object,
+		Children: []schema.Common{
+			{Name: "count", Type: schema.Int32},
+		},
+	}
+
+	msg := service.NewMessage(nil)
+	msg.MetaSetMut("schema_key", schemaMeta.ToAny())
+
+	record := map[string]any{} // "count" is entirely absent
+
+	icebergSchema, err := router.buildSchemaWithResolver(record, msg, tableKey{namespace: "ns", table: "t"})
+	require.NoError(t, err)
+
+	fields := icebergSchema.Fields()
+	require.Len(t, fields, 1)
+	assert.Equal(t, "count", fields[0].Name)
+	assert.Equal(t, "int", fields[0].Type.Type())
+}
+
+// TestBuildSchemaWithResolverMetadataOnlyFieldOrdering verifies that
+// metadata-only fields (absent from the record) appear first in the schema in
+// metadata declaration order, followed by record-only fields in sorted order.
+func TestBuildSchemaWithResolverMetadataOnlyFieldOrdering(t *testing.T) {
+	router := &Router{
+		caseSensitive: true,
+		resolver:      newTypeResolver("schema_key", nil, true, nil),
+	}
+
+	schemaMeta := schema.Common{
+		Type: schema.Object,
+		Children: []schema.Common{
+			{Name: "b", Type: schema.Int32},
+			{Name: "a", Type: schema.String},
+		},
+	}
+
+	msg := service.NewMessage(nil)
+	msg.MetaSetMut("schema_key", schemaMeta.ToAny())
+
+	// "extra" is in the record but not in metadata; "b" and "a" are in metadata
+	// but absent from the record.
+	record := map[string]any{
+		"extra": "hello",
+	}
+
+	icebergSchema, err := router.buildSchemaWithResolver(record, msg, tableKey{namespace: "ns", table: "t"})
+	require.NoError(t, err)
+
+	fields := icebergSchema.Fields()
+	require.Len(t, fields, 3)
+	assert.Equal(t, "b", fields[0].Name) // metadata order first
+	assert.Equal(t, "a", fields[1].Name)
+	assert.Equal(t, "extra", fields[2].Name) // record-only field last
+}

internal/impl/iceberg/type_resolver.go

@@ -103,18 +103,19 @@ func (r *typeResolver) resolveTypeForCreateTable(
 	if err != nil {
 		return nil, err
 	}
-	if inferredType == nil {
-		return nil, nil // nil value, skip
-	}
 
 	// Stage 2: schema_metadata override (uses shared ti so override structs share the ID space)
 	path := icebergx.Path{{Kind: icebergx.PathField, Name: fieldName}}
 	if metaType, err := r.resolveFromCommon(common, path, ti); err != nil {
 		return nil, fmt.Errorf("resolving type from schema metadata for field %q: %w", fieldName, err)
 	} else if metaType != nil {
-		// If the metatype was not found then just stick with the default inferred type
+		// Metadata takes precedence over runtime inference. This also ensures
+		// metadata-declared columns are created even when the observed value is nil.
 		inferredType = metaType
 	}
+	if inferredType == nil {
+		return nil, nil // nil value with no metadata type, skip
+	}
 
 	// Stage 3: Bloblang mapping override (only for primitive/leaf types)
 	if r.newColumnTypeMapping != nil && isPrimitiveType(inferredType) {