count.h

#ifndef CPARK_COUNT_H
#define CPARK_COUNT_H
 
#include <future>
#include <vector>
#include <numeric>
 
#include "base_rdd.h"
#include "utils.h"
 
namespace cpark {
 
class Count {
public:
  explicit Count() = default;
 
  template <concepts::Rdd R>
  unsigned long long operator()(const R& rdd) const {
    // Parallel compute the result.
    std::vector<std::future<unsigned long long>> futures{};
    for (const concepts::Split auto& split : rdd)
      futures.emplace_back(std::async([this, &split]() {
         return static_cast<unsigned long long>(split.size());
      }));
 
    auto results =
        std::ranges::subrange(futures) |
        std::views::transform([](std::future<unsigned long long>& fut) { return fut.get(); });
 
    return std::accumulate(results.begin(), results.end(), 0);
  }
};
 
template <concepts::Rdd R>
auto operator|(const R& r, const Count& count) {
  return count(r);
}
 
 
}  // namespace cpark
 
#endif  //CPARK_COUNT_H