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: twmb

CHANGELOG.md

@@ -3,6 +3,23 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+## Unreleased
+
+### Fixed
+
+- **BREAKING:** schema_registry_decode (avro): Avro logical types — `timestamp-{millis,micros,nanos}`, `local-timestamp-{millis,micros,nanos}`, `date`, `time-{millis,micros}`, and `uuid` — are now preserved end-to-end in the schema metadata produced by the schema-registry decoder. Previously only `decimal` was honoured; every other logical type silently degraded to its base primitive (`long`, `int`, or `string`). Downstream sinks that consume `schema_metadata` (notably `iceberg`) now create the correct column type. ([#4399](https://github.com/redpanda-data/connect/issues/4399))
+
+  Pipeline values flow through unchanged in both `preserve_logical_types=false` (default — values stay numeric) and `preserve_logical_types=true` (values stay rich Go time types). Bloblang behaviour and JSON-output bytes are unaffected.
+
+  **What this changes for existing pipelines:**
+  - **iceberg with existing tables that have BIGINT / INT / STRING columns from this bug**: the connector now wants to create or evolve those columns to TIMESTAMP / TIMESTAMPTZ / DATE / TIME / UUID. Iceberg disallows BIGINT → TIMESTAMP schema evolution, so the first write after upgrade will fail loudly. Drop and re-create the table, or use Iceberg-native column-rename + add-new-column tooling to migrate before upgrading.
+  - **Pipelines whose own code reads the `schema_metadata` bytes via `meta()`** and pattern-matches the historical INT64 shape: schemas now contain `TIMESTAMP` / `DATE` / `TIME_OF_DAY` / `UUID` along with new `unit` and `adjust_to_utc` fields. Update the pattern.
+  - **iceberg shredder** is now schema-aware for numeric inputs: a numeric millisecond value declared by the schema as `timestamp-millis` is correctly interpreted as milliseconds rather than as Unix seconds. This closes a previously-silent corruption case where an int64 millis input into a TIMESTAMPTZ column would land ~50,000 years in the future.
+
+### Changed
+
+- iceberg: `NewWriter` now takes a `*typeResolver` argument so the writer can use schema metadata to interpret numeric inputs into time-typed columns at shredding time. Internal API change only.
+
 ## 4.90.0 - 2026-04-30
 
 ### Added

go.mod

@@ -144,7 +144,7 @@ require (
 	github.com/rabbitmq/amqp091-go v1.10.0
 	github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9
 	github.com/redis/go-redis/v9 v9.18.0
-	github.com/redpanda-data/benthos/v4 v4.72.0
+	github.com/redpanda-data/benthos/v4 v4.72.1-0.20260506203543-b13320be0389
 	github.com/redpanda-data/common-go/authz v0.2.1-0.20260319205134-242ab3c168b8
 	github.com/redpanda-data/common-go/license v0.0.0-20260318014216-2bbd72bde0a0
 	github.com/redpanda-data/common-go/redpanda-otel-exporter v0.4.0

go.sum

@@ -1551,8 +1551,8 @@ github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 h1:bsUq1dX0N8A
 github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/redis/go-redis/v9 v9.18.0 h1:pMkxYPkEbMPwRdenAzUNyFNrDgHx9U+DrBabWNfSRQs=
 github.com/redis/go-redis/v9 v9.18.0/go.mod h1:k3ufPphLU5YXwNTUcCRXGxUoF1fqxnhFQmscfkCoDA0=
-github.com/redpanda-data/benthos/v4 v4.72.0 h1:Aj63fY6nZBGL17YDvHZV88caN2/jHlza9ZS57BGtxhs=
-github.com/redpanda-data/benthos/v4 v4.72.0/go.mod h1:if/3gnj/gIz3mKIiz2MGF7gNag/gv7ak0snVxP81BM4=
+github.com/redpanda-data/benthos/v4 v4.72.1-0.20260506203543-b13320be0389 h1:Zd/+nUlzUlEh1xpLuX1JHw1XDoqccnfTOKQfqCnXdLk=
+github.com/redpanda-data/benthos/v4 v4.72.1-0.20260506203543-b13320be0389/go.mod h1:if/3gnj/gIz3mKIiz2MGF7gNag/gv7ak0snVxP81BM4=
 github.com/redpanda-data/common-go/authz v0.2.1-0.20260319205134-242ab3c168b8 h1:hZTIp81OUDNOTCTD0gM01b1t821pDbToU9jWnZRnd/E=
 github.com/redpanda-data/common-go/authz v0.2.1-0.20260319205134-242ab3c168b8/go.mod h1:sHhzCYf64ZYUBi7snbopQl+wQaKySbFsKCvGhmSckhk=
 github.com/redpanda-data/common-go/license v0.0.0-20260318014216-2bbd72bde0a0 h1:xL2THs63tUTZmTiBfBm/mrjFMrwQaHKduvgQ6gIizXg=

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,149 @@ 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")
+}
+
+// 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"])
+		})
+	}
+}