chore: bump twmb/avro to v1.7.3 with field-level logicalType lift

Picks up two upstream landings from the rolling-fix work:

  1. twmb/avro PR #38 (Jeffail) — the field-level logicalType lift our
     own metadata parser already handles. Pulling it in means the
     value-side decoder now produces time.Time for sibling-form
     timestamp-millis (and the rest of the matrix) natively, instead
     of returning int64 and relying on the iceberg shredder's
     metadata-driven numeric scaling bridge to reconcile.

  2. twmb/avro PR #39 (twmb) — a cumulative perf, parity, and
     spec-compliance pass. Includes "decimal precision/scale, spec
     form" which changes how decimal-typed values serialise under
     EncodeJSON: a scale-2 value 0.33 (wire bytes 0x21) now emits as
     the codepoint-mapped string "!" rather than the numeric 0.33,
     matching Java's JsonEncoder output.

The shredder coerce bridge in iceberg/shredder/temporal.go stays —
it's now a safety net rather than load-bearing infrastructure. The
metadata-side fix in confluent/ecs_avro.go also stays because it
parses schemas into schema.Common independently of twmb (the iceberg
output's schema_metadata path uses Common, not twmb's schemaNode).

Coverage:
  - TestUpstreamTwmbHonoursSiblingFormLogicalType (new): pins the
    upstream PR #38 behaviour by asserting that sibling-form schemas
    decode to time.Time end-to-end. If twmb ever regresses on this,
    the test surfaces it in the package that depends on the contract.

  - TestSchemaRegistryDecodeAvro / TestSchemaRegistryDecodeAvroRawJson:
    pos_0_33333333 default-mode expectation updated from `0.33` to
    `"!"` per the spec form. Preserved-mode expectation unchanged —
    our preserveLogicalTypeOpts decimal CustomType still produces
    json.Number, which the SetStructuredMut path preserves through
    Go's json.Marshal.

CHANGELOG: a "Changed (potentially breaking)" entry documents the
decimal serialisation shape change for default-mode users and points
at preserve_logical_types: true as the migration knob.

Author: Jeffail

go.mod

