[HUDI-6872] Simplify Out Of Box Schema Evolution Functionality

hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/AvroConversionUtils.scala

@@ -182,27 +182,24 @@ object AvroConversionUtils {
           } else {
             field.doc()
           }
-          val newSchema = getAvroSchemaWithDefaults(field.schema(), structFields(i).dataType)
-          field.schema().getType match {
-            case Schema.Type.UNION => {
-              val innerFields = newSchema.getTypes
-              val containsNullSchema = innerFields.foldLeft(false)((nullFieldEncountered, schema) => nullFieldEncountered | schema.getType == Schema.Type.NULL)
-              if(containsNullSchema) {
-                // Need to re shuffle the fields in list because to set null as default, null schema must be head in union schema
-                val restructuredNewSchema = Schema.createUnion(List(Schema.create(Schema.Type.NULL)) ++ innerFields.filter(innerSchema => !(innerSchema.getType == Schema.Type.NULL)))
-                new Schema.Field(field.name(), restructuredNewSchema, comment, JsonProperties.NULL_VALUE)
-              } else {
-                new Schema.Field(field.name(), newSchema, comment, field.defaultVal())
-              }
-            }
-            case _ => new Schema.Field(field.name(), newSchema, comment, field.defaultVal())
+          //need special handling for union because we update field default to null if it's in the union
+          val (newSchema, containsNullSchema) = field.schema().getType match {
+            case Schema.Type.UNION => resolveUnion(field.schema(), structFields(i).dataType)
+            case _ => (getAvroSchemaWithDefaults(field.schema(), structFields(i).dataType), false)
           }
+          new Schema.Field(field.name(), newSchema, comment,
+            if (containsNullSchema) {
+              JsonProperties.NULL_VALUE
+            } else {
+              field.defaultVal()
+            })
         }).toList
         Schema.createRecord(schema.getName, schema.getDoc, schema.getNamespace, schema.isError, modifiedFields)
       }
 
       case Schema.Type.UNION => {
-        Schema.createUnion(schema.getTypes.map(innerSchema => getAvroSchemaWithDefaults(innerSchema, dataType)))
+       val (resolved, _) = resolveUnion(schema, dataType)
+        resolved
       }
 
       case Schema.Type.MAP => {
@@ -217,6 +214,25 @@ object AvroConversionUtils {
     }
   }
 
+  /**
+   * Helper method for getAvroSchemaWithDefaults for schema type union
+   * re-arrange so that null is first if it is in the union
+   *
+   * @param schema input avro schema
+   * @return Avro schema with null default set to nullable fields and bool that is true if the union contains null
+   *
+   * */
+  private def resolveUnion(schema: Schema, dataType: DataType): (Schema, Boolean) = {
+    val innerFields = schema.getTypes
+    val containsNullSchema = innerFields.foldLeft(false)((nullFieldEncountered, schema) => nullFieldEncountered | schema.getType == Schema.Type.NULL)
+    (if (containsNullSchema) {
+      Schema.createUnion(List(Schema.create(Schema.Type.NULL)) ++ innerFields.filter(innerSchema => !(innerSchema.getType == Schema.Type.NULL))
+        .map(innerSchema => getAvroSchemaWithDefaults(innerSchema, dataType)))
+    } else {
+      Schema.createUnion(schema.getTypes.map(innerSchema => getAvroSchemaWithDefaults(innerSchema, dataType)))
+    }, containsNullSchema)
+  }
+
   /**
    * Please use [[AvroSchemaUtils.getAvroRecordQualifiedName(String)]]
    */

hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaCompatibility.java

@@ -43,6 +43,7 @@
 import java.util.TreeSet;
 import java.util.stream.Collectors;
 
+import static org.apache.hudi.avro.HoodieAvroUtils.isTypeNumeric;
 import static org.apache.hudi.common.util.ValidationUtils.checkState;
 
 /**
@@ -62,10 +63,15 @@
 public class AvroSchemaCompatibility {
   private static final Logger LOG = LoggerFactory.getLogger(AvroSchemaCompatibility.class);
 
-  /** Utility class cannot be instantiated. */
-  private AvroSchemaCompatibility() {}
+  /**
+   * Utility class cannot be instantiated.
+   */
+  private AvroSchemaCompatibility() {
+  }
 
-  /** Message to annotate reader/writer schema pairs that are compatible. */
+  /**
+   * Message to annotate reader/writer schema pairs that are compatible.
+   */
   public static final String READER_WRITER_COMPATIBLE_MESSAGE = "Reader schema can always successfully decode data written using the writer schema.";
 
   /**
@@ -161,7 +167,7 @@ public static Field lookupWriterField(final Schema writerSchema, final Field rea
 
   /**
    * Reader/writer schema pair that can be used as a key in a hash map.
-   *
+   * <p>
    * This reader/writer pair differentiates Schema objects based on their system
    * hash code.
    */
