@Override
  public void run()
      throws Exception {
    //init all file systems
    List<PinotFSSpec> pinotFSSpecs = _spec.getPinotFSSpecs();
    for (PinotFSSpec pinotFSSpec : pinotFSSpecs) {
      PinotFSFactory.register(pinotFSSpec.getScheme(), pinotFSSpec.getClassName(), new PinotConfiguration(pinotFSSpec));
    }
    //Get list of files to process
    URI inputDirURI = new URI(_spec.getInputDirURI());
    if (inputDirURI.getScheme() == null) {
      inputDirURI = new File(_spec.getInputDirURI()).toURI();
    }
    PinotFS inputDirFS = PinotFSFactory.create(inputDirURI.getScheme());
    List<String> filteredFiles = SegmentGenerationUtils.listMatchedFilesWithRecursiveOption(inputDirFS, inputDirURI,
        _spec.getIncludeFileNamePattern(), _spec.getExcludeFileNamePattern(), _spec.isSearchRecursively());

    //Get outputFS for writing output pinot segments
    URI outputDirURI = new URI(_spec.getOutputDirURI());
    if (outputDirURI.getScheme() == null) {
      outputDirURI = new File(_spec.getOutputDirURI()).toURI();
    }
    PinotFS outputDirFS = PinotFSFactory.create(outputDirURI.getScheme());
    outputDirFS.mkdir(outputDirURI);

    //Get staging directory for temporary output pinot segments
    String stagingDir = _spec.getExecutionFrameworkSpec().getExtraConfigs().get(STAGING_DIR_FIELD);
    Preconditions.checkNotNull(stagingDir, "Please set config: stagingDir under 'executionFrameworkSpec.extraConfigs'");
    URI stagingDirURI = URI.create(stagingDir);
    if (stagingDirURI.getScheme() == null) {
      stagingDirURI = new File(stagingDir).toURI();
    }
    if (!outputDirURI.getScheme().equals(stagingDirURI.getScheme())) {
      throw new RuntimeException(String
          .format("The scheme of staging directory URI [%s] and output directory URI [%s] has to be same.",
              stagingDirURI, outputDirURI));
    }
    if (outputDirFS.exists(stagingDirURI)) {
      LOGGER.info("Clearing out existing staging directory: [{}]", stagingDirURI);
      outputDirFS.delete(stagingDirURI, true);
    }

    outputDirFS.mkdir(stagingDirURI);
    Path stagingInputDir = new Path(stagingDirURI.toString(), "input");
    outputDirFS.mkdir(stagingInputDir.toUri());
    Path stagingSegmentTarUri = new Path(stagingDirURI.toString(), SEGMENT_TAR_SUBDIR_NAME);
    outputDirFS.mkdir(stagingSegmentTarUri.toUri());

    // numDataFiles is guaranteed to be greater than zero since listMatchedFilesWithRecursiveOption will throw
    // runtime exception if the matched files list is empty.
    int numDataFiles = filteredFiles.size();

    LOGGER.info("Creating segments with data files: {}", filteredFiles);
    if (!SegmentGenerationJobUtils.useGlobalDirectorySequenceId(_spec.getSegmentNameGeneratorSpec())) {
      Map<String, List<String>> localDirIndex = new HashMap<>();
      for (String filteredFile : filteredFiles) {
        java.nio.file.Path filteredParentPath = Paths.get(filteredFile).getParent();
        if (!localDirIndex.containsKey(filteredParentPath.toString())) {
          localDirIndex.put(filteredParentPath.toString(), new ArrayList<>());
        }
        localDirIndex.get(filteredParentPath.toString()).add(filteredFile);
      }
      for (String parentPath : localDirIndex.keySet()) {
        List<String> siblingFiles = localDirIndex.get(parentPath);
        Collections.sort(siblingFiles);
        for (int i = 0; i < siblingFiles.size(); i++) {
          URI inputFileURI = SegmentGenerationUtils.getFileURI(siblingFiles.get(i),
              SegmentGenerationUtils.getDirectoryURI(parentPath));
          createInputFileUriAndSeqIdFile(inputFileURI, outputDirFS, stagingInputDir, i);
        }
      }
    } else {
      for (int i = 0; i < numDataFiles; i++) {
        URI inputFileURI = SegmentGenerationUtils.getFileURI(filteredFiles.get(i), inputDirURI);
        createInputFileUriAndSeqIdFile(inputFileURI, outputDirFS, stagingInputDir, i);
      }
    }
    try {
      // Set up the job
      Job job = Job.getInstance(getConf());
      job.setJobName(getClass().getSimpleName());

      // Our class is in the batch-ingestion-hadoop plugin, so we want to pick a class
      // that's in the main jar (the pinot-all-${PINOT_VERSION}-jar-with-dependencies.jar)
      job.setJarByClass(SegmentGenerationJobSpec.class);

      // Disable speculative execution, as otherwise two map tasks can wind up writing to the same staging file.
      job.getConfiguration().setBoolean(MRJobConfig.MAP_SPECULATIVE, false);

      // But we have to copy ourselves to HDFS, and add us to the distributed cache, so
      // that the mapper code is available.
      addMapperJarToDistributedCache(job, outputDirFS, stagingDirURI);

      org.apache.hadoop.conf.Configuration jobConf = job.getConfiguration();
      String hadoopTokenFileLocation = System.getenv("HADOOP_TOKEN_FILE_LOCATION");
      if (hadoopTokenFileLocation != null) {
        jobConf.set("mapreduce.job.credentials.binary", hadoopTokenFileLocation);
      }
      int jobParallelism = _spec.getSegmentCreationJobParallelism();
      if (jobParallelism <= 0 || jobParallelism > numDataFiles) {
        jobParallelism = numDataFiles;
      }
      jobConf.setInt(JobContext.NUM_MAPS, jobParallelism);

      // Pinot plugins are necessary to launch Pinot ingestion job from every mapper.
      // In order to ensure pinot plugins would be loaded to each worker, this method
      // tars entire plugins directory and set this file into Distributed cache.
      // Then each mapper job will untar the plugin tarball, and set system properties accordingly.
      // Note that normally we'd just use Hadoop's support for putting jars on the
      // classpath via the distributed cache, but some of the plugins (e.g. the pinot-parquet
      // input format) include Hadoop classes, which can be incompatibile with the Hadoop
      // installation/jars being used to run the mapper, leading to errors such as:
      // java.lang.NoSuchMethodError: org.apache.hadoop.ipc.RPC.getServer(...
      //
      packPluginsToDistributedCache(job, outputDirFS, stagingDirURI);

      // Add dependency jars, if we're provided with a directory containing these.
      String dependencyJarsSrcDir = _spec.getExecutionFrameworkSpec().getExtraConfigs().get(DEPS_JAR_DIR_FIELD);
      if (dependencyJarsSrcDir != null) {
        Path dependencyJarsDestPath = new Path(stagingDirURI.toString(), DEPS_JAR_SUBDIR_NAME);
        addJarsToDistributedCache(job, new File(dependencyJarsSrcDir), outputDirFS, dependencyJarsDestPath.toUri(),
            false);
      }

      _spec.setOutputDirURI(stagingSegmentTarUri.toUri().toString());
      jobConf.set(SEGMENT_GENERATION_JOB_SPEC, new Yaml().dumpAsMap(_spec));
      _spec.setOutputDirURI(outputDirURI.toString());

      job.setMapperClass(getMapperClass());
      job.setNumReduceTasks(0);

      job.setInputFormatClass(TextInputFormat.class);
      job.setOutputFormatClass(TextOutputFormat.class);

      job.setMapOutputKeyClass(LongWritable.class);
      job.setMapOutputValueClass(Text.class);

      FileInputFormat.addInputPath(job, stagingInputDir);
      FileOutputFormat.setOutputPath(job, new Path(stagingDir, "output"));

      // Submit the job
      job.waitForCompletion(true);
      if (!job.isSuccessful()) {
        throw new RuntimeException("Job failed: " + job);
      }

      LOGGER.info("Moving segment tars from staging directory [{}] to output directory [{}]", stagingDirURI,
          outputDirURI);
      moveFiles(outputDirFS, new Path(stagingDir, SEGMENT_TAR_SUBDIR_NAME).toUri(), outputDirURI,
          _spec.isOverwriteOutput());
    } finally {
      LOGGER.info("Trying to clean up staging directory: [{}]", stagingDirURI);
      outputDirFS.delete(stagingDirURI, true);
    }
  }
