Count extends DataSetAnalyticBase<T, Long> { @Override public Count<T> run(DataSet<T> input) throws Exception { super.run(input); countHelper = new CountHelper<>(); input .output(countHelper) .name("Count"); return this; }  @Override Count<T> run(DataSet<T> input); @Override Long getResult();  }
@Test public void testEmptyList() throws Exception { DataSet<Long> dataset = env.fromCollection(Collections.emptyList(), TypeInformation.of(new TypeHint<Long>(){})); long count = new Count<Long>().run(dataset).execute(); assertEquals(0, count); }