public static InternalSchema reconcileSchema(Schema incomingSchema, InternalSchema oldTableSchema, boolean makeMissingFieldsNullable) {
    /* If incoming schema is null, we fall back on table schema. */
    if (incomingSchema.getType() == Schema.Type.NULL) {
      return oldTableSchema;
    }
    InternalSchema inComingInternalSchema = convert(incomingSchema, oldTableSchema.getNameToPosition());
    // check column add/missing
    List<String> colNamesFromIncoming = inComingInternalSchema.getAllColsFullName();
    List<String> colNamesFromOldSchema = oldTableSchema.getAllColsFullName();
    List<String> diffFromOldSchema = colNamesFromOldSchema.stream().filter(f -> !colNamesFromIncoming.contains(f)).collect(Collectors.toList());
    List<String> diffFromEvolutionColumns = colNamesFromIncoming.stream().filter(f -> !colNamesFromOldSchema.contains(f)).collect(Collectors.toList());
    // check type change.
    List<String> typeChangeColumns = colNamesFromIncoming
        .stream()
        .filter(f -> colNamesFromOldSchema.contains(f) && !inComingInternalSchema.findType(f).equals(oldTableSchema.findType(f)))
        .collect(Collectors.toList());
    if (colNamesFromIncoming.size() == colNamesFromOldSchema.size() && diffFromOldSchema.size() == 0 && typeChangeColumns.isEmpty()) {
      return oldTableSchema;
    }

    // Remove redundancy from diffFromEvolutionSchema.
    // for example, now we add a struct col in evolvedSchema, the struct col is " user struct<name:string, age:int> "
    // when we do diff operation: user, user.name, user.age will appear in the resultSet which is redundancy, user.name and user.age should be excluded.
    // deal with add operation
    TreeMap<Integer, String> finalAddAction = new TreeMap<>();
    for (int i = 0; i < diffFromEvolutionColumns.size(); i++) {
      String name = diffFromEvolutionColumns.get(i);
      int splitPoint = name.lastIndexOf(".");
      String parentName = splitPoint > 0 ? name.substring(0, splitPoint) : "";
      if (!parentName.isEmpty() && diffFromEvolutionColumns.contains(parentName)) {
        // find redundancy, skip it
        continue;
      }
      finalAddAction.put(inComingInternalSchema.findIdByName(name), name);
    }

    TableChanges.ColumnAddChange addChange = TableChanges.ColumnAddChange.get(oldTableSchema);
    finalAddAction.entrySet().stream().forEach(f -> {
      String name = f.getValue();
      int splitPoint = name.lastIndexOf(".");
      String parentName = splitPoint > 0 ? name.substring(0, splitPoint) : "";
      String rawName = splitPoint > 0 ? name.substring(splitPoint + 1) : name;
      // try to infer add position.
      java.util.Optional<String> inferPosition =
          colNamesFromIncoming.stream().filter(c ->
              c.lastIndexOf(".") == splitPoint
                  && c.startsWith(parentName)
                  && inComingInternalSchema.findIdByName(c) > inComingInternalSchema.findIdByName(name)
                  && oldTableSchema.findIdByName(c) > 0).sorted((s1, s2) -> oldTableSchema.findIdByName(s1) - oldTableSchema.findIdByName(s2)).findFirst();
      addChange.addColumns(parentName, rawName, inComingInternalSchema.findType(name), null);
      inferPosition.map(i -> addChange.addPositionChange(name, i, "before"));
    });

    // do type evolution.
    InternalSchema internalSchemaAfterAddColumns = SchemaChangeUtils.applyTableChanges2Schema(oldTableSchema, addChange);
    TableChanges.ColumnUpdateChange typeChange = TableChanges.ColumnUpdateChange.get(internalSchemaAfterAddColumns);
    typeChangeColumns.stream().filter(f -> !inComingInternalSchema.findType(f).isNestedType()).forEach(col -> {
      typeChange.updateColumnType(col, inComingInternalSchema.findType(col));
    });

    if (makeMissingFieldsNullable) {
      // mark columns missing from incoming schema as nullable
      Set<String> visited = new HashSet<>();
      diffFromOldSchema.stream()
          // ignore meta fields
          .filter(col -> !META_FIELD_NAMES.contains(col))
          .sorted()
          .forEach(col -> {
            // if parent is marked as nullable, only update the parent and not all the missing children field
            String parent = TableChangesHelper.getParentName(col);
            if (!visited.contains(parent)) {
              typeChange.updateColumnNullability(col, true);
            }
            visited.add(col);
          });
    }

    return SchemaChangeUtils.applyTableChanges2Schema(internalSchemaAfterAddColumns, typeChange);
  }
