AVRO-4267: [java] Compare bytes as unsigned

lang/java/avro/src/main/java/org/apache/avro/generic/GenericData.java

@@ -1318,11 +1318,29 @@ protected int compare(Object o1, Object o2, Schema s, boolean equals) {
       CharSequence cs1 = o1 instanceof CharSequence ? (CharSequence) o1 : o1.toString();
       CharSequence cs2 = o2 instanceof CharSequence ? (CharSequence) o2 : o2.toString();
       return Utf8.compareSequences(cs1, cs2);
+    case BYTES:
+      if (o1 instanceof ByteBuffer && o2 instanceof ByteBuffer) {
+        return compareByteBuffers((ByteBuffer) o1, (ByteBuffer) o2);
+      }
+      return ((Comparable) o1).compareTo(o2);
     default:
       return ((Comparable) o1).compareTo(o2);
     }
   }
 
+  private static int compareByteBuffers(ByteBuffer buffer1, ByteBuffer buffer2) {
+    int position1 = buffer1.position();
+    int position2 = buffer2.position();
+    int length1 = buffer1.remaining();
+    int length2 = buffer2.remaining();
+    for (int i = 0; i < Math.min(length1, length2); i++) {
+      int result = Byte.compareUnsigned(buffer1.get(position1 + i), buffer2.get(position2 + i));
+      if (result != 0)
+        return result;
+    }
+    return Integer.compare(length1, length2);
+  }
+
   private final ConcurrentMap<Field, Object> defaultValueCache = new ConcurrentReferenceHashMap<>(128, WEAK);
 
   /**

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

@@ -22,7 +22,9 @@
 import java.io.ByteArrayOutputStream;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import java.io.IOException;
+import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 
 import org.apache.avro.generic.GenericArray;
@@ -63,6 +65,33 @@ void bytes() throws Exception {
     check("\"bytes\"", ByteBuffer.wrap(new byte[] { 1, 2 }), ByteBuffer.wrap(new byte[] { 2 }));
   }
 
+  @Test
+  void bytesAreOrderedAsUnsignedValues() throws Exception {
+    checkBytesOrder(ByteBuffer.wrap(new byte[] { 0x7F }), ByteBuffer.wrap(new byte[] { (byte) 0xFF }));
+    checkBytesOrder(ByteBuffer.wrap(new byte[] { 1, 0x7F }), ByteBuffer.wrap(new byte[] { 1, (byte) 0x80 }));
+    checkBytesOrder(ByteBuffer.wrap(new byte[] { (byte) 0xFF }), ByteBuffer.wrap(new byte[] { (byte) 0xFF, 0 }));
+  }
+
+  @Test
+  void bytesOrderDependsOnlyOnTheRemainingRange() throws Exception {
+    byte[] backing1 = { (byte) 0xFF, 0x7F, (byte) 0xFF };
+    byte[] backing2 = { 0x00, (byte) 0xFF, 0x00 };
+    checkBytesOrder(ByteBuffer.wrap(backing1, 1, 1), ByteBuffer.wrap(backing2, 1, 1));
+    checkBytesOrder(ByteBuffer.wrap(backing1, 1, 1).slice(), ByteBuffer.wrap(backing2, 1, 1).slice());
+    checkBytesOrder(direct((byte) 0x7F), direct((byte) 0xFF));
+    checkBytesOrder(ByteBuffer.wrap(new byte[] { 0x7F }).asReadOnlyBuffer(),
+        ByteBuffer.wrap(new byte[] { (byte) 0xFF }).asReadOnlyBuffer());
+  }
+
+  @Test
+  void bytesLogicalTypesKeepTheirComparableOrder() {
+    Schema schema = LogicalTypes.decimal(9, 2).addToSchema(Schema.create(Schema.Type.BYTES));
+    GenericData comparator = GenericData.get();
+    assertTrue(comparator.compare(new BigDecimal("1.00"), new BigDecimal("2.00"), schema) < 0);
+    assertTrue(comparator.compare(new BigDecimal("2.00"), new BigDecimal("1.00"), schema) > 0);
+    assertEquals(0, comparator.compare(new BigDecimal("1.00"), new BigDecimal("1.00"), schema));
+  }
+
   @Test
   void testInt() throws Exception {
     check("\"int\"", -1, 0);
@@ -201,6 +230,33 @@ private static int compare(Object o1, Object o2, Schema schema, boolean comparab
     return comparable ? ((Comparable<Object>) o1).compareTo(o2) : comparator.compare(o1, o2, schema);
   }
 
+  private static void checkBytesOrder(ByteBuffer smaller, ByteBuffer larger) throws Exception {
+    Schema schema = SchemaParser.parseSingle("\"bytes\"");
+    byte[] b1 = render(smaller.duplicate(), schema, new GenericDatumWriter<>());
+    byte[] b2 = render(larger.duplicate(), schema, new GenericDatumWriter<>());
+    assertTrue(BinaryData.compare(b1, 0, b2, 0, schema) < 0);
+    assertTrue(BinaryData.compare(b2, 0, b1, 0, schema) > 0);
+
+    int smallerPosition = smaller.position();
+    int largerPosition = larger.position();
+    GenericData comparator = GenericData.get();
+
+    assertTrue(comparator.compare(smaller, larger, schema) < 0);
+    assertTrue(comparator.compare(larger, smaller, schema) > 0);
+    assertEquals(0, comparator.compare(smaller, smaller.duplicate(), schema));
+    assertEquals(0, comparator.compare(larger, larger.duplicate(), schema));
+
+    assertEquals(smallerPosition, smaller.position());
+    assertEquals(largerPosition, larger.position());
+  }
+
+  private static ByteBuffer direct(byte... bytes) {
+    ByteBuffer buffer = ByteBuffer.allocateDirect(bytes.length);
+    buffer.put(bytes);
+    buffer.flip();
+    return buffer;
+  }
+
   private static <T> byte[] render(T datum, Schema schema, DatumWriter<T> writer) throws IOException {
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     writer.setSchema(schema);