private Schema flatten(Schema schema, boolean shouldPopulateLineage, boolean flattenComplexTypes) {
    Schema flattenedSchema;

    // Process all Schema Types
    // (Primitives are simply cloned)
    switch (schema.getType()) {
      case ARRAY:
        // Array might be an array of recursive Records, flatten them
        if (flattenComplexTypes) {
          flattenedSchema = Schema.createArray(flatten(schema.getElementType(), false));
        } else {
          flattenedSchema = Schema.createArray(schema.getElementType());
        }
        break;
      case BOOLEAN:
        flattenedSchema = Schema.create(schema.getType());
        break;
      case BYTES:
        flattenedSchema = Schema.create(schema.getType());
        break;
      case DOUBLE:
        flattenedSchema = Schema.create(schema.getType());
        break;
      case ENUM:
        flattenedSchema =
            Schema.createEnum(schema.getName(), schema.getDoc(), schema.getNamespace(), schema.getEnumSymbols());
        break;
      case FIXED:
        flattenedSchema =
            Schema.createFixed(schema.getName(), schema.getDoc(), schema.getNamespace(), schema.getFixedSize());
        break;
      case FLOAT:
        flattenedSchema = Schema.create(schema.getType());
        break;
      case INT:
        flattenedSchema = Schema.create(schema.getType());
        break;
      case LONG:
        flattenedSchema = Schema.create(schema.getType());
        break;
      case MAP:
        if (flattenComplexTypes) {
          flattenedSchema = Schema.createMap(flatten(schema.getValueType(), false));
        } else {
          flattenedSchema = Schema.createMap(schema.getValueType());
        }
        break;
      case NULL:
        flattenedSchema = Schema.create(schema.getType());
        break;
      case RECORD:
        flattenedSchema = flattenRecord(schema, shouldPopulateLineage, flattenComplexTypes);
        break;
      case STRING:
        flattenedSchema = Schema.create(schema.getType());
        break;
      case UNION:
        flattenedSchema = flattenUnion(schema, shouldPopulateLineage, flattenComplexTypes);
        break;
      default:
        String exceptionMessage = String.format("Schema flattening failed for \"%s\" ", schema);
        LOG.error(exceptionMessage);

        throw new AvroRuntimeException(exceptionMessage);
    }

    // Copy schema metadata
    copyProperties(schema, flattenedSchema);

    return flattenedSchema;
  }
private Schema flatten ( Schema schema , boolean shouldPopulateLineage , boolean flattenComplexTypes ) { Schema flattenedSchema ; switch ( schema . getType ( ) ) { case ARRAY : if ( flattenComplexTypes ) { flattenedSchema = Schema . createArray ( flatten ( schema . getElementType ( ) , false ) ) ; } else { flattenedSchema = Schema . createArray ( schema . getElementType ( ) ) ; } break ; case BOOLEAN : flattenedSchema = Schema . create ( schema . getType ( ) ) ; break ; case BYTES : flattenedSchema = Schema . create ( schema . getType ( ) ) ; break ; case DOUBLE : flattenedSchema = Schema . create ( schema . getType ( ) ) ; break ; case ENUM : flattenedSchema = Schema . createEnum ( schema . getName ( ) , schema . getDoc ( ) , schema . getNamespace ( ) , schema . getEnumSymbols ( ) ) ; break ; case FIXED : flattenedSchema = Schema . createFixed ( schema . getName ( ) , schema . getDoc ( ) , schema . getNamespace ( ) , schema . getFixedSize ( ) ) ; break ; case FLOAT : flattenedSchema = Schema . create ( schema . getType ( ) ) ; break ; case INT : flattenedSchema = Schema . create ( schema . getType ( ) ) ; break ; case LONG : flattenedSchema = Schema . create ( schema . getType ( ) ) ; break ; case MAP : if ( flattenComplexTypes ) { flattenedSchema = Schema . createMap ( flatten ( schema . getValueType ( ) , false ) ) ; } else { flattenedSchema = Schema . createMap ( schema . getValueType ( ) ) ; } break ; case NULL : flattenedSchema = Schema . create ( schema . getType ( ) ) ; break ; case RECORD : flattenedSchema = flattenRecord ( schema , shouldPopulateLineage , flattenComplexTypes ) ; break ; case STRING : flattenedSchema = Schema . create ( schema . getType ( ) ) ; break ; case UNION : flattenedSchema = flattenUnion ( schema , shouldPopulateLineage , flattenComplexTypes ) ; break ; default : String exceptionMessage = String . format ( "Schema flattening failed for \"%s\" " , schema ) ; LOG . error ( exceptionMessage ) ; throw new AvroRuntimeException ( exceptionMessage ) ; } copyProperties ( schema , flattenedSchema ) ; return flattenedSchema ; }
*
Flatten the Schema to un-nest recursive Records (to make it optimal for ORC)
@param schema Schema to flatten
@param shouldPopulateLineage is set to true if the field is going to be flattened and moved up the hierarchy -
so that lineage information can be tagged to it; which happens when there is a
Record within a Record OR Record within Option within Record and so on,
however not when there is a Record within Map or Array
@param flattenComplexTypes Flatten complex types recursively other than Record and Option
@return Flattened Avro Schema