mergeNode.projections(Arrays.asList(groupProjection, topNProjection));
mergeNode.outputTypes(Arrays.<DataType>asList(DataTypes.STRING, DataTypes.LONG));
Streamer<?>[] mapperOutputStreamer = new Streamer[] {
new AggregationStateStreamer(countAggregation),
DataTypes.STRING.streamer()
};
NoopListener noopListener = new NoopListener();