AVRO-3918: Add UUID with fixed[16]

doc/content/en/docs/++version++/Specification/_index.md

@@ -827,7 +827,25 @@ Here, as scale property is stored in value itself it needs more bytes than prece
 ### UUID
 The `uuid` logical type represents a random generated universally unique identifier (UUID).
 
-A `uuid` logical type annotates an Avro `string`. The string has to conform with [RFC-4122](https://www.ietf.org/rfc/rfc4122.txt)
+A `uuid` logical type annotates an Avro `string` or `fixed` of length 16. The string has to conform with [RFC-4122](https://www.ietf.org/rfc/rfc4122.txt)
+
+The following schemas represent a uuid:
+```json
+{
+  "type": "string",
+  "logicalType": "uuid"
+}
+```
+
+```json
+{
+  "type": "fixed",
+  "size": "16",
+  "logicalType": "uuid"
+}
+```
+
+(UUID will be sorted differently if the underlying type is a string or a fixed)
 
 ### Date
 The `date` logical type represents a date within the calendar, with no reference to a particular time zone or time of day.

lang/java/avro/src/main/java/org/apache/avro/Conversions.java

@@ -44,6 +44,7 @@
 public class Conversions {
 
   public static class UUIDConversion extends Conversion<UUID> {
+
     @Override
     public Class<UUID> getConvertedType() {
       return UUID.class;
@@ -68,6 +69,22 @@ public UUID fromCharSequence(CharSequence value, Schema schema, LogicalType type
     public CharSequence toCharSequence(UUID value, Schema schema, LogicalType type) {
       return value.toString();
     }
+
+    @Override
+    public UUID fromFixed(final GenericFixed value, final Schema schema, final LogicalType type) {
+      ByteBuffer buffer = ByteBuffer.wrap(value.bytes());
+      long mostSigBits = buffer.getLong();
+      long leastSigBits = buffer.getLong();
+      return new UUID(mostSigBits, leastSigBits);
+    }
+
+    @Override
+    public GenericFixed toFixed(final UUID value, final Schema schema, final LogicalType type) {
+      ByteBuffer buffer = ByteBuffer.allocate(2 * Long.BYTES);
+      buffer.putLong(value.getMostSignificantBits());
+      buffer.putLong(value.getLeastSignificantBits());
+      return new GenericData.Fixed(schema, buffer.array());
+    }
   }
 
   public static class DecimalConversion extends Conversion<BigDecimal> {

lang/java/avro/src/main/java/org/apache/avro/LogicalTypes.java

@@ -289,15 +289,21 @@ public static LocalTimestampMicros localTimestampNanos() {
 
   /** Uuid represents a uuid without a time */
   public static class Uuid extends LogicalType {
+
+    private static final int UUID_BYTES = 2 * Long.BYTES;
+
     private Uuid() {
       super(UUID);
     }
 
     @Override
     public void validate(Schema schema) {
       super.validate(schema);
-      if (schema.getType() != Schema.Type.STRING) {
-        throw new IllegalArgumentException("Uuid can only be used with an underlying string type");
+      if (schema.getType() != Schema.Type.STRING && schema.getType() != Schema.Type.FIXED) {
+        throw new IllegalArgumentException("Uuid can only be used with an underlying string or fixed type");
+      }
+      if (schema.getType() == Schema.Type.FIXED && schema.getFixedSize() != UUID_BYTES) {
+        throw new IllegalArgumentException("Uuid with fixed type must have a size of " + UUID_BYTES + " bytes");
       }
     }
   }

lang/java/avro/src/test/java/org/apache/avro/TestLogicalType.java

@@ -208,7 +208,7 @@ void uuidExtendsString() {
     assertEquals(LogicalTypes.uuid(), uuidSchema.getLogicalType());
 
     assertThrows("UUID requires a string", IllegalArgumentException.class,
-        "Uuid can only be used with an underlying string type",
+        "Uuid can only be used with an underlying string or fixed type",
         () -> LogicalTypes.uuid().addToSchema(Schema.create(Schema.Type.INT)));
   }
 

lang/java/avro/src/test/java/org/apache/avro/TestUuidConversions.java

@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.avro;
+
+import org.apache.avro.generic.GenericFixed;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.math.BigInteger;
+import java.util.UUID;
+import java.util.stream.Stream;
+
+public class TestUuidConversions {
+
+  private Conversions.UUIDConversion uuidConversion = new Conversions.UUIDConversion();
+
+  private Schema fixed = Schema.createFixed("fixed", "doc", "", Long.BYTES * 2);
+  private Schema fixedUuid = LogicalTypes.uuid().addToSchema(fixed);
+
+  private Schema string = Schema.createFixed("fixed", "doc", "", Long.BYTES * 2);
+  private Schema stringUuid = LogicalTypes.uuid().addToSchema(string);
+
+  @ParameterizedTest
+  @MethodSource("uuidData")
+  void uuidFixed(UUID uuid) {
+    GenericFixed value = uuidConversion.toFixed(uuid, fixedUuid, LogicalTypes.uuid());
+
+    byte[] b = new byte[Long.BYTES];
+    System.arraycopy(value.bytes(), 0, b, 0, b.length);
+    Assertions.assertEquals(uuid.getMostSignificantBits(), new BigInteger(b).longValue());
+    System.arraycopy(value.bytes(), Long.BYTES, b, 0, b.length);
+    Assertions.assertEquals(uuid.getLeastSignificantBits(), new BigInteger(b).longValue());
+
+    UUID uuid1 = uuidConversion.fromFixed(value, fixedUuid, LogicalTypes.uuid());
+    Assertions.assertEquals(uuid, uuid1);
+  }
+
+  @ParameterizedTest
+  @MethodSource("uuidData")
+  void uuidCharSequence(UUID uuid) {
+    CharSequence value = uuidConversion.toCharSequence(uuid, stringUuid, LogicalTypes.uuid());
+
+    Assertions.assertEquals(uuid.toString(), value.toString());
+
+    UUID uuid1 = uuidConversion.fromCharSequence(value, stringUuid, LogicalTypes.uuid());
+    Assertions.assertEquals(uuid, uuid1);
+  }
+
+  public static Stream<Arguments> uuidData() {
+    return Stream.of(Arguments.of(new UUID(Long.MIN_VALUE, Long.MAX_VALUE)), Arguments.of(new UUID(-1, 0)),
+        Arguments.of(UUID.randomUUID()), Arguments.of(UUID.randomUUID()));
+  }
+
+}