class StatsService {

  @Transactional
  public GetFeatureStatisticsResponse getFeatureStatistics(GetFeatureStatisticsRequest request)
      throws IOException {

    // Validate the request
    validateRequest(request);

    // Get the stats retriever for the store requested
    StatisticsRetriever statisticsRetriever = getStatisticsRetriever(request.getStore());

    // 1. Retrieve the feature set spec from the db
    FeatureSet featureSet = getFeatureSet(request.getFeatureSetId());
    if (featureSet == null) {
      throw new IllegalArgumentException(
          String.format(
              "Illegal request. Unable to find feature set %s", request.getFeatureSetId()));
    }

    // 2. Filter out the features requested by the user. If none are provided,
    // use all features in the feature set.
    List<String> features = request.getFeaturesList();
    if (features.size() == 0) {
      features =
          featureSet.getFeatures().stream()
              .filter(feature -> !feature.isArchived())
              .map(Feature::getName)
              .collect(Collectors.toList());
    }

    // 3. Retrieve the statistics from the StatsRetriever.
    List<List<FeatureNameStatistics>> featureNameStatisticsList = new ArrayList<>();
    if (request.getIngestionIdsCount() == 0) {
      Timestamp endDate = request.getEndDate();
      Timestamp startDate = request.getStartDate();
      // If no dataset provided, retrieve by date

      long timestamp = startDate.getSeconds();
      while (timestamp < endDate.getSeconds()) {
        List<FeatureNameStatistics> featureNameStatistics =
            getFeatureNameStatisticsByDate(
                statisticsRetriever, featureSet, features, timestamp, request.getForceRefresh());
        if (featureNameStatistics.size() != 0) {
          featureNameStatisticsList.add(featureNameStatistics);
        }
        timestamp += 86400; // advance by a day
      }
      if (featureNameStatisticsList.size() == 0) {
        DateTimeFormatter fmt = DateTimeFormat.forPattern("yyyy-MM-dd");
        DateTime startDateTime = new DateTime(startDate.getSeconds() * 1000, DateTimeZone.UTC);
        DateTime endDateTime = new DateTime(endDate.getSeconds() * 1000, DateTimeZone.UTC);
        throw new RetrievalException(
            String.format(
                "Unable to find any data over provided dates [%s, %s)",
                fmt.print(startDateTime), fmt.print(endDateTime)));
      }
    } else {
      // else, retrieve by dataset
      for (String datasetId : request.getIngestionIdsList()) {
        List<FeatureNameStatistics> featureNameStatistics =
            getFeatureNameStatisticsByDataset(
                statisticsRetriever, featureSet, features, datasetId, request.getForceRefresh());
        if (featureNameStatistics.size() != 0) {
          featureNameStatisticsList.add(featureNameStatistics);
        }
      }
      if (featureNameStatisticsList.size() == 0) {
        throw new RetrievalException(
            String.format(
                "Unable to find any data over provided data sets %s",
                request.getIngestionIdsList()));
      }
    }

    // Merge statistics values across days/datasets
    List<FeatureNameStatistics> featureNameStatistics = mergeStatistics(featureNameStatisticsList);
    long totalCount = getTotalCount(featureNameStatistics.get(0));
    return GetFeatureStatisticsResponse.newBuilder()
        .setDatasetFeatureStatisticsList(
            DatasetFeatureStatisticsList.newBuilder()
                .addDatasets(
                    DatasetFeatureStatistics.newBuilder()
                        .setNumExamples(totalCount)
                        .addAllFeatures(featureNameStatistics)))
        .build();
  }

  @Autowired public  StatsService(
      StoreRepository storeRepository,
      FeatureStatisticsRepository featureStatisticsRepository,
      FeatureSetRepository featureSetRepository);

}