build_dataframe_selects.scala

    /** Get the intersection of field names in new and old schema, newSchema is expected to be a superset */
    def orderedSelect(newSchema: StructType, oldSchema: StructType): Seq[String] = {
        fieldSelect(newSchema).intersect(fieldSelect(oldSchema))
    }

    /** build a list of field nameswith nested struct fields */ 
    def fieldSelect(schema: StructType): Seq[String] = {
        schema.flatMap { field =>
            field.dataType match {
                case struct: StructType => fieldSelect(struct).map(field.name + "." + _)
                case _ => Seq(field.name)
            }
         }
    }