confluent,iceberg: preserve avro logical types end-to-end (#4399)

The schema-registry Avro decoder now honours every Avro logical type the
spec defines (previously only `decimal`), and the iceberg connector maps
the resulting schema metadata to the right column type and interprets
numeric values using the schema's declared unit.

Decoder (internal/impl/confluent/ecs_avro.go):
  - Replace the single decimal-only branch with a dispatcher covering
    timestamp-{millis,micros,nanos}, local-timestamp-{millis,micros,nanos},
    date, time-{millis,micros}, and uuid. Per Avro 1.10 spec, unrecognised
    logicalType annotations and primitive/logical-type mismatches fall
    back silently to the base primitive.

Encoder (internal/impl/confluent/common_to_avro.go):
  - Symmetric encode for the new common types. Timestamp without explicit
    Logical params keeps emitting `timestamp-millis` via EffectiveTimestamp,
    preserving pre-PR output for legacy schemas. Date and TimeOfDay
    paths reject Avro-inexpressible shapes (e.g. nanos for time-of-day,
    AdjustToUTC for time-of-day) with field-naming errors rather than
    silently downcasting.

Iceberg type resolver (internal/impl/iceberg/type_resolver.go):
  - Map schema.Timestamp through EffectiveTimestamp so legacy schemas
    keep landing on TimestampTzType. Honour Logical.Timestamp.Unit and
    AdjustToUTC to pick TimestampType / TimestampTzType / TimestampNsType
    / TimestampTzNsType. Add Date, TimeOfDay, UUID arms; reject TimeOfDay
    shapes Iceberg can't faithfully represent (AdjustToUTC=true, nanos)
    with field-naming errors.

Iceberg shredder (internal/impl/iceberg/shredder/{shredder,temporal}.go):
  - Plumb a fieldID -> *schema.Common map onto RecordShredder via
    SetFieldSchemaMetadata. The leaf-value converter looks up the
    metadata for time-typed columns and uses the declared unit to scale
    numeric inputs into the column's internal representation. Without
    metadata, the converter accepts time.Time / time.Duration directly
    and falls back to bloblang.ValueAsTimestamp's seconds-default for
    bare numerics — preserving existing behaviour for callers that
    genuinely store unix-seconds.
  - This closes the silent-corruption case where a numeric millisecond
    value declared by the schema as timestamp-millis would land
    ~50,000 years in the future when the column type flipped from BIGINT
    to TIMESTAMPTZ. The schema's declared unit is now the source of
    truth for unit interpretation.

Iceberg writer (internal/impl/iceberg/writer.go, router.go):
  - NewWriter accepts the *typeResolver. The writer parses
    schema_metadata from the first message of a batch and builds the
    field-ID lookup the shredder consults. Internal API change only —
    the only call site is the router.

Breaking surface is documented in CHANGELOG.md under Unreleased.
Pipeline values flow through unchanged in both preserve_logical_types
modes; bloblang behaviour and JSON output bytes are unaffected.
The breakage is concentrated in (a) iceberg tables that already exist
with BIGINT/INT/STRING columns from this bug, which hit Iceberg's
schema-evolution wall, and (b) custom code that pattern-matches the
historical schema_metadata shape via meta() lookups.

Companion to redpanda-data/benthos#429 which adds the new schema.Common
types and parameter blocks. The go.mod replace directive is a
development crutch and must flip to a tagged release before merge.

Closes #4399

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: 2 people

internal/impl/confluent/common_to_avro.go

@@ -72,9 +72,48 @@ func commonToAvroInner(c schema.Common, recordName, namespace string, isRoot boo
 	case schema.Any:
 		return "bytes", nil
 	case schema.Timestamp:
+		// Honour Logical.Timestamp params if present; legacy nil-Logical
+		// schemas fall through EffectiveTimestamp() to {Millis, UTC},
+		// preserving the pre-PR encoder output exactly.
+		p := c.EffectiveTimestamp()
+		base := "long"
+		logicalName, err := avroTimestampLogicalName(p)
+		if err != nil {
+			return nil, fmt.Errorf("timestamp field %q: %w", c.Name, err)
+		}
+		return map[string]any{
+			"type":        base,
+			"logicalType": logicalName,
+		}, nil
+	case schema.Date:
 		return map[string]any{
-			"type":        "long",
-			"logicalType": "timestamp-millis",
+			"type":        "int",
+			"logicalType": "date",
+		}, nil
+	case schema.TimeOfDay:
+		if c.Logical == nil || c.Logical.TimeOfDay == nil {
+			return nil, fmt.Errorf("time-of-day field %q missing Logical.TimeOfDay parameters", c.Name)
+		}
+		// Avro time-{millis,micros} carry no zone semantics, so a
+		// TimeOfDay{AdjustToUTC=true} cannot be expressed faithfully.
+		// Reject loudly rather than silently dropping that bit.
+		if c.Logical.TimeOfDay.AdjustToUTC {
+			return nil, fmt.Errorf("time-of-day field %q has AdjustToUTC=true; Avro time-millis/time-micros have no UTC-adjust variant — coerce upstream before encoding", c.Name)
+		}
+		// Avro defines only time-millis (int) and time-micros (long); reject
+		// anything else loudly rather than silently downcasting.
+		switch c.Logical.TimeOfDay.Unit {
+		case schema.TimeUnitMillis:
+			return map[string]any{"type": "int", "logicalType": "time-millis"}, nil
+		case schema.TimeUnitMicros:
+			return map[string]any{"type": "long", "logicalType": "time-micros"}, nil
+		default:
+			return nil, fmt.Errorf("time-of-day field %q has unit %v which Avro cannot express; only MILLIS and MICROS are supported", c.Name, c.Logical.TimeOfDay.Unit)
+		}
+	case schema.UUID:
+		return map[string]any{
+			"type":        "string",
+			"logicalType": "uuid",
 		}, nil
 	case schema.Decimal:
 		if c.Logical == nil || c.Logical.Decimal == nil {
@@ -172,6 +211,28 @@ func commonToAvroUnion(c schema.Common) (any, error) {
 	return variants, nil
 }
 
+// avroTimestampLogicalName picks the Avro logical-type name for a
+// TimestampParams value. The mapping mirrors the reverse path in
+// applyAvroLogicalType: AdjustToUTC=true picks `timestamp-*`, false picks
+// `local-timestamp-*`. Avro 1.10+ supports nanos; older brokers may reject it.
+func avroTimestampLogicalName(p schema.TimestampParams) (string, error) {
+	var unit string
+	switch p.Unit {
+	case schema.TimeUnitMillis:
+		unit = "millis"
+	case schema.TimeUnitMicros:
+		unit = "micros"
+	case schema.TimeUnitNanos:
+		unit = "nanos"
+	default:
+		return "", fmt.Errorf("unsupported timestamp unit %v (Avro supports only MILLIS, MICROS, NANOS)", p.Unit)
+	}
+	if p.AdjustToUTC {
+		return "timestamp-" + unit, nil
+	}
+	return "local-timestamp-" + unit, nil
+}
+
 // sanitizeAvroName derives a valid Avro name from an arbitrary subject string.
 // Avro names must match [A-Za-z_][A-Za-z0-9_]*. Invalid characters are replaced
 // with underscores and a leading digit is prefixed with an underscore.

internal/impl/confluent/common_to_avro_test.go

@@ -219,3 +219,186 @@ func TestSanitizeAvroName(t *testing.T) {
 		})
 	}
 }
