public static Set<String> getFieldsForRecordExtractor(@Nullable IngestionConfig ingestionConfig, Schema schema) {
    Set<String> fieldsForRecordExtractor = new HashSet<>();

    if (null != ingestionConfig && (null != ingestionConfig.getSchemaConformingTransformerConfig()
        || null != ingestionConfig.getSchemaConformingTransformerV2Config())) {
      // The SchemaConformingTransformer requires that all fields are extracted, indicated by returning an empty set
      // here. Compared to extracting the fields specified below, extracting all fields should be a superset.
      return fieldsForRecordExtractor;
    }

    extractFieldsFromIngestionConfig(ingestionConfig, fieldsForRecordExtractor);
    extractFieldsFromSchema(schema, fieldsForRecordExtractor);
    fieldsForRecordExtractor = getFieldsToReadWithComplexType(fieldsForRecordExtractor, ingestionConfig);
    return fieldsForRecordExtractor;
  }
@Test
  public void testComplexTypeConfig() {
    IngestionConfig ingestionConfig = new IngestionConfig();
    ComplexTypeConfig complexTypeConfig = new ComplexTypeConfig(null, "__",
        ComplexTypeConfig.CollectionNotUnnestedToJson.NON_PRIMITIVE, null);
    Schema schema = new Schema();

    ingestionConfig.setComplexTypeConfig(complexTypeConfig);
    schema.addField(new DimensionFieldSpec("a_b__c_d", FieldSpec.DataType.STRING, true));
    schema.addField(new DimensionFieldSpec("f_d", FieldSpec.DataType.STRING, false));
    schema.addField(new DimensionFieldSpec("ab__cd", FieldSpec.DataType.STRING, true));
    Set<String> fields = IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema);
    Assert.assertEquals(fields.size(), 3);
    Assert.assertTrue(fields.containsAll(Sets.newHashSet("a_b", "f_d", "ab")));
  }