Repository: hamba/avro
Pull request #368 (2024-04-04 19:29:34)

Error on invalid type/logicalType
- Ran into a nasty issue where I had incorrectly used a `time.Duration` with a `long` Avro type and `timestamp-millis`. Hamba was encoding the entire int64 for this case instead of erroring on an invalid schema. When Spark/SQL tried to [convert](https://github.com/apache/spark/blob/v3.5.0/connector/avro/src/main/scala/org/apache/spark/sql/avro/AvroDeserializer.scala#L142C38-L142C52) this into a microseconds (it wants everything as micros), it eventually tries to do this:
```
def millisToMicros(millis: Long): Long = {
    Math.multiplyExact(millis, MICROS_PER_MILLIS)
}
```
And [Math.multiplyExact](https://docs.oracle.com/en/java/javase/22/docs/api/java.base/java/lang/Math.html#multiplyExact(long,int)) detects an overflow because it is trying to multiply the long by 1000, which doesn't fit. This fix will give the user an error if they use the `logicalType` incorrectly. The error will look something like `avro: time.Duration is unsupported for Avro long and logicalType timestamp-micros`.


Patch:
diff --git a/codec_native.go b/codec_native.go
index 62d4239..fd809cc 100644
--- a/codec_native.go
+++ b/codec_native.go
@@ -77,12 +77,16 @@ func createDecoderOfNative(schema *PrimitiveSchema, typ reflect2.Type) ValDecode
 				convert: createLongConverter(schema.encodedType),
 			}
 
-		case st == Long:
+		case st == Long && lt == "":
 			if resolved {
 				return &longConvCodec[int64]{convert: createLongConverter(schema.encodedType)}
 			}
 			return &longCodec[int64]{}
 
+		case lt != "":
+			return &errorDecoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s and logicalType %s",
+				typ.String(), schema.Type(), lt)}
+
 		default:
 			break
 		}
@@ -169,6 +173,7 @@ func createDecoderOfNative(schema *PrimitiveSchema, typ reflect2.Type) ValDecode
 	return &errorDecoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
 }
 
+//nolint:maintidx // Splitting this would not make it simpler.
 func createEncoderOfNative(schema Schema, typ reflect2.Type) ValEncoder {
 	switch typ.Kind() {
 	case reflect.Bool:
@@ -228,10 +233,17 @@ func createEncoderOfNative(schema Schema, typ reflect2.Type) ValEncoder {
 		switch {
 		case st == Int && lt == TimeMillis: // time.Duration
 			return &timeMillisCodec{}
+
 		case st == Long && lt == TimeMicros: // time.Duration
 			return &timeMicrosCodec{}
-		case st == Long:
+
+		case st == Long && lt == "":
 			return &longCodec[int64]{}
+
+		case lt != "":
+			return &errorEncoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s and logicalType %s",
+				typ.String(), schema.Type(), lt)}
+
 		default:
 			break
 		}


Test patch:
diff --git a/decoder_native_test.go b/decoder_native_test.go
index 0915c15..398bed1 100644
--- a/decoder_native_test.go
+++ b/decoder_native_test.go
@@ -641,6 +641,20 @@ func TestDecoder_Duration_TimeMicros(t *testing.T) {
 	assert.Equal(t, 123456789123*time.Microsecond, got)
 }
 
+func TestDecoder_Duration_InvalidLogicalType(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x86, 0xEA, 0xC8, 0xE9, 0x97, 0x07}
+	schema := `{"type":"long","logicalType":"timestamp-micros"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	require.NoError(t, err)
+
+	var got time.Duration
+	err = dec.Decode(&got)
+
+	assert.Error(t, err)
+}
+
 func TestDecoder_DurationInvalidSchema(t *testing.T) {
 	defer ConfigTeardown()
 
diff --git a/encoder_native_test.go b/encoder_native_test.go
index f201e06..739b1fe 100644
--- a/encoder_native_test.go
+++ b/encoder_native_test.go
@@ -626,6 +626,19 @@ func TestEncoder_Duration_TimeMicros(t *testing.T) {
 	assert.Equal(t, []byte{0x86, 0xEA, 0xC8, 0xE9, 0x97, 0x07}, buf.Bytes())
 }
 
+func TestEncoder_Duration_InvalidLogicalType(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"long","logicalType":"timestamp-micros"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	require.NoError(t, err)
+
+	err = enc.Encode(123456789123 * time.Microsecond)
+
+	assert.Error(t, err)
+}
+
 func TestEncoder_DurationInvalidSchema(t *testing.T) {
 	defer ConfigTeardown()