public List<File> process()
      throws Exception {
    try {
      return doProcess();
    } catch (Exception e) {
      // Cleaning up output dir as processing has failed. file managers left from map or reduce phase will be cleaned
      // up in the respective phases.
      FileUtils.deleteQuietly(_segmentsOutputDir);
      throw e;
    } finally {
      FileUtils.deleteDirectory(_mapperOutputDir);
      FileUtils.deleteDirectory(_reducerOutputDir);
    }
  }
@Test
  public void testSegmentGenerationWithComplexType() throws Exception {
    File workingDir = new File(TEMP_DIR, "single_segment_complex_type_output");
    FileUtils.forceMkdir(workingDir);
    IngestionConfig ingestionConfig = new IngestionConfig();
    List<String> fieldsToUnnest = new ArrayList<>();
    fieldsToUnnest.add("targetusers");

    ingestionConfig.setComplexTypeConfig(
        new ComplexTypeConfig(fieldsToUnnest, ".", null, null));
    _tableConfig.setIngestionConfig(ingestionConfig);
    // Default configs
    SegmentProcessorConfig config =
        new SegmentProcessorConfig.Builder().setTableConfig(_tableConfig).setSchema(_schemaWithComplexType).build();
    SegmentProcessorFramework framework =
        new SegmentProcessorFramework(_recordReaderWithComplexType, config, workingDir);
    List<File> outputSegments = framework.process();
    ImmutableSegment segment = ImmutableSegmentLoader.load(outputSegments.get(0), ReadMode.mmap);
    SegmentMetadata segmentMetadata = segment.getSegmentMetadata();
    // Pick the column created from complex type
    ColumnMetadata campaignInner2Metadata = segmentMetadata.getColumnMetadataFor("campaign.inner1.inner2");
    // Verify we see a specific value parsed from the complexType
    Assert.assertEquals(campaignInner2Metadata.getMinValue().compareTo("inner2v"), 0);
    ColumnMetadata campaignMetadata = segmentMetadata.getColumnMetadataFor("campaign");
    Assert.assertEquals(
        campaignMetadata.getMinValue().compareTo("{\"inner1\":{\"inner2\":\"inner2v\"},\"inner\":\"innerv\"}"), 0);

    ColumnMetadata listMetadata = segmentMetadata.getColumnMetadataFor("targetusers.user");
    Assert.assertEquals(listMetadata.getMinValue().compareTo("barfoo"), 0);
  }