Keep avro on char sequence

Author: Piyush Narang

parquet-avro/src/main/java/org/apache/parquet/avro/AvroWriteSupport.java

@@ -364,7 +364,7 @@ private Binary fromAvroString(Object value) {
       Utf8 utf8 = (Utf8) value;
       return Binary.fromReusedByteArray(utf8.getBytes(), 0, utf8.getByteLength());
     }
-    return Binary.fromString(value.toString());
+    return Binary.fromCharSequence((CharSequence) value);
   }
 
   private static GenericData getDataModel(Configuration conf) {

parquet-avro/src/test/java/org/apache/parquet/avro/TestReflectLogicalTypes.java

@@ -426,7 +426,7 @@ public void testWriteNullableUUID() throws IOException {
         read(REFLECT, nullableUuidStringSchema, test));
   }
 
-//  @Test(expected = ClassCastException.class)
+  @Test(expected = ClassCastException.class)
   public void testWriteUUIDMissingLogicalType() throws IOException {
     Schema uuidSchema = SchemaBuilder.record(RecordWithUUID.class.getName())
         .fields().requiredString("uuid").endRecord();

parquet-column/src/main/java/org/apache/parquet/io/api/Binary.java

@@ -214,7 +214,7 @@ public void writeTo(DataOutput out) throws IOException {
 
   }
 
-  private static class FromStringBinary extends ByteArrayBackedBinary {
+  private static class FromStringBinary extends ByteBufferBackedBinary {
     public FromStringBinary(String value) {
       // reused is false, because we do not hold on to the buffer after
       // conversion, and nobody else has a handle to it
@@ -226,23 +226,44 @@ public String toString() {
       return "Binary{\"" + toStringUsingUTF8() + "\"}";
     }
 
-    private static final ThreadLocal<CharsetEncoder> ENCODER =
-        new ThreadLocal<CharsetEncoder>() {
-          @Override
-          protected CharsetEncoder initialValue() {
-            return StandardCharsets.UTF_8.newEncoder();
-          }
-        };
-
-    private static byte[] encodeUTF8(String value) {
+    private static ByteBuffer encodeUTF8(String value) {
       try {
-        return value.getBytes("UTF-8");
+        return ByteBuffer.wrap(value.getBytes("UTF-8"));
       } catch (UnsupportedEncodingException e) {
         throw new ParquetEncodingException("UTF-8 not supported.", e);
       }
     }
   }
 
+  private static class FromCharSequenceBinary extends ByteBufferBackedBinary {
+    public FromCharSequenceBinary(CharSequence value) {
+      // reused is false, because we do not hold on to the buffer after
+      // conversion, and nobody else has a handle to it
+      super(encodeUTF8(value), false);
+    }
+
+    @Override
+    public String toString() {
+      return "Binary{\"" + toStringUsingUTF8() + "\"}";
+    }
+
+    private static final ThreadLocal<CharsetEncoder> ENCODER =
+      new ThreadLocal<CharsetEncoder>() {
+        @Override
+        protected CharsetEncoder initialValue() {
+          return StandardCharsets.UTF_8.newEncoder();
+        }
+      };
+
+    private static ByteBuffer encodeUTF8(CharSequence value) {
+      try {
+        return ENCODER.get().encode(CharBuffer.wrap(value));
+      } catch (CharacterCodingException e) {
+        throw new ParquetEncodingException("UTF-8 not supported.", e);
+      }
+    }
+  }
+
   public static Binary fromReusedByteArray(final byte[] value, final int offset, final int length) {
     return new ByteArraySliceBackedBinary(value, offset, length, true);
   }
@@ -569,6 +590,10 @@ public static Binary fromString(String value) {
     return new FromStringBinary(value);
   }
 
+  public static Binary fromCharSequence(CharSequence value) {
+    return new FromCharSequenceBinary(value);
+  }
+
   /**
    * @see {@link Arrays#hashCode(byte[])}
    * @param array