@Override
  public ResourceStatistics getStatistics(String location, Job job) throws IOException {
    if (LOG.isDebugEnabled()) {
      String jobToString = String.format("job[id=%s, name=%s]", job.getJobID(), job.getJobName());
      LOG.debug("LoadMetadata.getStatistics({}, {})", location, jobToString);
    }
    /* We need to call setInput since setLocation is not
    guaranteed to be called before this */
    setInput(location, job);
    long length = 0;
    try {
      for (InputSplit split : getParquetInputFormat().getSplits(job)) {
        length += split.getLength();
      }
    } catch (InterruptedException e) {
      LOG.warn("Interrupted", e);
      Thread.currentThread().interrupt();
      return null;
    }
    ResourceStatistics stats = new ResourceStatistics();
    // TODO use pig-0.12 setBytes api when its available
    stats.setmBytes(length / 1024 / 1024);
    return stats;
  }
@Test
  public void testPredicatePushdown() throws Exception {
    Configuration conf = new Configuration();
    conf.setBoolean(ParquetLoader.ENABLE_PREDICATE_FILTER_PUSHDOWN, true);

    PigServer pigServer = new PigServer(ExecType.LOCAL, conf);
    pigServer.setValidateEachStatement(true);

    String out = "target/out";
    String out2 = "target/out2";
    int rows = 10;
    Data data = Storage.resetData(pigServer);
    List<Tuple> list = new ArrayList<Tuple>();
    for (int i = 0; i < rows; i++) {
      list.add(Storage.tuple(i, i * 1.0, i * 2L, "v" + i));
    }
    data.set("in", "c1:int, c2:double, c3:long, c4:chararray", list);
    pigServer.setBatchOn();
    pigServer.registerQuery("A = LOAD 'in' USING mock.Storage();");
    pigServer.deleteFile(out);
    pigServer.registerQuery("Store A into '" + out + "' using " + ParquetStorer.class.getName() + "();");
    pigServer.executeBatch();

    pigServer.deleteFile(out2);
    pigServer.registerQuery("B = LOAD '" + out + "' using " + ParquetLoader.class.getName()
        + "('c1:int, c2:double, c3:long, c4:chararray');");
    pigServer.registerQuery("C = FILTER B by c1 == 1 or c1 == 5;");
    pigServer.registerQuery("STORE C into '" + out2 + "' using mock.Storage();");
    List<ExecJob> jobs = pigServer.executeBatch();

    long recordsRead = jobs.get(0).getStatistics().getInputStats().get(0).getNumberRecords();

    assertEquals(2, recordsRead);
  }