ddl.scala

   /**
    * This checks that the two columns are compatible.  Here, compatible means that:
    * - Their names are resolvable (usually only case differences)
    * - AND
    * - They are either:
    * -- The same dataType
    * -- StructTypes, for which common sub fields are all (recursively) compatible.
    *
    * @param ours             our schema
    * @param theirs           their schema
    * @return
    */
  private def columnsCompatible(
    ours: StructField,
    theirs: StructField,
    resolver: Resolver
  ): Boolean = {
    // If we can't resolve between the column names, then no these could be compatible.
    if (!resolver(ours.name, theirs.name)) {
      false
    }

    // Else ok, we resolved names and they have same dataType, cool!
    else if (ours.dataType == theirs.dataType) {
      true
    }

    // Else if we get here, the types didn't match, but both types were StructTypes.
    // We need to recurse into theirs to check each sub field.
    else if (isStructType(ours) && isStructType(theirs)) {
      // Check that any fields in theirSchema that are also in our schema
      // have identical types.  The exception is StructType fields, for which
      // we will recurse and ensure that each of the contained StructFields
      // themselves have identical types.
      val ourSchema = ours.dataType.asInstanceOf[StructType]
      val theirSchema = theirs.dataType.asInstanceOf[StructType]

      // Get all fields in theirSchema that are also in ourSchema
      // and make sure that all of these fields are also compatible,
      // Recursing if needed.
      theirSchema
        .filter(field => findColumnByName(ourSchema, field.name, resolver).isDefined)
        .forall { theirField => columnsCompatible(
            theirField,
            findColumnByName(ourSchema, theirField.name, resolver).get,
            resolver
        )}

    }

    // Else at least one of the dataTypes was not a StructType,
    // definitely not compatible.
    else {
      false
    }
  }