@Test
  public void testEvolutionSchemaFromNewAvroSchema() {
    Types.RecordType oldRecord = Types.RecordType.get(
        Types.Field.get(0, false, "id", Types.IntType.get()),
        Types.Field.get(1, true, "data", Types.StringType.get()),
        Types.Field.get(2, true, "preferences",
            Types.RecordType.get(
                Types.Field.get(5, false, "feature1", Types.BooleanType.get()),
                Types.Field.get(6, true, "featurex", Types.BooleanType.get()),
                Types.Field.get(7, true, "feature2", Types.BooleanType.get()))),
        Types.Field.get(3, false,"doubles", Types.ArrayType.get(8, false, Types.DoubleType.get())),
        Types.Field.get(4, false, "locations", Types.MapType.get(9, 10, Types.StringType.get(),
            Types.RecordType.get(
                Types.Field.get(11, false, "laty", Types.FloatType.get()),
                Types.Field.get(12, false, "long", Types.FloatType.get())), false)
        )
    );
    InternalSchema oldSchema = new InternalSchema(oldRecord);
    Types.RecordType evolvedRecord = Types.RecordType.get(
        Types.Field.get(0, false, "id", Types.IntType.get()),
        Types.Field.get(1, true, "data", Types.StringType.get()),
        Types.Field.get(2, true, "preferences",
            Types.RecordType.get(
                Types.Field.get(5, false, "feature1", Types.BooleanType.get()),
                Types.Field.get(5, true, "featurex", Types.BooleanType.get()),
                Types.Field.get(6, true, "feature2", Types.BooleanType.get()),
                Types.Field.get(5, true, "feature3", Types.BooleanType.get()))),
        Types.Field.get(3, false,"doubles", Types.ArrayType.get(7, false, Types.DoubleType.get())),
        Types.Field.get(4, false, "locations", Types.MapType.get(8, 9, Types.StringType.get(),
            Types.RecordType.get(
                Types.Field.get(10, false, "laty", Types.FloatType.get()),
                Types.Field.get(11, false, "long", Types.FloatType.get())), false)
        ),
        Types.Field.get(0, false, "add1", Types.IntType.get()),
        Types.Field.get(2, true, "addStruct",
            Types.RecordType.get(
                Types.Field.get(5, false, "nest1", Types.BooleanType.get()),
                Types.Field.get(5, true, "nest2", Types.BooleanType.get())))
    );
    evolvedRecord = (Types.RecordType)InternalSchemaBuilder.getBuilder().refreshNewId(evolvedRecord, new AtomicInteger(0));
    Schema evolvedAvroSchema = AvroInternalSchemaConverter.convert(evolvedRecord, "test1");
    InternalSchema result = AvroSchemaEvolutionUtils.reconcileSchema(evolvedAvroSchema, oldSchema, false);
    Types.RecordType checkedRecord = Types.RecordType.get(
        Types.Field.get(0, false, "id", Types.IntType.get()),
        Types.Field.get(1, true, "data", Types.StringType.get()),
        Types.Field.get(2, true, "preferences",
            Types.RecordType.get(
                Types.Field.get(5, false, "feature1", Types.BooleanType.get()),
                Types.Field.get(6, true, "featurex", Types.BooleanType.get()),
                Types.Field.get(7, true, "feature2", Types.BooleanType.get()),
                Types.Field.get(17, true, "feature3", Types.BooleanType.get()))),
        Types.Field.get(3, false,"doubles", Types.ArrayType.get(8, false, Types.DoubleType.get())),
        Types.Field.get(4, false, "locations", Types.MapType.get(9, 10, Types.StringType.get(),
            Types.RecordType.get(
                Types.Field.get(11, false, "laty", Types.FloatType.get()),
                Types.Field.get(12, false, "long", Types.FloatType.get())), false)
        ),
        Types.Field.get(13, true, "add1", Types.IntType.get()),
        Types.Field.get(14, true, "addStruct",
            Types.RecordType.get(
                Types.Field.get(15, false, "nest1", Types.BooleanType.get()),
                Types.Field.get(16, true, "nest2", Types.BooleanType.get())))
    );
    Assertions.assertEquals(result.getRecord(), checkedRecord);
  }