Remove Avro reflection

See `DISCUSS` on dev-list:

https://lists.apache.org/thread/c91s61tqkbbrc7xj180xh2rx89yx8pfk

Author: Fokko

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

@@ -18,16 +18,11 @@
  */
 package org.apache.parquet.avro;
 
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
 import java.nio.ByteBuffer;
-import java.util.Arrays;
-import java.util.List;
 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;
@@ -266,58 +261,6 @@ public Utf8 convert(Binary binary) {
     }
   }
 
-  static final class FieldStringableConverter extends BinaryConverter<Object> {
-    private final String stringableName;
-    private final Constructor<?> ctor;
-
-    public FieldStringableConverter(ParentValueContainer parent, Class<?> stringableClass) {
-      super(parent);
-      checkSecurity(stringableClass);
-      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 Object convert(Binary binary) {
-      try {
-        return ctor.newInstance(binary.toStringUsingUTF8());
-      } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
-        throw new ParquetDecodingException("Cannot convert binary to " + stringableName, e);
-      }
-    }
-
-    private void checkSecurity(Class<?> clazz) throws SecurityException {
-      List<String> trustedPackages = Arrays.asList(SERIALIZABLE_PACKAGES);
-
-      boolean trustAllPackages = trustedPackages.size() == 1 && "*".equals(trustedPackages.get(0));
-      if (trustAllPackages || clazz.isPrimitive()) {
-        return;
-      }
-
-      boolean found = false;
-      Package thePackage = clazz.getPackage();
-      if (thePackage != null) {
-        for (String trustedPackage : trustedPackages) {
-          if (thePackage.getName().equals(trustedPackage)
-              || thePackage.getName().startsWith(trustedPackage + ".")) {
-            found = true;
-            break;
-          }
-        }
-        if (!found) {
-          throw new SecurityException("Forbidden " + clazz
-              + "! This class is not trusted to be included in Avro schema using java-class."
-              + " Please set org.apache.parquet.avro.SERIALIZABLE_PACKAGES system property"
-              + " with the packages you trust.");
-        }
-      }
-    }
-  }
-
   static final class FieldEnumConverter extends BinaryConverter<Object> {
     private final Schema schema;
     private final GenericData model;

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

@@ -52,12 +52,10 @@
 import org.apache.avro.reflect.AvroIgnore;
 import org.apache.avro.reflect.AvroName;
 import org.apache.avro.reflect.ReflectData;
-import org.apache.avro.reflect.Stringable;
 import org.apache.avro.specific.SpecificData;
 import org.apache.avro.util.ClassUtils;
 import org.apache.parquet.Preconditions;
 import org.apache.parquet.avro.AvroConverters.FieldStringConverter;
-import org.apache.parquet.avro.AvroConverters.FieldStringableConverter;
 import org.apache.parquet.io.InvalidRecordException;
 import org.apache.parquet.io.api.Converter;
 import org.apache.parquet.io.api.GroupConverter;
@@ -144,18 +142,6 @@ public void add(Object value) {
       converters[parquetFieldIndex] =
           newConverter(nonNullSchema, parquetField, this.model, fieldClass, container);
 
-      // @Stringable doesn't affect the reflected schema; must be enforced here
-      if (recordClass != null && converters[parquetFieldIndex] instanceof FieldStringConverter) {
-        try {
-          Field field = recordClass.getDeclaredField(avroField.name());
-          if (field.isAnnotationPresent(Stringable.class)) {
-            converters[parquetFieldIndex] = new FieldStringableConverter(container, field.getType());
-          }
-        } catch (NoSuchFieldException e) {
-          // must not be stringable
-        }
-      }
-
       parquetFieldIndex += 1;
     }
 
@@ -411,7 +397,7 @@ private static Converter newStringConverter(Schema schema, GenericData model, Pa
     } else if (stringableClass == CharSequence.class) {
       return new AvroConverters.FieldUTF8Converter(parent);
     }
-    return new FieldStringableConverter(parent, stringableClass);
+    return null;
   }
 
   private static Class<?> getStringableClass(Schema schema, GenericData model) {

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

@@ -37,7 +37,6 @@
 import org.apache.avro.util.Utf8;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
-import org.apache.parquet.UntrustedStringableClass;
 import org.apache.parquet.hadoop.ParquetReader;
 import org.apache.parquet.hadoop.ParquetWriter;
 import org.apache.parquet.hadoop.metadata.CompressionCodecName;
@@ -77,40 +76,6 @@ public void testWriteReflectReadGeneric() throws IOException {
     }
   }
 
-  @Test(expected = SecurityException.class)
-  public void testUntrustedStringableClass() {
-    new AvroConverters.FieldStringableConverter(
-        new ParentValueContainer() {
-          @Override
-          public void add(Object value) {}
-
-          @Override
-          public void addBoolean(boolean value) {}
-
-          @Override
-          public void addInt(int value) {}
-
-          @Override
-          public void addLong(long value) {}
-
-          @Override
-          public void addFloat(float value) {}
-
-          @Override
-          public void addDouble(double value) {}
-
-          @Override
-          public void addChar(char value) {}
-
-          @Override
-          public void addByte(byte value) {}
-
-          @Override
-          public void addShort(short value) {}
-        },
-        UntrustedStringableClass.class);
-  }
-
   private GenericRecord getGenericPojoUtf8() {
     Schema schema = ReflectData.get().getSchema(Pojo.class);
     GenericData.Record record = new GenericData.Record(schema);