@@ -180,13 +186,17 @@ public ReaderWriter(final Schema reader, final Schema writer) {
       mWriter = writer;
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public int hashCode() {
       return System.identityHashCode(mReader) ^ System.identityHashCode(mWriter);
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public boolean equals(Object obj) {
       if (!(obj instanceof ReaderWriter)) {
@@ -197,7 +207,9 @@ public boolean equals(Object obj) {
       return (this.mReader == that.mReader) && (this.mWriter == that.mWriter);
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public String toString() {
       return String.format("ReaderWriter{reader:%s, writer:%s}", mReader, mWriter);
@@ -279,8 +291,8 @@ private SchemaCompatibilityResult getCompatibility(final Schema reader,
      * {@link #getCompatibility(Schema, Schema)}.
      * </p>
      *
-     * @param reader   Reader schema to test.
-     * @param writer   Writer schema to test.
+     * @param reader    Reader schema to test.
+     * @param writer    Writer schema to test.
      * @param locations Stack with which to track the location within the schema.
      * @return the compatibility of the reader/writer schema pair.
      */
@@ -372,7 +384,8 @@ private SchemaCompatibilityResult calculateCompatibility(final Schema reader, fi
             return (writer.getType() == Type.STRING) ? result : result.mergedWith(typeMismatch(reader, writer, locations));
           }
           case STRING: {
-            return (writer.getType() == Type.BYTES) ? result : result.mergedWith(typeMismatch(reader, writer, locations));
+            return (isTypeNumeric(writer.getType()) || (writer.getType() == Schema.Type.BYTES)
+                ? result : result.mergedWith(typeMismatch(reader, writer, locations)));
           }
 
           case ARRAY:
@@ -540,7 +553,9 @@ private static List<String> asList(Deque<LocationInfo> deque) {
   public enum SchemaCompatibilityType {
     COMPATIBLE, INCOMPATIBLE,
 
-    /** Used internally to tag a reader/writer schema pair and prevent recursion. */
+    /**
+     * Used internally to tag a reader/writer schema pair and prevent recursion.
+     */
     RECURSION_IN_PROGRESS;
   }
 
@@ -565,7 +580,7 @@ public static final class SchemaCompatibilityResult {
      * @param toMerge The {@code SchemaCompatibilityResult} to merge with the
      *                current instance.
      * @return A {@code SchemaCompatibilityResult} that combines the state of the
-     *         current and supplied instances.
+     * current and supplied instances.
      */
     public SchemaCompatibilityResult mergedWith(SchemaCompatibilityResult toMerge) {
       List<Incompatibility> mergedIncompatibilities = new ArrayList<>(mIncompatibilities);
@@ -595,7 +610,7 @@ private SchemaCompatibilityResult(SchemaCompatibilityType compatibilityType,
      * Returns a details object representing a compatible schema pair.
      *
      * @return a SchemaCompatibilityDetails object with COMPATIBLE
-     *         SchemaCompatibilityType, and no other state.
+     * SchemaCompatibilityType, and no other state.
      */
     public static SchemaCompatibilityResult compatible() {
       return COMPATIBLE;
@@ -606,7 +621,7 @@ public static SchemaCompatibilityResult compatible() {
      * progress.
      *
      * @return a SchemaCompatibilityDetails object with RECURSION_IN_PROGRESS
-     *         SchemaCompatibilityType, and no other state.
+     * SchemaCompatibilityType, and no other state.
      */
     public static SchemaCompatibilityResult recursionInProgress() {
       return RECURSION_IN_PROGRESS;
@@ -617,7 +632,7 @@ public static SchemaCompatibilityResult recursionInProgress() {
      * error details.
      *
      * @return a SchemaCompatibilityDetails object with INCOMPATIBLE
-     *         SchemaCompatibilityType, and state representing the violating part.
+     * SchemaCompatibilityType, and state representing the violating part.
      */
     public static SchemaCompatibilityResult incompatible(SchemaIncompatibilityType incompatibilityType,
                                                          Schema readerFragment, Schema writerFragment, String message, List<String> location) {
@@ -641,13 +656,15 @@ public SchemaCompatibilityType getCompatibility() {
      * Incompatibilities} found, otherwise an empty list.
      *
      * @return a list of {@link Incompatibility Incompatibilities}, may be empty,
-     *         never null.
+     * never null.
      */
     public List<Incompatibility> getIncompatibilities() {
       return mIncompatibilities;
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public int hashCode() {
       final int prime = 31;
@@ -657,7 +674,9 @@ public int hashCode() {
       return result;
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public boolean equals(Object obj) {
       if (this == obj) {
@@ -680,7 +699,9 @@ public boolean equals(Object obj) {
       return mCompatibilityType == other.mCompatibilityType;
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public String toString() {
       return String.format("SchemaCompatibilityResult{compatibility:%s, incompatibilities:%s}", mCompatibilityType,
@@ -737,8 +758,8 @@ public Schema getWriterFragment() {
      * Returns a human-readable message with more details about what failed. Syntax
      * depends on the SchemaIncompatibilityType.
      *
-     * @see #getType()
      * @return a String with details about the incompatibility.
+     * @see #getType()
      */
     public String getMessage() {
       return mMessage;
@@ -768,7 +789,9 @@ public String getLocation() {
       return s.toString();
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public int hashCode() {
       final int prime = 31;
@@ -781,7 +804,9 @@ public int hashCode() {
       return result;
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public boolean equals(Object obj) {
       if (this == obj) {
@@ -825,7 +850,9 @@ public boolean equals(Object obj) {
       }
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public String toString() {
       return String.format("Incompatibility{type:%s, location:%s, message:%s, reader:%s, writer:%s}", mType,
@@ -837,21 +864,29 @@ public String toString() {
   /**
    * Provides information about the compatibility of a single reader and writer
    * schema pair.
-   *
+   * <p>
    * Note: This class represents a one-way relationship from the reader to the
    * writer schema.
    */
   public static final class SchemaPairCompatibility {
-    /** The details of this result. */
+    /**
+     * The details of this result.
+     */
     private final SchemaCompatibilityResult mResult;
 
-    /** Validated reader schema. */
+    /**
+     * Validated reader schema.
+     */
     private final Schema mReader;
 
-    /** Validated writer schema. */
+    /**
+     * Validated writer schema.
+     */
     private final Schema mWriter;
 
-    /** Human readable description of this result. */
+    /**
+     * Human readable description of this result.
+     */
     private final String mDescription;
 
     /**
@@ -915,14 +950,18 @@ public String getDescription() {
       return mDescription;
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public String toString() {
       return String.format("SchemaPairCompatibility{result:%s, readerSchema:%s, writerSchema:%s, description:%s}",
           mResult, mReader, mWriter, mDescription);
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public boolean equals(Object other) {
       if ((other instanceof SchemaPairCompatibility)) {
@@ -934,14 +973,18 @@ public boolean equals(Object other) {
       }
     }
 
-    /** {@inheritDoc} */
+    /**
+     * {@inheritDoc}
+     */
     @Override
     public int hashCode() {
-      return Arrays.hashCode(new Object[] { mResult, mReader, mWriter, mDescription });
+      return Arrays.hashCode(new Object[] {mResult, mReader, mWriter, mDescription});
     }
   }
 
-  /** Borrowed from Guava's Objects.equal(a, b) */
+  /**
+   * Borrowed from Guava's Objects.equal(a, b)
+   */
   private static boolean objectsEqual(Object obj1, Object obj2) {
     return Objects.equals(obj1, obj2);
   }

hudi-common/src/main/java/org/apache/hudi/avro/AvroSchemaUtils.java

@@ -116,9 +116,24 @@ public static String getAvroRecordQualifiedName(String tableName) {
     return "hoodie." + sanitizedTableName + "." + sanitizedTableName + "_record";
   }
 
+  /**
+   * Validate whether the {@code targetSchema} is a valid evolution of {@code sourceSchema}.
+   * Basically {@link #isCompatibleProjectionOf(Schema, Schema)} but type promotion in the
+   * opposite direction
+   */
+  public static boolean isValidEvolutionOf(Schema sourceSchema, Schema targetSchema) {
+    return (sourceSchema.getType() == Schema.Type.NULL) || isProjectionOfInternal(sourceSchema, targetSchema,
+        AvroSchemaUtils::isAtomicSchemasCompatibleEvolution);
+  }
+
+  private static boolean isAtomicSchemasCompatibleEvolution(Schema oneAtomicType, Schema anotherAtomicType) {
+    // NOTE: Checking for compatibility of atomic types, we should ignore their
+    //       corresponding fully-qualified names (as irrelevant)
+    return isSchemaCompatible(anotherAtomicType, oneAtomicType, false, true);
+  }
+
   /**
    * Validate whether the {@code targetSchema} is a "compatible" projection of {@code sourceSchema}.
-   *
    * Only difference of this method from {@link #isStrictProjectionOf(Schema, Schema)} is
    * the fact that it allows some legitimate type promotions (like {@code int -> long},
    * {@code decimal(3, 2) -> decimal(5, 2)}, etc) that allows projection to have a "wider"