+
+// TestCommonToAvroLogicalTypeRoundTrip drives both halves of the Avro
+// adapter to confirm encode/decode are symmetric for every logical type
+// the connector now preserves. A schema.Common produced by a synthetic
+// decoder run, re-encoded to Avro JSON, must yield byte-equivalent output
+// to the original Avro spec for the same field.
+func TestCommonToAvroLogicalTypeRoundTrip(t *testing.T) {
+	cases := []struct {
+		name      string
+		input     schema.Common
+		wantOuter map[string]any
+	}{
+		{
+			name:      "Timestamp default (millis, UTC)",
+			input:     schema.Common{Type: schema.Timestamp},
+			wantOuter: map[string]any{"type": "long", "logicalType": "timestamp-millis"},
+		},
+		{
+			name: "Timestamp millis explicit",
+			input: schema.Common{
+				Type:    schema.Timestamp,
+				Logical: &schema.LogicalParams{Timestamp: &schema.TimestampParams{Unit: schema.TimeUnitMillis, AdjustToUTC: true}},
+			},
+			wantOuter: map[string]any{"type": "long", "logicalType": "timestamp-millis"},
+		},
+		{
+			name: "Timestamp micros UTC",
+			input: schema.Common{
+				Type:    schema.Timestamp,
+				Logical: &schema.LogicalParams{Timestamp: &schema.TimestampParams{Unit: schema.TimeUnitMicros, AdjustToUTC: true}},
+			},
+			wantOuter: map[string]any{"type": "long", "logicalType": "timestamp-micros"},
+		},
+		{
+			name: "Timestamp nanos UTC",
+			input: schema.Common{
+				Type:    schema.Timestamp,
+				Logical: &schema.LogicalParams{Timestamp: &schema.TimestampParams{Unit: schema.TimeUnitNanos, AdjustToUTC: true}},
+			},
+			wantOuter: map[string]any{"type": "long", "logicalType": "timestamp-nanos"},
+		},
+		{
+			name: "Timestamp local micros",
+			input: schema.Common{
+				Type:    schema.Timestamp,
+				Logical: &schema.LogicalParams{Timestamp: &schema.TimestampParams{Unit: schema.TimeUnitMicros, AdjustToUTC: false}},
+			},
+			wantOuter: map[string]any{"type": "long", "logicalType": "local-timestamp-micros"},
+		},
+		{
+			name:      "Date",
+			input:     schema.Common{Type: schema.Date},
+			wantOuter: map[string]any{"type": "int", "logicalType": "date"},
+		},
+		{
+			name: "TimeOfDay millis",
+			input: schema.Common{
+				Type:    schema.TimeOfDay,
+				Logical: &schema.LogicalParams{TimeOfDay: &schema.TimeOfDayParams{Unit: schema.TimeUnitMillis}},
+			},
+			wantOuter: map[string]any{"type": "int", "logicalType": "time-millis"},
+		},
+		{
+			name: "TimeOfDay micros",
+			input: schema.Common{
+				Type:    schema.TimeOfDay,
+				Logical: &schema.LogicalParams{TimeOfDay: &schema.TimeOfDayParams{Unit: schema.TimeUnitMicros}},
+			},
+			wantOuter: map[string]any{"type": "long", "logicalType": "time-micros"},
+		},
+		{
+			name:      "UUID",
+			input:     schema.Common{Type: schema.UUID},
+			wantOuter: map[string]any{"type": "string", "logicalType": "uuid"},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := avroUnmarshal(t, tc.input, "", "")
+			m, ok := got.(map[string]any)
+			require.True(t, ok, "expected object output, got %T", got)
+			assert.Equal(t, tc.wantOuter["type"], m["type"])
+			assert.Equal(t, tc.wantOuter["logicalType"], m["logicalType"])
+		})
+	}
+}
+
+// TestCommonToAvroTimeOfDayRejectsAdjustToUTC verifies that the Avro
+// encoder refuses to emit a TimeOfDay schema with AdjustToUTC=true.
+// Avro's time-millis / time-micros logical types carry no UTC-adjust bit,
+// so silently dropping it would lose the metadata. Fail loud instead.
+func TestCommonToAvroTimeOfDayRejectsAdjustToUTC(t *testing.T) {
+	c := schema.Common{
+		Name:    "shift_start",
+		Type:    schema.TimeOfDay,
+		Logical: &schema.LogicalParams{TimeOfDay: &schema.TimeOfDayParams{Unit: schema.TimeUnitMicros, AdjustToUTC: true}},
+	}
+	_, err := commonToAvroSchema(c, "Row", "")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "AdjustToUTC=true")
+	assert.Contains(t, err.Error(), "shift_start")
+}
+
+// TestCommonToAvroTimeOfDayRejectsUnsupportedUnit verifies that the
+// encoder refuses TimeOfDay schemas with units Avro doesn't define
+// (Seconds, Nanos), with a field-naming error mentioning the supported
+// units. Without this guard, an encoder downcast would silently change
+// resolution.
+func TestCommonToAvroTimeOfDayRejectsUnsupportedUnit(t *testing.T) {
+	for _, u := range []schema.TimeUnit{schema.TimeUnitSeconds, schema.TimeUnitNanos} {
+		t.Run(u.String(), func(t *testing.T) {
+			c := schema.Common{
+				Name:    "open_at",
+				Type:    schema.TimeOfDay,
+				Logical: &schema.LogicalParams{TimeOfDay: &schema.TimeOfDayParams{Unit: u}},
+			}
+			_, err := commonToAvroSchema(c, "Row", "")
+			require.Error(t, err)
+			assert.Contains(t, err.Error(), "MILLIS and MICROS")
+			assert.Contains(t, err.Error(), "open_at")
+		})
+	}
+}
+
+// TestCommonToAvroTimestampRejectsUnsupportedUnit verifies that the
+// avroTimestampLogicalName helper rejects timestamp units Avro doesn't
+// define (Seconds, and any other invalid TimeUnit). Avro 1.10+ supports
+// millis/micros/nanos.
+func TestCommonToAvroTimestampRejectsUnsupportedUnit(t *testing.T) {
+	c := schema.Common{
+		Name:    "event_time",
+		Type:    schema.Timestamp,
+		Logical: &schema.LogicalParams{Timestamp: &schema.TimestampParams{Unit: schema.TimeUnitSeconds, AdjustToUTC: true}},
+	}
+	_, err := commonToAvroSchema(c, "Row", "")
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "event_time")
+	assert.Contains(t, err.Error(), "MILLIS, MICROS, NANOS")
+}
+
+// TestCommonToAvroDecodeEncodeRoundTrip ensures that decoding an Avro spec
+// via ecsAvroParseFromBytes and re-encoding via commonToAvroSchema yields a
+// schema with the same logicalType annotation. This is the symmetry contract
+// future format adapters should maintain.
+func TestCommonToAvroDecodeEncodeRoundTrip(t *testing.T) {
+	cases := []struct {
+		name      string
+		fieldType string
+	}{
+		{"timestamp-millis", `{"type":"long","logicalType":"timestamp-millis"}`},
+		{"timestamp-micros", `{"type":"long","logicalType":"timestamp-micros"}`},
+		{"local-timestamp-millis", `{"type":"long","logicalType":"local-timestamp-millis"}`},
+		{"date", `{"type":"int","logicalType":"date"}`},
+		{"time-millis", `{"type":"int","logicalType":"time-millis"}`},
+		{"time-micros", `{"type":"long","logicalType":"time-micros"}`},
+		{"uuid", `{"type":"string","logicalType":"uuid"}`},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			spec := []byte(`{
+				"type":"record","name":"Row",
+				"fields":[{"name":"f","type":` + tc.fieldType + `}]
+			}`)
+			c, err := ecsAvroParseFromBytes(ecsAvroConfig{}, spec)
+			require.NoError(t, err)
+			require.Len(t, c.Children, 1)
+
+			// Re-encode the decoded child and verify the logicalType is
+			// the same as what we started with.
+			f := c.Children[0]
+			out, err := commonToAvroSchema(f, "Inner", "")
+			require.NoError(t, err)
+			var rt map[string]any
+			require.NoError(t, json.Unmarshal([]byte(out), &rt))
+
+			var want map[string]any
+			require.NoError(t, json.Unmarshal([]byte(tc.fieldType), &want))
+			assert.Equal(t, want["logicalType"], rt["logicalType"])
+		})
+	}
+}