@Test
  public void testSegmentGeneration() throws Exception {
    File testDir = Files.createTempDirectory("testSegmentGeneration-").toFile();
    testDir.delete();
    testDir.mkdirs();

    File inputDir = new File(testDir, "input");
    inputDir.mkdirs();
    File inputFile = new File(inputDir, "input.csv");
    FileUtils.writeLines(inputFile, Lists.newArrayList("col1,col2", "value1,1", "value2,2"));

    final String outputFilename = "myTable_OFFLINE_0.tar.gz";
    final String otherFilename = "myTable_OFFLINE_100.tar.gz";
    File outputDir = new File(testDir, "output");
    FileUtils.touch(new File(outputDir, outputFilename));
    FileUtils.touch(new File(outputDir, otherFilename));

    // Set up schema file.
    final String schemaName = "myTable";
    File schemaFile = new File(testDir, "myTable.schema");
    Schema schema = new SchemaBuilder()
      .setSchemaName(schemaName)
      .addSingleValueDimension("col1", DataType.STRING)
      .addMetric("col2", DataType.INT)
      .build();
    FileUtils.write(schemaFile, schema.toPrettyJsonString(), StandardCharsets.UTF_8);

    // Set up table config file.
    File tableConfigFile = new File(testDir, "myTable.table");
    TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE)
      .setTableName("myTable")
      .setNumReplicas(1)
      .build();
    FileUtils.write(tableConfigFile, tableConfig.toJsonString(), StandardCharsets.UTF_8);

    File stagingDir = new File(testDir, "staging");
    stagingDir.mkdir();
    // Add the staging output dir, which should cause code to fail unless we've added code to remove
    // the staging dir if it exists.
    FileUtils.touch(new File(stagingDir, "output"));

    // Set up a plugins dir, with a sub-directory. We'll use an external jar,
    // since using a class inside of Pinot to find the enclosing jar is somehow
    // finding the directory of classes vs. the actual jar, on the build server
    // (though it works fine in other configurations).
    File pluginsDir = new File(testDir, "plugins");
    File myPluginDir = new File(pluginsDir, "my-plugin");
    myPluginDir.mkdirs();
    File pluginJar = new File(StringUtils.class.getProtectionDomain().getCodeSource().getLocation().toURI());
    FileUtils.copyFile(pluginJar, new File(myPluginDir, pluginJar.getName()));

    // Set up dependency jars dir.
    // FUTURE set up jar with class that we need for reading file, so we know it's working
    File dependencyJarsDir = new File(testDir, "jars");
    dependencyJarsDir.mkdir();
    File extraJar = new File(Gson.class.getProtectionDomain().getCodeSource().getLocation().toURI());
    FileUtils.copyFile(extraJar, new File(dependencyJarsDir, extraJar.getName()));

    SegmentGenerationJobSpec jobSpec = new SegmentGenerationJobSpec();
    jobSpec.setJobType("SegmentCreation");
    jobSpec.setInputDirURI(inputDir.toURI().toString());
    jobSpec.setOutputDirURI(outputDir.toURI().toString());
    jobSpec.setOverwriteOutput(false);

    RecordReaderSpec recordReaderSpec = new RecordReaderSpec();
    recordReaderSpec.setDataFormat("csv");
    recordReaderSpec.setClassName(CSVRecordReader.class.getName());
    recordReaderSpec.setConfigClassName(CSVRecordReaderConfig.class.getName());
    jobSpec.setRecordReaderSpec(recordReaderSpec);

    TableSpec tableSpec = new TableSpec();
    tableSpec.setTableName("myTable");
    tableSpec.setSchemaURI(schemaFile.toURI().toString());
    tableSpec.setTableConfigURI(tableConfigFile.toURI().toString());
    jobSpec.setTableSpec(tableSpec);

    ExecutionFrameworkSpec efSpec = new ExecutionFrameworkSpec();
    efSpec.setName("hadoop");
    efSpec.setSegmentGenerationJobRunnerClassName(HadoopSegmentGenerationJobRunner.class.getName());
    Map<String, String> extraConfigs = new HashMap<>();
    extraConfigs.put("stagingDir", stagingDir.toURI().toString());
    extraConfigs.put("dependencyJarDir", dependencyJarsDir.getAbsolutePath());
    efSpec.setExtraConfigs(extraConfigs);
    jobSpec.setExecutionFrameworkSpec(efSpec);

    PinotFSSpec pfsSpec = new PinotFSSpec();
    pfsSpec.setScheme("file");
    pfsSpec.setClassName(LocalPinotFS.class.getName());
    jobSpec.setPinotFSSpecs(Collections.singletonList(pfsSpec));

    jobSpec.setFailOnEmptySegment(true);

    System.setProperty(PluginManager.PLUGINS_DIR_PROPERTY_NAME, pluginsDir.getAbsolutePath());
    HadoopSegmentGenerationJobRunner jobRunner = new HadoopSegmentGenerationJobRunner(jobSpec);
    jobRunner.run();
    Assert.assertFalse(stagingDir.exists());

    // The output directory should still have the original file in it.
    File oldSegmentFile = new File(outputDir, otherFilename);
    Assert.assertTrue(oldSegmentFile.exists());

    // The output directory should have the original file in it (since we aren't overwriting)
    File newSegmentFile = new File(outputDir, outputFilename);
    Assert.assertTrue(newSegmentFile.exists());
    Assert.assertTrue(newSegmentFile.isFile());
    Assert.assertTrue(newSegmentFile.length() == 0);

    // Now run again, but this time with overwriting of output files, and confirm we got a valid segment file.
    jobSpec.setOverwriteOutput(true);
    jobRunner = new HadoopSegmentGenerationJobRunner(jobSpec);
    jobRunner.run();
    Assert.assertFalse(stagingDir.exists());

    // The original file should still be there.
    Assert.assertTrue(oldSegmentFile.exists());

    Assert.assertTrue(newSegmentFile.exists());
    Assert.assertTrue(newSegmentFile.isFile());
    Assert.assertTrue(newSegmentFile.length() > 0);

    // FUTURE - validate contents of file?
  }