[HUDI-7713] Enforce ordering of fields during schema reconciliation (apache#11154)

* Enforce ordering of fields during schema reconciliation

* update ordering of expected columns

* update tests to match new expectations

* update test field ordering

* set input batch schema

* reorder row fields as well

* only reorder fields if reconcile is false

* fix schema evoution test expectations around field ordering

* minor updates

* allow for different ordering modes

* use position instead of existin field ID

* pr feedback, add test, fix visitor

* cleanup test

* fix assertion

* pr feedback

Author: the-other-tim-brown

hudi-common/src/main/java/org/apache/hudi/internal/schema/InternalSchema.java

@@ -37,6 +37,9 @@
  * used to support schema evolution.
  */
 public class InternalSchema implements Serializable {
+  public static final String ARRAY_ELEMENT = "element";
+  public static final String MAP_KEY = "key";
+  public static final String MAP_VALUE = "value";
 
   private static final InternalSchema EMPTY_SCHEMA = new InternalSchema(-1L, RecordType.get());
 
@@ -50,6 +53,7 @@ public class InternalSchema implements Serializable {
   private transient Map<Integer, Field> idToField = null;
   private transient Map<String, Integer> nameToId = null;
   private transient Map<Integer, String> idToName = null;
+  private transient Map<String, Integer> nameToPosition = null;
 
   public static InternalSchema getEmptyInternalSchema() {
     return EMPTY_SCHEMA;
@@ -268,6 +272,19 @@ public int findIdByName(String name) {
     return buildNameToId().getOrDefault(name, -1);
   }
 
+  /**
+   * Returns the full name of the field and its position in the schema.
+   * This differs from its ID in cases where new fields are not appended to the end of schemas.
+   * The output is used when reconciling the order of fields while ingesting.
+   * @return a mapping from full field name to a position
+   */
+  public Map<String, Integer> getNameToPosition() {
+    if (nameToPosition == null) {
+      nameToPosition = InternalSchemaBuilder.getBuilder().buildNameToPosition(record);
+    }
+    return nameToPosition;
+  }
+
   @Override
   public String toString() {
     return String.format("table {\n%s\n}",

hudi-common/src/main/java/org/apache/hudi/internal/schema/InternalSchemaBuilder.java

@@ -20,6 +20,7 @@
 
 import org.apache.hudi.internal.schema.visitor.InternalSchemaVisitor;
 import org.apache.hudi.internal.schema.visitor.NameToIDVisitor;
+import org.apache.hudi.internal.schema.visitor.NameToPositionVisitor;
 
 import java.io.Serializable;
 import java.util.ArrayList;
@@ -67,6 +68,10 @@ public Map<String, Integer> buildNameToId(Type type) {
     return visit(type, new NameToIDVisitor());
   }
 
+  Map<String, Integer> buildNameToPosition(Type type) {
+    return visit(type, new NameToPositionVisitor());
+  }
+
   /**
    * Use to traverse all types in internalSchema with visitor.
    *

hudi-common/src/main/java/org/apache/hudi/internal/schema/convert/AvroInternalSchemaConverter.java

@@ -31,13 +31,16 @@
 import org.apache.avro.Schema;
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
 import java.util.Deque;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
 
 import static org.apache.avro.Schema.Type.UNION;
 
@@ -117,10 +120,18 @@ public static Schema convert(Type type, String name) {
 
   /** Convert an avro schema into internal type. */
   public static Type convertToField(Schema schema) {
-    return buildTypeFromAvroSchema(schema);
+    return buildTypeFromAvroSchema(schema, Collections.emptyMap());
+  }
+
+  private static Type convertToField(Schema schema, Map<String, Integer> existingFieldNameToPositionMapping) {
+    return buildTypeFromAvroSchema(schema, existingFieldNameToPositionMapping);
   }
 
   /** Convert an avro schema into internalSchema. */
+  public static InternalSchema convert(Schema schema, Map<String, Integer> existingFieldNameToPositionMapping) {
+    return new InternalSchema((Types.RecordType) convertToField(schema, existingFieldNameToPositionMapping));
+  }
+
   public static InternalSchema convert(Schema schema) {
     return new InternalSchema((Types.RecordType) convertToField(schema));
   }
@@ -151,11 +162,11 @@ public static Schema nullableSchema(Schema schema) {
    * @param schema a avro schema.
    * @return a hudi type.
    */
-  public static Type buildTypeFromAvroSchema(Schema schema) {
+  public static Type buildTypeFromAvroSchema(Schema schema, Map<String, Integer> existingNameToPositions) {
     // set flag to check this has not been visited.
-    Deque<String> visited = new LinkedList();
-    AtomicInteger nextId = new AtomicInteger(1);
-    return visitAvroSchemaToBuildType(schema, visited, true, nextId);
+    Deque<String> visited = new LinkedList<>();
+    AtomicInteger nextId = new AtomicInteger(0);
+    return visitAvroSchemaToBuildType(schema, visited, "", nextId, existingNameToPositions);
   }
 
   private static void checkNullType(Type fieldType, String fieldName, Deque<String> visited) {
@@ -172,11 +183,11 @@ private static void checkNullType(Type fieldType, String fieldName, Deque<String
       throw new HoodieNullSchemaTypeException(sb.toString());
     } else if (fieldType.typeId() == Type.TypeID.ARRAY) {
       visited.push(fieldName);
-      checkNullType(((Types.ArrayType) fieldType).elementType(), "element", visited);
+      checkNullType(((Types.ArrayType) fieldType).elementType(), InternalSchema.ARRAY_ELEMENT, visited);
       visited.pop();
     } else if (fieldType.typeId() == Type.TypeID.MAP) {
       visited.push(fieldName);
-      checkNullType(((Types.MapType) fieldType).valueType(), "value", visited);
+      checkNullType(((Types.MapType) fieldType).valueType(), InternalSchema.MAP_VALUE, visited);
       visited.pop();
     }
   }
@@ -186,28 +197,27 @@ private static void checkNullType(Type fieldType, String fieldName, Deque<String
    *
    * @param schema a avro schema.
    * @param visited track the visit node when do traversal for avro schema; used to check if the name of avro record schema is correct.
-   * @param firstVisitRoot track whether the current visited schema node is a root node.
+   * @param currentFieldPath the dot-separated path to the current field; empty at the root and always ends in a '.' otherwise for ease of concatenation.
    * @param nextId an initial id which used to create id for all fields.
    * @return a hudi type match avro schema.
    */
-  private static Type visitAvroSchemaToBuildType(Schema schema, Deque<String> visited, Boolean firstVisitRoot, AtomicInteger nextId) {
+  private static Type visitAvroSchemaToBuildType(Schema schema, Deque<String> visited, String currentFieldPath, AtomicInteger nextId, Map<String, Integer> existingNameToPosition) {
     switch (schema.getType()) {
       case RECORD:
         String name = schema.getFullName();
         if (visited.contains(name)) {
           throw new HoodieSchemaException(String.format("cannot convert recursive avro record %s", name));
         }
         visited.push(name);
-        List<Schema.Field> fields = schema.getFields();
+        List<Schema.Field> fields = existingNameToPosition.isEmpty() ? schema.getFields() :
+            schema.getFields().stream()
+                .sorted(Comparator.comparing(field -> existingNameToPosition.getOrDefault(currentFieldPath + field.name(), Integer.MAX_VALUE)))
+                .collect(Collectors.toList());
         List<Type> fieldTypes = new ArrayList<>(fields.size());
         int nextAssignId = nextId.get();
-        // when first visit root record, set nextAssignId = 0;
-        if (firstVisitRoot) {
-          nextAssignId = 0;
-        }
         nextId.set(nextAssignId + fields.size());
-        fields.stream().forEach(field -> {
-          Type fieldType = visitAvroSchemaToBuildType(field.schema(), visited, false, nextId);
+        fields.forEach(field -> {
+          Type fieldType = visitAvroSchemaToBuildType(field.schema(), visited, currentFieldPath + field.name() + ".", nextId, existingNameToPosition);
           checkNullType(fieldType, field.name(), visited);
           fieldTypes.add(fieldType);
         });
@@ -226,22 +236,24 @@ private static Type visitAvroSchemaToBuildType(Schema schema, Deque<String> visi
         //       them up into namespace/struct-name pair)
         return Types.RecordType.get(internalFields, schema.getFullName());
       case UNION:
-        List<Type> fTypes = new ArrayList<>();
-        schema.getTypes().stream().forEach(t -> {
-          fTypes.add(visitAvroSchemaToBuildType(t, visited, false, nextId));
+        List<Type> fTypes = new ArrayList<>(2);
+        schema.getTypes().forEach(t -> {
+          fTypes.add(visitAvroSchemaToBuildType(t, visited, currentFieldPath, nextId, existingNameToPosition));
         });
         return fTypes.get(0) == null ? fTypes.get(1) : fTypes.get(0);
       case ARRAY:
+        String elementPath = currentFieldPath + InternalSchema.ARRAY_ELEMENT + ".";
         Schema elementSchema = schema.getElementType();
         int elementId = nextId.get();
         nextId.set(elementId + 1);
-        Type elementType = visitAvroSchemaToBuildType(elementSchema, visited, false, nextId);
+        Type elementType = visitAvroSchemaToBuildType(elementSchema, visited, elementPath, nextId, existingNameToPosition);
         return Types.ArrayType.get(elementId, AvroInternalSchemaConverter.isOptional(schema.getElementType()), elementType);
       case MAP:
         int keyId = nextId.get();
         int valueId = keyId + 1;
         nextId.set(valueId + 1);
-        Type valueType = visitAvroSchemaToBuildType(schema.getValueType(),  visited, false, nextId);
+        String valuePath = currentFieldPath + InternalSchema.MAP_VALUE + ".";
+        Type valueType = visitAvroSchemaToBuildType(schema.getValueType(), visited, valuePath, nextId, existingNameToPosition);
         return Types.MapType.get(keyId, valueId, Types.StringType.get(), valueType, AvroInternalSchemaConverter.isOptional(schema.getValueType()));
       default:
         return visitAvroPrimitiveToBuildInternalType(schema);

hudi-common/src/main/java/org/apache/hudi/internal/schema/utils/AvroSchemaEvolutionUtils.java

@@ -24,6 +24,7 @@
 import org.apache.avro.Schema;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.TreeMap;
 import java.util.stream.Collectors;
@@ -58,7 +59,7 @@ public static InternalSchema reconcileSchema(Schema incomingSchema, InternalSche
     if (incomingSchema.getType() == Schema.Type.NULL) {
       return oldTableSchema;
     }
-    InternalSchema inComingInternalSchema = convert(incomingSchema);
+    InternalSchema inComingInternalSchema = convert(incomingSchema, oldTableSchema.getNameToPosition());
     // check column add/missing
     List<String> colNamesFromIncoming = inComingInternalSchema.getAllColsFullName();
     List<String> colNamesFromOldSchema = oldTableSchema.getAllColsFullName();
@@ -136,7 +137,7 @@ public static Schema reconcileSchema(Schema incomingSchema, Schema oldTableSchem
    * @param targetSchema target schema that source schema will be reconciled against
    * @return schema (based off {@code source} one) that has nullability constraints and datatypes reconciled
    */
-  public static Schema reconcileSchemaRequirements(Schema sourceSchema, Schema targetSchema) {
+  public static Schema reconcileSchemaRequirements(Schema sourceSchema, Schema targetSchema, boolean shouldReorderColumns) {
     if (targetSchema.getType() == Schema.Type.NULL || targetSchema.getFields().isEmpty()) {
       return sourceSchema;
     }
@@ -145,8 +146,9 @@ public static Schema reconcileSchemaRequirements(Schema sourceSchema, Schema tar
       return targetSchema;
     }
 
-    InternalSchema sourceInternalSchema = convert(sourceSchema);
     InternalSchema targetInternalSchema = convert(targetSchema);
+    // Use existing fieldIds for consistent field ordering between commits when shouldReorderColumns is true
+    InternalSchema sourceInternalSchema = convert(sourceSchema, shouldReorderColumns ? targetInternalSchema.getNameToPosition() : Collections.emptyMap());
 
     List<String> colNamesSourceSchema = sourceInternalSchema.getAllColsFullName();
     List<String> colNamesTargetSchema = targetInternalSchema.getAllColsFullName();

hudi-common/src/main/java/org/apache/hudi/internal/schema/visitor/NameToIDVisitor.java

@@ -95,14 +95,14 @@ public Map<String, Integer> field(Types.Field field, Map<String, Integer> fieldR
 
   @Override
   public Map<String, Integer> array(Types.ArrayType array, Map<String, Integer> elementResult) {
-    nameToId.put(createFullName("element", fieldNames), array.elementId());
+    nameToId.put(createFullName(InternalSchema.ARRAY_ELEMENT, fieldNames), array.elementId());
     return nameToId;
   }
 
   @Override
   public Map<String, Integer> map(Types.MapType map, Map<String, Integer> keyResult, Map<String, Integer> valueResult) {
-    nameToId.put(createFullName("key", fieldNames), map.keyId());
-    nameToId.put(createFullName("value", fieldNames), map.valueId());
+    nameToId.put(createFullName(InternalSchema.MAP_KEY, fieldNames), map.keyId());
+    nameToId.put(createFullName(InternalSchema.MAP_VALUE, fieldNames), map.valueId());
     return nameToId;
   }
 

hudi-common/src/main/java/org/apache/hudi/internal/schema/visitor/NameToPositionVisitor.java

@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.internal.schema.visitor;
+
+import org.apache.hudi.internal.schema.InternalSchema;
+import org.apache.hudi.internal.schema.Type;
+import org.apache.hudi.internal.schema.Types;
+
+import java.util.Deque;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.hudi.internal.schema.utils.InternalSchemaUtils.createFullName;
+
+/**
+ * Schema visitor to produce name -> position map for internalSchema. Positions are assigned in a depth-first manner.
+ */
+public class NameToPositionVisitor extends InternalSchemaVisitor<Map<String, Integer>> {
+  private final Deque<String> fieldNames = new LinkedList<>();
+  private final Map<String, Integer> nameToId = new HashMap<>();
+  private final AtomicInteger position = new AtomicInteger(0);
+
+  @Override
+  public void beforeField(Types.Field field) {
+    nameToId.put(createFullName(field.name(), fieldNames), position.getAndIncrement());
+    fieldNames.push(field.name());
+  }
+
+  @Override
+  public void afterField(Types.Field field) {
+    fieldNames.pop();
+  }
+
+  @Override
+  public void beforeArrayElement(Types.Field elementField) {
+    nameToId.put(createFullName(InternalSchema.ARRAY_ELEMENT, fieldNames), position.getAndIncrement());
+    fieldNames.push(elementField.name());
+  }
+
+  @Override
+  public void afterArrayElement(Types.Field elementField) {
+    fieldNames.pop();
+  }
+
+  @Override
+  public void beforeMapKey(Types.Field keyField) {
+    nameToId.put(createFullName(InternalSchema.MAP_KEY, fieldNames), position.getAndIncrement());
+    fieldNames.push(keyField.name());
+  }
+
+  @Override
+  public void afterMapKey(Types.Field keyField) {
+    fieldNames.pop();
+  }
+
+  @Override
+  public void beforeMapValue(Types.Field valueField) {
+    nameToId.put(createFullName(InternalSchema.MAP_VALUE, fieldNames), position.getAndIncrement());
+    fieldNames.push(valueField.name());
+  }
+
+  @Override
+  public void afterMapValue(Types.Field valueField) {
+    fieldNames.pop();
+  }
+
+  @Override
+  public Map<String, Integer> schema(InternalSchema schema, Map<String, Integer> recordResult) {
+    return nameToId;
+  }
+
+  @Override
+  public Map<String, Integer> record(Types.RecordType record, List<Map<String, Integer>> fieldResults) {
+    return nameToId;
+  }
+
+  @Override
+  public Map<String, Integer> field(Types.Field field, Map<String, Integer> fieldResult) {
+    return nameToId;
+  }
+
+  @Override
+  public Map<String, Integer> array(Types.ArrayType array, Map<String, Integer> elementResult) {
+    return nameToId;
+  }
+
+  @Override
+  public Map<String, Integer> map(Types.MapType map, Map<String, Integer> keyResult, Map<String, Integer> valueResult) {
+    return nameToId;
+  }
+
+  @Override
+  public Map<String, Integer> primitive(Type.PrimitiveType primitive) {
+    return nameToId;
+  }
+}

hudi-common/src/test/java/org/apache/hudi/internal/schema/utils/TestAvroSchemaEvolutionUtils.java

@@ -152,9 +152,9 @@ private Schema create(String name, Schema.Field... fields) {
 
   @Test
   public void testArrayType() {
-    Type arrayNestRecordType = Types.ArrayType.get(1, false,
-        Types.RecordType.get(Arrays.asList(Types.Field.get(2, false, "a", Types.FloatType.get()),
-            Types.Field.get(3, false, "b", Types.FloatType.get()))));
+    Type arrayNestRecordType = Types.ArrayType.get(0, false,
+        Types.RecordType.get(Arrays.asList(Types.Field.get(1, false, "a", Types.FloatType.get()),
+            Types.Field.get(2, false, "b", Types.FloatType.get()))));
 
     Schema schema = SchemaBuilder.array().items(create("t1",
         new Schema.Field("a", Schema.create(Schema.Type.FLOAT), null, null),
@@ -526,15 +526,16 @@ public void testReconcileSchema() {
         new Schema.Field("d2", AvroInternalSchemaConverter.nullableSchema(LogicalTypes.date().addToSchema(Schema.create(Schema.Type.INT))), null, JsonProperties.NULL_VALUE));
 
     Schema simpleCheckSchema = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"simple\",\"fields\":[{\"name\":\"a\",\"type\":[\"null\",\"boolean\"],\"default\":null},"
-        + "{\"name\":\"b\",\"type\":[\"null\",\"int\"],\"default\":null},{\"name\":\"a1\",\"type\":[\"null\",\"long\"],\"default\":null},"
+        + "{\"name\":\"b\",\"type\":[\"null\",\"int\"],\"default\":null},"
         + "{\"name\":\"c\",\"type\":[\"null\",\"long\"],\"default\":null},"
-        + "{\"name\":\"c1\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"c2\",\"type\":[\"null\",\"long\"],\"default\":null},"
         + "{\"name\":\"d\",\"type\":[\"null\",{\"type\":\"int\",\"logicalType\":\"date\"}],\"default\":null},"
+        + "{\"name\":\"a1\",\"type\":[\"null\",\"long\"],\"default\":null},"
+        + "{\"name\":\"c1\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"c2\",\"type\":[\"null\",\"long\"],\"default\":null},"
         + "{\"name\":\"d1\",\"type\":[\"null\",{\"type\":\"int\",\"logicalType\":\"date\"}],\"default\":null},"
         + "{\"name\":\"d2\",\"type\":[\"null\",{\"type\":\"int\",\"logicalType\":\"date\"}],\"default\":null}]}");
 
     Schema simpleReconcileSchema = AvroInternalSchemaConverter.convert(AvroSchemaEvolutionUtils
         .reconcileSchema(incomingSchema, AvroInternalSchemaConverter.convert(schema)), "schemaNameFallback");
-    Assertions.assertEquals(simpleReconcileSchema, simpleCheckSchema);
+    Assertions.assertEquals(simpleCheckSchema, simpleReconcileSchema);
   }
 }