@Override
  public int run() throws IOException {
    Preconditions.checkArgument(sourceFiles != null && !sourceFiles.isEmpty(), "Missing file name");

    // Ensure all source files have the columns specified first
    Map<String, Schema> schemas = new HashMap<>();
    for (String sourceFile : sourceFiles) {
      Schema schema = getAvroSchema(sourceFile);
      schemas.put(sourceFile, Expressions.filterSchema(schema, columns));
    }

    long totalStartTime = System.currentTimeMillis();
    long totalCount = 0;
    for (String sourceFile : sourceFiles) {
      long startTime = System.currentTimeMillis();
      Iterable<Object> reader = openDataFile(sourceFile, schemas.get(sourceFile));
      boolean threw = true;
      long count = 0;
      try {
        for (Object record : reader) {
          count += 1;
        }
        threw = false;
      } catch (RuntimeException e) {
        throw new RuntimeException("Failed on record " + count + " in " + sourceFile, e);
      } finally {
        if (reader instanceof Closeable) {
          Closeables.close((Closeable) reader, threw);
        }
      }
      totalCount += count;
      if (1 < sourceFiles.size()) {
        long endTime = System.currentTimeMillis();
        console.info("Scanned " + count + " records from " + sourceFile + " in "
            + (endTime - startTime) / 1000.0 + " s");
      }
    }
    long totalEndTime = System.currentTimeMillis();
    console.info("Scanned " + totalCount + " records from " + sourceFiles.size() + " file(s)");
    console.info("Time: " + (totalEndTime - totalStartTime) / 1000.0 + " s");
    return 0;
  }
@Test
  public void testScanCommandWithMultipleSourceFiles() throws IOException {
    File file = parquetFile();
    ScanCommand command = new ScanCommand(createLogger());
    command.sourceFiles = Arrays.asList(file.getAbsolutePath(), file.getAbsolutePath());
    command.setConf(new Configuration());
    Assert.assertEquals(0, command.run());
  }