@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 source : sourceFiles) {
      Schema schema = getAvroSchema(source);
      schemas.put(source, Expressions.filterSchema(schema, columns));
    }

    for (String source : sourceFiles) {
      Schema projection = schemas.get(source);
      Iterable<Object> reader = openDataFile(source, projection);
      boolean threw = true;
      long count = 0;
      try {
        for (Object record : reader) {
          if (numRecords > 0 && count >= numRecords) {
            break;
          }
          if (columns == null || columns.size() != 1) {
            console.info(String.valueOf(record));
          } else {
            console.info(String.valueOf(select(projection, record, columns.get(0))));
          }
          count += 1;
        }
        threw = false;
      } catch (RuntimeException e) {
        throw new RuntimeException("Failed on record " + count + " in file " + source, e);
      } finally {
        if (reader instanceof Closeable) {
          Closeables.close((Closeable) reader, threw);
        }
      }
    }

    return 0;
  }
@Test(expected = IllegalArgumentException.class)
  public void testCatCommandWithInvalidColumn() throws IOException {
    File file = parquetFile();
    CatCommand command = new CatCommand(createLogger(), 0);
    command.sourceFiles = Arrays.asList(file.getAbsolutePath());
    command.columns = Arrays.asList("invalid_field");
    command.setConf(new Configuration());
    command.run();
  }