class AvroSchemaUtil {

  public static Schema extractSchemaFromNullable(Schema schema, String columnName) {
    if (!schema.isUnion()) {
      throw UserException.validationError()
        .message("Expected union type, but received: %s", schema.getType())
        .addContext("Column", columnName)
        .build(logger);
    }
    List<Schema> unionSchemas = schema.getTypes();

    // exclude all schemas with null type
    List<Schema> nonNullSchemas = unionSchemas.stream()
      .filter(unionSchema -> !Schema.Type.NULL.equals(unionSchema.getType()))
      .collect(Collectors.toList());

    // if original schema has two elements and only one non-nullable schema, this is simple nullable type
    if (unionSchemas.size() == 2 && nonNullSchemas.size() == 1) {
      return nonNullSchemas.get(0);
    } else {
      return throwUnsupportedErrorForType("complex union", columnName);
    }
  }

}