PARQUET-286: Update String support to match upstream Avro.

This adds getStringableClass, which determines what String
representation upstream Avro would use. Specific and reflect will use an
alternative String class if java-class is set that is instantiated using
a constructor that takes a String. Otherwise, reflect will always use
String and both specific and generic will use Utf8 or String depending
on whether avro.java.string is set to "string".

The new string representations required two new converters: one for Utf8
and one for stringable classes (those with constructors that take a
single String). The converters have also been refactored so that all
binary converters now implement dictionary support.

Author: Ryan Blue <blue@apache.org>

Closes #201 from rdblue/PARQUET-286-avro-utf8-support and squashes the following commits:

beb5a44 [Ryan Blue] PARQUET-286: Add tests, support for stringable map keys.
0e9240f [Ryan Blue] PARQUET-286: Update string support to match upstream Avro.

Author: rdblue

parquet-avro/pom.xml

@@ -84,6 +84,14 @@
   </dependencies>
 
   <build>
+    <resources>
+      <resource>
+        <directory>src/test/avro</directory>
+      </resource>
+      <resource>
+        <directory>src/main/resources</directory>
+      </resource>
+    </resources>
     <plugins>
       <plugin>
         <artifactId>maven-enforcer-plugin</artifactId>
@@ -97,7 +105,15 @@
         <artifactId>avro-maven-plugin</artifactId>
         <version>${avro.version}</version>
         <executions>
+          <execution>
+            <id>compile-avsc</id>
+            <phase>generate-test-sources</phase>
+            <goals>
+              <goal>schema</goal>
+            </goals>
+          </execution>
             <execution>
+                <id>compile-idl</id>
                 <phase>generate-test-sources</phase>
                 <goals>
                     <goal>idl-protocol</goal>

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

@@ -18,10 +18,14 @@
  */
 package org.apache.parquet.avro;
 
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
 import java.nio.ByteBuffer;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
+import org.apache.avro.util.Utf8;
 import org.apache.parquet.column.Dictionary;
+import org.apache.parquet.io.ParquetDecodingException;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.io.api.GroupConverter;
 import org.apache.parquet.io.api.PrimitiveConverter;
@@ -44,6 +48,40 @@ public AvroPrimitiveConverter(ParentValueContainer parent) {
     }
   }
 
