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 testExtractFieldsIngestionConfig() {
    Schema schema = new Schema();

    // filter config
    IngestionConfig ingestionConfig = new IngestionConfig();
    ingestionConfig.setFilterConfig(new FilterConfig("Groovy({x > 100}, x)"));
    Set<String> fields = IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema);
    Assert.assertEquals(fields.size(), 1);
    Assert.assertTrue(fields.containsAll(Sets.newHashSet("x")));

    schema.addField(new DimensionFieldSpec("y", FieldSpec.DataType.STRING, true));
    fields = IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema);
    Assert.assertEquals(fields.size(), 2);
    Assert.assertTrue(fields.containsAll(Sets.newHashSet("x", "y")));

    // transform configs
    schema = new Schema.SchemaBuilder().addSingleValueDimension("d1", FieldSpec.DataType.STRING).build();
    ingestionConfig = new IngestionConfig();
    ingestionConfig.setTransformConfigs(
        Collections.singletonList(new TransformConfig("d1", "Groovy({function}, argument1, argument2)")));
    List<String> extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
    Assert.assertEquals(extract.size(), 3);
    Assert.assertTrue(extract.containsAll(Arrays.asList("d1", "argument1", "argument2")));

    // groovy function, no arguments
    ingestionConfig.setTransformConfigs(Collections.singletonList(new TransformConfig("d1", "Groovy({function})")));
    extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
    Assert.assertEquals(extract.size(), 1);
    Assert.assertTrue(extract.contains("d1"));

    // inbuilt functions
    schema = new Schema.SchemaBuilder().addSingleValueDimension("hoursSinceEpoch", FieldSpec.DataType.LONG).build();
    ingestionConfig.setTransformConfigs(
        Collections.singletonList(new TransformConfig("hoursSinceEpoch", "toEpochHours(timestampColumn)")));
    extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
    Assert.assertEquals(extract.size(), 2);
    Assert.assertTrue(extract.containsAll(Arrays.asList("timestampColumn", "hoursSinceEpoch")));

    // inbuilt functions with literal
    schema =
        new Schema.SchemaBuilder().addSingleValueDimension("tenMinutesSinceEpoch", FieldSpec.DataType.LONG).build();
    ingestionConfig.setTransformConfigs(Collections.singletonList(
        new TransformConfig("tenMinutesSinceEpoch", "toEpochMinutesBucket(timestampColumn, 10)")));
    extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
    Assert.assertEquals(extract.size(), 2);
    Assert.assertTrue(extract.containsAll(Lists.newArrayList("tenMinutesSinceEpoch", "timestampColumn")));

    // inbuilt functions on DateTimeFieldSpec
    schema = new Schema.SchemaBuilder().addDateTime("dateColumn", FieldSpec.DataType.STRING,
        "1:DAYS:SIMPLE_DATE_FORMAT:yyyy-MM-dd", "1:DAYS").build();
    ingestionConfig.setTransformConfigs(
        Collections.singletonList(new TransformConfig("dateColumn", "toDateTime(timestampColumn, 'yyyy-MM-dd')")));
    extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
    Assert.assertEquals(extract.size(), 2);
    Assert.assertTrue(extract.containsAll(Lists.newArrayList("dateColumn", "timestampColumn")));

    // filter + transform configs + schema fields  + schema transform
    schema = new Schema.SchemaBuilder().addSingleValueDimension("d1", FieldSpec.DataType.STRING)
        .addSingleValueDimension("d2", FieldSpec.DataType.STRING).addMetric("m1", FieldSpec.DataType.INT)
        .addDateTime("dateColumn", FieldSpec.DataType.STRING, "1:DAYS:SIMPLE_DATE_FORMAT:yyyy-MM-dd", "1:DAYS").build();
    schema.getFieldSpecFor("d2").setTransformFunction("reverse(xy)");
    ingestionConfig.setFilterConfig(new FilterConfig("Groovy({d1 == \"10\"}, d1)"));
    extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
    Assert.assertEquals(extract.size(), 6);
    Assert.assertTrue(extract.containsAll(Lists.newArrayList("d1", "d2", "m1", "dateColumn", "xy", "timestampColumn")));

    // filter + transform configs + schema fields  + schema transform + complex type configs
    schema = new Schema.SchemaBuilder().addSingleValueDimension("d1", FieldSpec.DataType.STRING)
        .addSingleValueDimension("d2", FieldSpec.DataType.STRING).addMetric("m1", FieldSpec.DataType.INT)
        .addDateTime("dateColumn", FieldSpec.DataType.STRING, "1:DAYS:SIMPLE_DATE_FORMAT:yyyy-MM-dd", "1:DAYS").build();
    schema.getFieldSpecFor("d2").setTransformFunction("reverse(xy)");
    ingestionConfig.setComplexTypeConfig(new ComplexTypeConfig(Arrays.asList("before.test", "after.test"), ".",
        ComplexTypeConfig.CollectionNotUnnestedToJson.NON_PRIMITIVE, Collections.singletonMap("before", "after")));
    extract = new ArrayList<>(IngestionUtils.getFieldsForRecordExtractor(ingestionConfig, schema));
    Assert.assertEquals(extract.size(), 8);
    List<String> expectedColumns =
        Arrays.asList("d1", "d2", "m1", "dateColumn", "xy", "timestampColumn", "before", "after");
    Assert.assertTrue(extract.containsAll(expectedColumns));
  }