class AggregationService {

    public AggregationResult aggregate(AggregationParameters parameters, DataSet dataset) {

        // check the parameters
        if (parameters.getOperations().isEmpty() || parameters.getGroupBy().isEmpty()) {
            throw new TDPException(CommonErrorCodes.BAD_AGGREGATION_PARAMETERS);
        }

        AggregationResult result = new AggregationResult(parameters.getOperations().get(0).getOperator());

        // get the aggregator
        Aggregator aggregator = factory.get(parameters);

        // Build optional filter
        final DataSetMetadata metadata = dataset.getMetadata();
        final RowMetadata rowMetadata = metadata != null ? metadata.getRowMetadata() : new RowMetadata();
        final Predicate<DataSetRow> filter = filterService.build(parameters.getFilter(), rowMetadata);

        // process the dataset
        dataset.getRecords().filter(filter).forEach(row -> aggregator.accept(row, result));

        // Normalize result (perform clean / optimization now that all input was processed).
        aggregator.normalize(result);

        return result;
    }

}