+  abstract static class BinaryConverter<T> extends AvroPrimitiveConverter {
+    private T[] dict = null;
+
+    public BinaryConverter(ParentValueContainer parent) {
+      super(parent);
+    }
+
+    public abstract T convert(Binary binary);
+
+    @Override
+    public void addBinary(Binary value) {
+      parent.add(convert(value));
+    }
+
+    @Override
+    public boolean hasDictionarySupport() {
+      return true;
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    public void setDictionary(Dictionary dictionary) {
+      dict = (T[]) new Object[dictionary.getMaxId() + 1];
+      for (int i = 0; i <= dictionary.getMaxId(); i++) {
+        dict[i] = convert(dictionary.decodeToBinary(i));
+      }
+    }
+
+    @Override
+    public void addValueFromDictionary(int dictionaryId) {
+      parent.add(dict[dictionaryId]);
+    }
+  }
+
   static final class FieldByteConverter extends AvroPrimitiveConverter {
     public FieldByteConverter(ParentValueContainer parent) {
       super(parent);
@@ -54,6 +92,7 @@ public void addInt(int value) {
       parent.addByte((byte) value);
     }
   }
+
   static final class FieldShortConverter extends AvroPrimitiveConverter {
     public FieldShortConverter(ParentValueContainer parent) {
       super(parent);
@@ -133,7 +172,6 @@ final public void addLong(long value) {
     final public void addFloat(float value) {
       parent.addFloat(value);
     }
-
   }
 
   static final class FieldDoubleConverter extends AvroPrimitiveConverter {
@@ -162,62 +200,84 @@ final public void addDouble(double value) {
     }
   }
 
-  static final class FieldByteArrayConverter extends AvroPrimitiveConverter {
+  static final class FieldByteArrayConverter extends BinaryConverter<byte[]> {
     public FieldByteArrayConverter(ParentValueContainer parent) {
       super(parent);
     }
 
     @Override
-    final public void addBinary(Binary value) {
-      parent.add(value.getBytes());
+    public byte[] convert(Binary binary) {
+      return binary.getBytes();
     }
   }
 
-  static final class FieldByteBufferConverter extends AvroPrimitiveConverter {
+  static final class FieldByteBufferConverter extends BinaryConverter<ByteBuffer> {
     public FieldByteBufferConverter(ParentValueContainer parent) {
       super(parent);
     }
 
     @Override
-    final public void addBinary(Binary value) {
-      parent.add(ByteBuffer.wrap(value.getBytes()));
+    public ByteBuffer convert(Binary binary) {
+      return ByteBuffer.wrap(binary.getBytes());
     }
   }
 
-  static final class FieldStringConverter extends AvroPrimitiveConverter {
-    // TODO: dictionary support should be generic and provided by a parent
-    // TODO: this always produces strings, but should respect avro.java.string
-    private String[] dict;
-
+  static final class FieldStringConverter extends BinaryConverter<String> {
     public FieldStringConverter(ParentValueContainer parent) {
       super(parent);
     }
 
     @Override
-    final public void addBinary(Binary value) {
-      parent.add(value.toStringUsingUTF8());
+    public String convert(Binary binary) {
+      return binary.toStringUsingUTF8();
     }
+  }
 
-    @Override
-    public boolean hasDictionarySupport() {
-      return true;
+  static final class FieldUTF8Converter extends BinaryConverter<Utf8> {
+    public FieldUTF8Converter(ParentValueContainer parent) {
+      super(parent);
     }
 
     @Override
-    public void setDictionary(Dictionary dictionary) {
-      dict = new String[dictionary.getMaxId() + 1];
-      for (int i = 0; i <= dictionary.getMaxId(); i++) {
-        dict[i] = dictionary.decodeToBinary(i).toStringUsingUTF8();
+    public Utf8 convert(Binary binary) {
+      return new Utf8(binary.getBytes());
+    }
+  }
+
+  static final class FieldStringableConverter extends BinaryConverter<Object> {
+    private final String stringableName;
+    private final Constructor<?> ctor;
+
+    public FieldStringableConverter(ParentValueContainer parent,
+                                    Class<?> stringableClass) {
+      super(parent);
+      stringableName = stringableClass.getName();
+      try {
+        this.ctor = stringableClass.getConstructor(String.class);
+      } catch (NoSuchMethodException e) {
+        throw new ParquetDecodingException(
+            "Unable to get String constructor for " + stringableName, e);
       }
     }
 
     @Override
-    public void addValueFromDictionary(int dictionaryId) {
-      parent.add(dict[dictionaryId]);
+    public Object convert(Binary binary) {
+      try {
+        return ctor.newInstance(binary.toStringUsingUTF8());
+      } catch (InstantiationException e) {
+        throw new ParquetDecodingException(
+            "Cannot convert binary to " + stringableName, e);
+      } catch (IllegalAccessException e) {
+        throw new ParquetDecodingException(
+            "Cannot convert binary to " + stringableName, e);
+      } catch (InvocationTargetException e) {
+        throw new ParquetDecodingException(
+            "Cannot convert binary to " + stringableName, e);
+      }
     }
   }
 
-  static final class FieldEnumConverter extends AvroPrimitiveConverter {
+  static final class FieldEnumConverter extends BinaryConverter<Object> {
     private final Schema schema;
     private final GenericData model;
 
@@ -229,12 +289,12 @@ public FieldEnumConverter(ParentValueContainer parent, Schema enumSchema,
     }
 
     @Override
-    final public void addBinary(Binary value) {
-      parent.add(model.createEnum(value.toStringUsingUTF8(), schema));
+    public Object convert(Binary binary) {
+      return model.createEnum(binary.toStringUsingUTF8(), schema);
     }
   }
 
-  static final class FieldFixedConverter extends AvroPrimitiveConverter {
+  static final class FieldFixedConverter extends BinaryConverter<Object> {
     private final Schema schema;
     private final GenericData model;
 
@@ -246,8 +306,8 @@ public FieldFixedConverter(ParentValueContainer parent, Schema avroSchema,
     }
 
     @Override
-    final public void addBinary(Binary value) {
-      parent.add(model.createFixed(null /* reuse */, value.getBytes(), schema));
+    public Object convert(Binary binary) {
+      return model.createFixed(null /* reuse */, binary.getBytes(), schema);
     }
   }
 }