@@ -170,7 +170,7 @@ require (
 	github.com/timeplus-io/proton-go-driver/v2 v2.1.4
 	github.com/tmc/langchaingo v0.1.14
 	github.com/trinodb/trino-go-client v0.333.0
-	github.com/twmb/avro v1.7.2
+	github.com/twmb/avro v1.7.3-0.20260513193503-1e5c2a3fc070
 	github.com/twmb/franz-go v1.20.7
 	github.com/twmb/franz-go/pkg/kadm v1.17.2
 	github.com/twmb/franz-go/pkg/kmsg v1.12.0

go.sum

@@ -1748,8 +1748,8 @@ github.com/trivago/grok v1.0.0/go.mod h1:9t59xLInhrncYq9a3J7488NgiBZi5y5yC7bss+w
 github.com/trivago/tgo v1.0.7 h1:uaWH/XIy9aWYWpjm2CU3RpcqZXmX2ysQ9/Go+d9gyrM=
 github.com/trivago/tgo v1.0.7/go.mod h1:w4dpD+3tzNIIiIfkWWa85w5/B77tlvdZckQ+6PkFnhc=
 github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqriFuLhtthL60Sar/7RFoluCcXsuvEwTV5KM=
-github.com/twmb/avro v1.7.2 h1:cmrEBRSbELRqsg/dRkQvVWuOaR2EfGifHIt/2iJ9lfI=
-github.com/twmb/avro v1.7.2/go.mod h1:X0fT1dY2xcbV4YuCE4mYro+qljHl4kUF5uA/2z1rgSk=
+github.com/twmb/avro v1.7.3-0.20260513193503-1e5c2a3fc070 h1:gYa95NoqeXYOMIVIe2/YcbC/l0s5q+wQ70Zo+TYQU4k=
+github.com/twmb/avro v1.7.3-0.20260513193503-1e5c2a3fc070/go.mod h1:X0fT1dY2xcbV4YuCE4mYro+qljHl4kUF5uA/2z1rgSk=
 github.com/twmb/franz-go v1.20.7 h1:P4MGSXJjjAPP3NRGPCks/Lrq+j+twWMVl1qYCVgNmWY=
 github.com/twmb/franz-go v1.20.7/go.mod h1:0bRX9HZVaoueqFWhPZNi2ODnJL7DNa6mK0HeCrC2bNU=
 github.com/twmb/franz-go/pkg/kadm v1.17.2 h1:g5f1sAxnTkYC6G96pV5u715HWhxd66hWaDZUAQ8xHY8=

internal/impl/confluent/avro_walker.go

@@ -28,6 +28,10 @@ import (
 // decoder path:
 //   - time-millis/time-micros: time.Duration → time.Time (time-of-day)
 //   - duration: avro.Duration → ISO 8601 duration string
+//   - decimal: raw []byte → json.Number for the SetStructuredMut path
+//     (Go's json.Marshal cannot natively format *big.Rat as a JSON
+//     number — it emits "33/100"-style fractional strings — so we
+//     pre-convert to json.Number for downstream bloblang/SQL use).
 func preserveLogicalTypeOpts() []avro.SchemaOpt {
 	return []avro.SchemaOpt{
 		avro.NewCustomType[time.Time, int32](
@@ -55,8 +59,6 @@ func preserveLogicalTypeOpts() []avro.SchemaOpt {
 				return avro.DurationFromBytes(b).String(), nil
 			},
 		},
-		// Decimal: raw type is []byte. Convert to json.Number for the
-		// SetStructuredMut path (json.Marshal can't handle *big.Rat).
 		avro.CustomType{
 			LogicalType: "decimal",
 			Decode: func(v any, node *avro.SchemaNode) (any, error) {

internal/impl/confluent/ecs_avro_field_level_logical_type_test.go

@@ -6,9 +6,11 @@ import (
 	"encoding/json"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/twmb/avro"
 
 	"github.com/redpanda-data/benthos/v4/public/schema"
 )
@@ -211,3 +213,66 @@ func mustJSON(v any) string {
 	b, _ := json.MarshalIndent(v, "", "  ")
 	return string(b)
 }
+
+// TestUpstreamTwmbHonoursSiblingFormLogicalType is a regression guard
+// for the upstream twmb/avro dependency. After the bump that landed our
+// own field-level logicalType fix upstream (twmb/avro PR #38), the
+// value-side decoder now produces time.Time for sibling-form schemas
+// natively — previously it returned int64 because the parser silently
+// dropped the field-level annotation.
+//
+// If twmb ever regresses on this, the shredder's metadata-driven
+// numeric scaling path in iceberg/shredder/temporal.go would become
+// load-bearing again. Pinning the upstream behaviour here makes any
+// such regression surface immediately, in the package that depends on
+// it, rather than as a customer report.
+func TestUpstreamTwmbHonoursSiblingFormLogicalType(t *testing.T) {
+	cases := []struct {
+		name   string
+		schema string
+	}{
+		{
+			"primitive timestamp-millis",
+			`{"type":"record","name":"R","fields":[
+				{"name":"ts","type":"long","logicalType":"timestamp-millis"}
+			]}`,
+		},
+		{
+			"union timestamp-millis (null first)",
+			`{"type":"record","name":"R","fields":[
+				{"name":"ts","type":["null","long"],"logicalType":"timestamp-millis"}
+			]}`,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			s, err := avro.Parse(tc.schema)
+			require.NoError(t, err)
+
+			// Encode a value via the binary-shape side schema to avoid
+			// requiring time.Time on the encode path; this is the same
+			// trick our integration test uses, and what a Java/JDBC
+			// producer would do — write raw long bytes on the wire and
+			// rely on the schema's logicalType for interpretation.
+			bin := avro.MustParse(`{"type":"record","name":"R","fields":[
+				{"name":"ts","type":["null","long"]}
+			]}`)
+			tsMillis := int64(1700000000000)
+			payload, err := bin.Encode(&struct {
+				TS *int64 `avro:"ts"`
+			}{TS: &tsMillis})
+			require.NoError(t, err)
+
+			var native any
+			_, err = s.Decode(payload, &native)
+			require.NoError(t, err)
+
+			row, ok := native.(map[string]any)
+			require.True(t, ok, "expected map, got %T", native)
+			_, isTime := row["ts"].(time.Time)
+			assert.True(t, isTime,
+				"upstream twmb must decode sibling-form timestamp-millis as time.Time; got %T (regression in twmb/avro PR #38)", row["ts"])
+		})
+	}
+}

internal/impl/confluent/processor_schema_registry_decode_test.go

@@ -290,10 +290,16 @@ func TestSchemaRegistryDecodeAvro(t *testing.T) {
 			output:   `{"Name":"foo","MaybeHobby":null,"Address": null}`,
 		},
 		{
-			schemaID:        4,
-			name:            "successful message with logical types",
-			input:           "\x00\x00\x00\x00\x04\x02\x90\xaf\xce!\x02\x80\x80揪\x97\t\x02\x80\x80\xde\xf2\xdf\xff\xdf\xdc\x01\x02\x02!",
-			output:          `{"int_time_millis":{"int.time-millis":35245000},"long_time_micros":{"long.time-micros":20192000000000},"long_timestamp_micros":{"long.timestamp-micros":62135596800000000},"pos_0_33333333":{"bytes.decimal":0.33}}`,
+			schemaID: 4,
+			name:     "successful message with logical types",
+			input:    "\x00\x00\x00\x00\x04\x02\x90\xaf\xce!\x02\x80\x80揪\x97\t\x02\x80\x80\xde\xf2\xdf\xff\xdf\xdc\x01\x02\x02!",
+			// pos_0_33333333 emits "!" — the spec-blessed codepoint-mapped
+			// string form for decimal under Avro JSON encoding (the single
+			// byte 0x21 = "!"). twmb/avro v1.7.3+ matches Java's
+			// JsonEncoder here; users who want a numeric form should set
+			// preserve_logical_types: true, which routes through our
+			// json.Number CustomType registration instead.
+			output:          `{"int_time_millis":{"int.time-millis":35245000},"long_time_micros":{"long.time-micros":20192000000000},"long_timestamp_micros":{"long.timestamp-micros":62135596800000000},"pos_0_33333333":{"bytes.decimal":"!"}}`,
 			preservedOutput: `{"int_time_millis":{"int.time-millis":"0001-01-01T09:47:25Z"},"long_time_micros":{"long.time-micros":"0001-08-22T16:53:20Z"},"long_timestamp_micros":{"long.timestamp-micros":"3939-01-01T00:00:00Z"},"pos_0_33333333":{"bytes.decimal":0.33}}`,
 		},
 		{
@@ -503,10 +509,12 @@ func TestSchemaRegistryDecodeAvroRawJson(t *testing.T) {
 			output:   `{"Name":"foo","MaybeHobby":null,"Address": null}`,
 		},
 		{
-			schemaID:        4,
-			name:            "successful message with logical types",
-			input:           "\x00\x00\x00\x00\x04\x02\x90\xaf\xce!\x02\x80\x80揪\x97\t\x02\x80\x80\xde\xf2\xdf\xff\xdf\xdc\x01\x02\x02!",
-			output:          `{"int_time_millis":35245000,"long_time_micros":20192000000000,"long_timestamp_micros":62135596800000000,"pos_0_33333333":0.33}`,
+			schemaID: 4,
+			name:     "successful message with logical types",
+			input:    "\x00\x00\x00\x00\x04\x02\x90\xaf\xce!\x02\x80\x80揪\x97\t\x02\x80\x80\xde\xf2\xdf\xff\xdf\xdc\x01\x02\x02!",
+			// pos_0_33333333 emits "!" in default mode — see the matching
+			// note on TestSchemaRegistryDecodeAvro for why.
+			output:          `{"int_time_millis":35245000,"long_time_micros":20192000000000,"long_timestamp_micros":62135596800000000,"pos_0_33333333":"!"}`,
 			preservedOutput: `{"int_time_millis":"0001-01-01T09:47:25Z","long_time_micros":"0001-08-22T16:53:20Z","long_timestamp_micros":"3939-01-01T00:00:00Z","pos_0_33333333":0.33}`,
 		},
 		{

internal/impl/confluent/serde_avro.go

@@ -201,7 +201,8 @@ func (s *schemaRegistryDecoder) getAvroDecoder(ctx context.Context, aschema fran
 
 	// Build parse options for preserve_logical_types: register custom
 	// types that convert time.Duration→time.Time for time-of-day fields,
-	// avro.Duration→string, and optionally Kafka Connect types.
+	// avro.Duration→string, decimal bytes→json.Number, and optionally
+	// Kafka Connect (Debezium) types.
 	var parseOpts []avro.SchemaOpt
 	if s.cfg.avro.preserveLogicalTypes {
 		parseOpts = append(parseOpts, preserveLogicalTypeOpts()...)