Improved conversions handling + pluggable conversions support [AVRO-1891, AVRO-2065]

lang/java/avro/src/main/java/org/apache/avro/data/RecordBuilderBase.java

@@ -17,19 +17,16 @@
  */
 package org.apache.avro.data;
 
-import java.io.IOException;
-import java.util.Arrays;
-
 import org.apache.avro.AvroRuntimeException;
-import org.apache.avro.Conversion;
-import org.apache.avro.Conversions;
-import org.apache.avro.LogicalType;
 import org.apache.avro.Schema;
 import org.apache.avro.Schema.Field;
 import org.apache.avro.Schema.Type;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.IndexedRecord;
 
+import java.io.IOException;
+import java.util.Arrays;
+
 /** Abstract base class for RecordBuilder implementations.  Not thread-safe. */
 public abstract class RecordBuilderBase<T extends IndexedRecord>
   implements RecordBuilder<T> {
@@ -138,29 +135,6 @@ protected Object defaultValue(Field field) throws IOException {
     return data.deepCopy(field.schema(), data.getDefaultValue(field));
   }
 
-  /**
-   * Gets the default value of the given field, if any. Pass in a conversion
-   * to convert data to logical type class. Please make sure the schema does
-   * have a logical type, otherwise an exception would be thrown out.
-   * @param field the field whose default value should be retrieved.
-   * @param conversion the tool to convert data to logical type class
-   * @return the default value associated with the given field,
-   * or null if none is specified in the schema.
-   * @throws IOException
-   */
-  @SuppressWarnings({ "rawtypes", "unchecked" })
-  protected Object defaultValue(Field field, Conversion<?> conversion) throws IOException {
-    Schema schema = field.schema();
-    LogicalType logicalType = schema.getLogicalType();
-    Object rawDefaultValue = data.deepCopy(schema, data.getDefaultValue(field));
-    if (conversion == null || logicalType == null) {
-      return rawDefaultValue;
-    } else {
-      return Conversions.convertToLogicalType(rawDefaultValue, schema,
-          logicalType, conversion);
-    }
-  }
-
   @Override
   public int hashCode() {
     final int prime = 31;

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

@@ -105,6 +105,10 @@ public GenericData(ClassLoader classLoader) {
   private Map<Class<?>, Map<String, Conversion<?>>> conversionsByClass =
       new IdentityHashMap<>();
 
+  public Collection<Conversion<?>> getConversions() {
+    return conversions.values();
+  }
+
   /**
    * Registers the given conversion to be used when reading and writing with
    * this data model.

lang/java/avro/src/main/java/org/apache/avro/specific/SpecificData.java

@@ -17,6 +17,7 @@
  */
 package org.apache.avro.specific;
 
+import java.lang.reflect.Field;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.Map;
@@ -132,6 +133,55 @@ public DatumWriter createDatumWriter(Schema schema) {
   /** Return the singleton instance. */
   public static SpecificData get() { return INSTANCE; }
 
+  /**
+   * For RECORD type schemas, this method returns the SpecificData instance of the class associated with the schema,
+   * in order to get the right conversions for any logical types used.
+   *
+   * @param reader the reader schema
+   * @return the SpecificData associated with the schema's class, or the default instance.
+   */
+  public static SpecificData getForSchema(Schema reader) {
+    if (reader.getType() == Type.RECORD) {
+      final String className = getClassName(reader);
+      if (className != null) {
+        final Class<?> clazz;
+        try {
+          clazz = Class.forName(className);
+          return getForClass(clazz);
+        } catch (ClassNotFoundException e) {
+          return SpecificData.get();
+        }
+      }
+    }
+    return SpecificData.get();
+  }
+
+  /**
+   * If the given class is assignable to {@link SpecificRecordBase}, this method returns the SpecificData instance
+   * from the field {@code MODEL$}, in order to get the correct {@link org.apache.avro.Conversion} instances for the class.
+   * Falls back to the default instance {@link SpecificData#get()} for other classes or if the field is not found.
+   *
+   * @param c A class
+   * @param <T> .
+   * @return The SpecificData from the SpecificRecordBase instance, or the default SpecificData instance.
+   */
+  public static <T> SpecificData getForClass(Class<T> c) {
+    if (SpecificRecordBase.class.isAssignableFrom(c)) {
+      final Field specificDataField;
+      try {
+        specificDataField = c.getDeclaredField("MODEL$");
+        specificDataField.setAccessible(true);
+        return (SpecificData) specificDataField.get(null);
+      } catch (NoSuchFieldException e) {
+        // Return default instance
+        return SpecificData.get();
+      } catch (IllegalAccessException e) {
+        throw new AvroRuntimeException(e);
+      }
+    }
+    return SpecificData.get();
+  }
+
   private boolean useCustomCoderFlag
     = Boolean.parseBoolean(System.getProperty("org.apache.avro.specific.use_custom_coders","false"));
 

lang/java/avro/src/main/java/org/apache/avro/specific/SpecificDatumReader.java

@@ -33,18 +33,18 @@ public SpecificDatumReader() {
 
   /** Construct for reading instances of a class. */
   public SpecificDatumReader(Class<T> c) {
-    this(new SpecificData(c.getClassLoader()));
+    this(SpecificData.getForClass(c));
     setSchema(getSpecificData().getSchema(c));
   }
 
   /** Construct where the writer's and reader's schemas are the same. */
   public SpecificDatumReader(Schema schema) {
-    this(schema, schema, SpecificData.get());
+    this(schema, schema, SpecificData.getForSchema(schema));
   }
 
   /** Construct given writer's and reader's schema. */
   public SpecificDatumReader(Schema writer, Schema reader) {
-    this(writer, reader, SpecificData.get());
+    this(writer, reader, SpecificData.getForSchema(reader));
   }
 
   /** Construct given writer's schema, reader's schema, and a {@link

lang/java/avro/src/main/java/org/apache/avro/specific/SpecificDatumWriter.java

@@ -32,11 +32,11 @@ public SpecificDatumWriter() {
   }
 
   public SpecificDatumWriter(Class<T> c) {
-    super(SpecificData.get().getSchema(c), SpecificData.get());
+    super(SpecificData.get().getSchema(c), SpecificData.getForClass(c));
   }
 
   public SpecificDatumWriter(Schema schema) {
-    super(schema, SpecificData.get());
+    super(schema, SpecificData.getForSchema(schema));
   }
 
   public SpecificDatumWriter(Schema root, SpecificData specificData) {

lang/java/avro/src/main/java/org/apache/avro/specific/SpecificRecordBase.java

@@ -37,6 +37,11 @@ public abstract class SpecificRecordBase
   public abstract Object get(int field);
   public abstract void put(int field, Object value);
 
+  public SpecificData getSpecificData() {
+    // Default implementation for backwards compatibility, overridden in generated code
+    return SpecificData.get();
+  }
+
   public Conversion<?> getConversion(int field) {
     // for backward-compatibility. no older specific classes have conversions.
     return null;
@@ -61,22 +66,22 @@ public boolean equals(Object that) {
     if (that == this) return true;                        // identical object
     if (!(that instanceof SpecificRecord)) return false;  // not a record
     if (this.getClass() != that.getClass()) return false; // not same schema
-    return SpecificData.get().compare(this, that, this.getSchema(), true) == 0;
+    return getSpecificData().compare(this, that, this.getSchema(), true) == 0;
   }
 
   @Override
   public int hashCode() {
-    return SpecificData.get().hashCode(this, this.getSchema());
+    return getSpecificData().hashCode(this, this.getSchema());
   }
 
   @Override
   public int compareTo(SpecificRecord that) {
-    return SpecificData.get().compare(this, that, this.getSchema());
+    return getSpecificData().compare(this, that, this.getSchema());
   }
 
   @Override
   public String toString() {
-    return SpecificData.get().toString(this);
+    return getSpecificData().toString(this);
   }
 
   @Override

lang/java/avro/src/main/java/org/apache/avro/specific/SpecificRecordBuilderBase.java

@@ -32,22 +32,22 @@ abstract public class SpecificRecordBuilderBase<T extends SpecificRecord>
    * @param schema the schema associated with the record class.
    */
   protected SpecificRecordBuilderBase(Schema schema) {
-    super(schema, SpecificData.get());
+    super(schema, SpecificData.getForSchema(schema));
   }
 
   /**
    * SpecificRecordBuilderBase copy constructor.
    * @param other SpecificRecordBuilderBase instance to copy.
    */
   protected SpecificRecordBuilderBase(SpecificRecordBuilderBase<T> other) {
-    super(other, SpecificData.get());
+    super(other, other.data());
   }
 
   /**
    * Creates a SpecificRecordBuilderBase by copying an existing record instance.
    * @param other the record instance to copy.
    */
   protected SpecificRecordBuilderBase(T other) {
-    super(other.getSchema(), SpecificData.get());
+    super(other.getSchema(), SpecificData.getForSchema(other.getSchema()));
   }
 }

lang/java/avro/src/test/java/org/apache/avro/specific/TestRecordWithJsr310LogicalTypes.java

@@ -854,16 +854,16 @@ public TestRecordWithJsr310LogicalTypes.Builder clearDec() {
     public TestRecordWithJsr310LogicalTypes build() {
       try {
         TestRecordWithJsr310LogicalTypes record = new TestRecordWithJsr310LogicalTypes();
-        record.b = fieldSetFlags()[0] ? this.b : (java.lang.Boolean) defaultValue(fields()[0], record.getConversion(0));
-        record.i32 = fieldSetFlags()[1] ? this.i32 : (java.lang.Integer) defaultValue(fields()[1], record.getConversion(1));
-        record.i64 = fieldSetFlags()[2] ? this.i64 : (java.lang.Long) defaultValue(fields()[2], record.getConversion(2));
-        record.f32 = fieldSetFlags()[3] ? this.f32 : (java.lang.Float) defaultValue(fields()[3], record.getConversion(3));
-        record.f64 = fieldSetFlags()[4] ? this.f64 : (java.lang.Double) defaultValue(fields()[4], record.getConversion(4));
-        record.s = fieldSetFlags()[5] ? this.s : (java.lang.CharSequence) defaultValue(fields()[5], record.getConversion(5));
-        record.d = fieldSetFlags()[6] ? this.d : (java.time.LocalDate) defaultValue(fields()[6], record.getConversion(6));
-        record.t = fieldSetFlags()[7] ? this.t : (java.time.LocalTime) defaultValue(fields()[7], record.getConversion(7));
-        record.ts = fieldSetFlags()[8] ? this.ts : (java.time.Instant) defaultValue(fields()[8], record.getConversion(8));
-        record.dec = fieldSetFlags()[9] ? this.dec : (java.math.BigDecimal) defaultValue(fields()[9], record.getConversion(9));
+        record.b = fieldSetFlags()[0] ? this.b : (java.lang.Boolean) defaultValue(fields()[0]);
+        record.i32 = fieldSetFlags()[1] ? this.i32 : (java.lang.Integer) defaultValue(fields()[1]);
+        record.i64 = fieldSetFlags()[2] ? this.i64 : (java.lang.Long) defaultValue(fields()[2]);
+        record.f32 = fieldSetFlags()[3] ? this.f32 : (java.lang.Float) defaultValue(fields()[3]);
+        record.f64 = fieldSetFlags()[4] ? this.f64 : (java.lang.Double) defaultValue(fields()[4]);
+        record.s = fieldSetFlags()[5] ? this.s : (java.lang.CharSequence) defaultValue(fields()[5]);
+        record.d = fieldSetFlags()[6] ? this.d : (java.time.LocalDate) defaultValue(fields()[6]);
+        record.t = fieldSetFlags()[7] ? this.t : (java.time.LocalTime) defaultValue(fields()[7]);
+        record.ts = fieldSetFlags()[8] ? this.ts : (java.time.Instant) defaultValue(fields()[8]);
+        record.dec = fieldSetFlags()[9] ? this.dec : (java.math.BigDecimal) defaultValue(fields()[9]);
         return record;
       } catch (java.lang.Exception e) {
         throw new org.apache.avro.AvroRuntimeException(e);

lang/java/compiler/src/main/java/org/apache/avro/compiler/specific/SpecificCompiler.java

@@ -293,6 +293,72 @@ public DateTimeLogicalTypeImplementation getDateTimeLogicalTypeImplementation()
     return dateTimeLogicalTypeImplementation;
   }
 
+  public void addCustomConversion(Class<?> conversionClass) {
+    try {
+      final Conversion<?> conversion = (Conversion<?>)conversionClass.newInstance();
+      specificData.addLogicalTypeConversion(conversion);
+    }  catch (IllegalAccessException | InstantiationException e) {
+      throw new RuntimeException("Failed to instantiate conversion class " + conversionClass, e);
+    }
+  }
+
+  public Collection<String> getUsedConversionClasses(Schema schema) {
+    LinkedHashMap<String, Conversion<?>> classnameToConversion = new LinkedHashMap<>();
+    for (Conversion<?> conversion : specificData.getConversions()) {
+      classnameToConversion.put(conversion.getConvertedType().getCanonicalName(), conversion);
+    }
+    Collection<String> result = new HashSet<>();
+    for (String className : getClassNamesOfPrimitiveFields(schema)) {
+      if (classnameToConversion.containsKey(className)) {
+        result.add(classnameToConversion.get(className).getClass().getCanonicalName());
+      }
+    }
+    return result;
+  }
+
+  private Set<String> getClassNamesOfPrimitiveFields(Schema schema) {
+    Set<String> result = new HashSet<>();
+    getClassNamesOfPrimitiveFields(schema, result, new HashSet<>());
+    return result;
+  }
+
+  private void getClassNamesOfPrimitiveFields(Schema schema, Set<String> result, Set<Schema> seenSchemas) {
+    if (seenSchemas.contains(schema)) {
+      return;
+    }
+    seenSchemas.add(schema);
+    switch (schema.getType()) {
+      case RECORD:
+        for (Schema.Field field : schema.getFields()) {
+          getClassNamesOfPrimitiveFields(field.schema(), result, seenSchemas);
+        }
+        break;
+      case MAP:
+        getClassNamesOfPrimitiveFields(schema.getValueType(), result, seenSchemas);
+        break;
+      case ARRAY:
+        getClassNamesOfPrimitiveFields(schema.getElementType(), result, seenSchemas);
+        break;
+      case UNION:
+        for (Schema s : schema.getTypes())
+          getClassNamesOfPrimitiveFields(s, result, seenSchemas);
+        break;
+      case ENUM:
+      case FIXED:
+      case NULL:
+        break;
+      case STRING: case BYTES:
+      case INT: case LONG:
+      case FLOAT: case DOUBLE:
+      case BOOLEAN:
+        result.add(javaType(schema));
+        break;
+      default: throw new RuntimeException("Unknown type: "+schema);
+    }
+  }
+
+  private static String logChuteName = null;
+
   private void initializeVelocity() {
     this.velocityEngine = new VelocityEngine();
 
@@ -810,14 +876,13 @@ public String conversionInstance(Schema schema) {
       return "null";
     }
 
-    if (LogicalTypes.date().equals(schema.getLogicalType())) {
-      return "DATE_CONVERSION";
-    } else if (LogicalTypes.timeMillis().equals(schema.getLogicalType())) {
-      return "TIME_CONVERSION";
-    } else if (LogicalTypes.timestampMillis().equals(schema.getLogicalType())) {
-      return "TIMESTAMP_CONVERSION";
-    } else if (LogicalTypes.Decimal.class.equals(schema.getLogicalType().getClass())) {
-      return enableDecimalLogicalType ? "DECIMAL_CONVERSION" : "null";
+    if (LogicalTypes.Decimal.class.equals(schema.getLogicalType().getClass()) && !enableDecimalLogicalType) {
+      return "null";
+    }
+
+    final Conversion<Object> conversion = specificData.getConversionFor(schema.getLogicalType());
+    if (conversion != null) {
+      return "new " + conversion.getClass().getCanonicalName() + "()";
     }
 
     return "null";