tuple_sketch.hpp

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
 
#ifndef TUPLE_SKETCH_HPP_
#define TUPLE_SKETCH_HPP_
 
#include <string>
 
#include "serde.hpp"
#include "theta_update_sketch_base.hpp"
 
namespace datasketches {
 
// forward declarations
template<typename S, typename A> class tuple_sketch;
template<typename S, typename U, typename P, typename A> class update_tuple_sketch;
template<typename S, typename A> class compact_tuple_sketch;
template<typename A> class theta_sketch_alloc;
 
template<typename K, typename V>
struct pair_extract_key {
  K& operator()(std::pair<K, V>& entry) const {
    return entry.first;
  }
  const K& operator()(const std::pair<K, V>& entry) const {
    return entry.first;
  }
};

template<
  typename Summary,
  typename Allocator = std::allocator<Summary>
>
class tuple_sketch {
public:
  using Entry = std::pair<uint64_t, Summary>;
  using ExtractKey = pair_extract_key<uint64_t, Summary>;
  using iterator = theta_iterator<Entry, ExtractKey>;
  using const_iterator = theta_const_iterator<Entry, ExtractKey>;
 
  virtual ~tuple_sketch() = default;

  virtual Allocator get_allocator() const = 0;

  virtual bool is_empty() const = 0;

  double get_estimate() const;

  double get_lower_bound(uint8_t num_std_devs, uint32_t num_subset_entries) const ;

  double get_lower_bound(uint8_t num_std_devs) const;
 

  double get_upper_bound(uint8_t num_std_devs, uint32_t num_subset_entries) const ;
 

  double get_upper_bound(uint8_t num_std_devs) const;

  bool is_estimation_mode() const;

  double get_theta() const;

  virtual uint64_t get_theta64() const = 0;

  virtual uint32_t get_num_retained() const = 0;

  virtual uint16_t get_seed_hash() const = 0;

  virtual bool is_ordered() const = 0;

  string<Allocator> to_string(bool print_items = false) const;

  virtual iterator begin() = 0;

  virtual iterator end() = 0;

  virtual const_iterator begin() const = 0;

  virtual const_iterator end() const = 0;
 
protected:
  virtual void print_specifics(std::ostringstream& os) const = 0;
 
  static uint16_t get_seed_hash(uint64_t seed);
 
  static void check_sketch_type(uint8_t actual, uint8_t expected);
  static void check_serial_version(uint8_t actual, uint8_t expected);
  static void check_seed_hash(uint16_t actual, uint16_t expected);
};
 
// update sketch
 
// for types with defined default constructor and + operation
template<typename Summary, typename Update>
struct default_tuple_update_policy {
  Summary create() const {
    return Summary();
  }
  void update(Summary& summary, const Update& update) const {
    summary += update;
  }
};

template<
  typename Summary,
  typename Update = Summary,
  typename Policy = default_tuple_update_policy<Summary, Update>,
  typename Allocator = std::allocator<Summary>
>
class update_tuple_sketch: public tuple_sketch<Summary, Allocator> {
public:
  using Base = tuple_sketch<Summary, Allocator>;
  using Entry = typename Base::Entry;
  using ExtractKey = typename Base::ExtractKey;
  using iterator = typename Base::iterator;
  using const_iterator = typename Base::const_iterator;
  using AllocEntry = typename std::allocator_traits<Allocator>::template rebind_alloc<Entry>;
  using tuple_map = theta_update_sketch_base<Entry, ExtractKey, AllocEntry>;
  using resize_factor = typename tuple_map::resize_factor;
 
  // No constructor here. Use builder instead.
  class builder;
 
  update_tuple_sketch(const update_tuple_sketch&) = default;
  update_tuple_sketch(update_tuple_sketch&&) noexcept = default;
  virtual ~update_tuple_sketch() = default;
  update_tuple_sketch& operator=(const update_tuple_sketch&) = default;
  update_tuple_sketch& operator=(update_tuple_sketch&&) = default;
 
  virtual Allocator get_allocator() const;
  virtual bool is_empty() const;
  virtual bool is_ordered() const;
  virtual uint64_t get_theta64() const;
  virtual uint32_t get_num_retained() const;
  virtual uint16_t get_seed_hash() const;

  uint8_t get_lg_k() const;

  resize_factor get_rf() const;

  template<typename FwdUpdate>
  inline void update(const std::string& key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(uint64_t key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(int64_t key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(uint32_t key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(int32_t key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(uint16_t key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(int16_t key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(uint8_t key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(int8_t key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(double key, FwdUpdate&& value);

  template<typename FwdUpdate>
  inline void update(float key, FwdUpdate&& value);

  template<typename FwdUpdate>
  void update(const void* key, size_t length, FwdUpdate&& value);

  void trim();

  void reset();

  compact_tuple_sketch<Summary, Allocator> compact(bool ordered = true) const;

  template<typename Predicate>
  compact_tuple_sketch<Summary, Allocator> filter(const Predicate& predicate) const;
 
  virtual iterator begin();
  virtual iterator end();
  virtual const_iterator begin() const;
  virtual const_iterator end() const;
 
protected:
  Policy policy_;
  tuple_map map_;
 
  // for builder
  update_tuple_sketch(uint8_t lg_cur_size, uint8_t lg_nom_size, resize_factor rf, float p, uint64_t theta, uint64_t seed, const Policy& policy, const Allocator& allocator);
 
  virtual void print_specifics(std::ostringstream& os) const;
};

template<
  typename Summary,
  typename Allocator = std::allocator<Summary>
>
class compact_tuple_sketch: public tuple_sketch<Summary, Allocator> {
public:
  using Base = tuple_sketch<Summary, Allocator>;
  using Entry = typename Base::Entry;
  using ExtractKey = typename Base::ExtractKey;
  using iterator = typename Base::iterator;
  using const_iterator = typename Base::const_iterator;
  using AllocEntry = typename std::allocator_traits<Allocator>::template rebind_alloc<Entry>;
  using AllocU64 = typename std::allocator_traits<Allocator>::template rebind_alloc<uint64_t>;
  using AllocBytes = typename std::allocator_traits<Allocator>::template rebind_alloc<uint8_t>;
  using vector_bytes = std::vector<uint8_t, AllocBytes>;
  using comparator = compare_by_key<ExtractKey>;
 
  static const uint8_t SERIAL_VERSION_LEGACY = 1;
  static const uint8_t SERIAL_VERSION = 3;
  static const uint8_t SKETCH_FAMILY = 9;
  static const uint8_t SKETCH_TYPE = 1;
  static const uint8_t SKETCH_TYPE_LEGACY = 5;
  enum flags { IS_BIG_ENDIAN, IS_READ_ONLY, IS_EMPTY, IS_COMPACT, IS_ORDERED };
 
  // Instances of this type can be obtained:
  // - by compacting an update_tuple_sketch
  // - as a result of a set operation
  // - by deserializing a previously serialized compact sketch

  compact_tuple_sketch(const Base& other, bool ordered);

  compact_tuple_sketch(const compact_tuple_sketch& other) = default;

  compact_tuple_sketch(compact_tuple_sketch&&) noexcept;
 
  virtual ~compact_tuple_sketch() = default;

  compact_tuple_sketch& operator=(const compact_tuple_sketch& other) = default;

  compact_tuple_sketch& operator=(compact_tuple_sketch&& other) = default;

  compact_tuple_sketch(const theta_sketch_alloc<AllocU64>& other, const Summary& summary, bool ordered = true);
 
  virtual Allocator get_allocator() const;
  virtual bool is_empty() const;
  virtual bool is_ordered() const;
  virtual uint64_t get_theta64() const;
  virtual uint32_t get_num_retained() const;
  virtual uint16_t get_seed_hash() const;

  template<typename Predicate>
  compact_tuple_sketch filter(const Predicate& predicate) const;

  template<typename Sketch, typename Predicate>
  static compact_tuple_sketch filter(const Sketch& sketch, const Predicate& predicate);

  template<typename SerDe = serde<Summary>>
  void serialize(std::ostream& os, const SerDe& sd = SerDe()) const;

  template<typename SerDe = serde<Summary>>
  vector_bytes serialize(unsigned header_size_bytes = 0, const SerDe& sd = SerDe()) const;
 
  virtual iterator begin();
  virtual iterator end();
  virtual const_iterator begin() const;
  virtual const_iterator end() const;

  template<typename SerDe = serde<Summary>>
  static compact_tuple_sketch deserialize(std::istream& is, uint64_t seed = DEFAULT_SEED,
      const SerDe& sd = SerDe(), const Allocator& allocator = Allocator());

  template<typename SerDe = serde<Summary>>
  static compact_tuple_sketch deserialize(const void* bytes, size_t size, uint64_t seed = DEFAULT_SEED,
      const SerDe& sd = SerDe(), const Allocator& allocator = Allocator());
 
protected:
  bool is_empty_;
  bool is_ordered_;
  uint16_t seed_hash_;
  uint64_t theta_;
  std::vector<Entry, AllocEntry> entries_;

  template<typename SerDe, typename SS = Summary, typename std::enable_if<std::is_arithmetic<SS>::value, int>::type = 0>
  size_t get_serialized_size_summaries_bytes(const SerDe& sd) const;

  template<typename SerDe, typename SS = Summary, typename std::enable_if<!std::is_arithmetic<SS>::value, int>::type = 0>
  size_t get_serialized_size_summaries_bytes(const SerDe& sd) const;
 
  // for deserialize
  class deleter_of_summaries {
  public:
    deleter_of_summaries(uint32_t num, bool destroy, const Allocator& allocator):
      allocator_(allocator), num_(num), destroy_(destroy) {}
    void set_destroy(bool destroy) { destroy_ = destroy; }
    void operator() (Summary* ptr) {
      if (ptr != nullptr) {
        if (destroy_) {
          for (uint32_t i = 0; i < num_; ++i) ptr[i].~Summary();
        }
        allocator_.deallocate(ptr, num_);
      }
    }
  private:
    Allocator allocator_;
    uint32_t num_;
    bool destroy_;
  };
 
  virtual void print_specifics(std::ostringstream& os) const;
 
  template<typename E, typename EK, typename P, typename S, typename CS, typename A> friend class theta_union_base;
  template<typename E, typename EK, typename P, typename S, typename CS, typename A> friend class theta_intersection_base;
  template<typename E, typename EK, typename CS, typename A> friend class theta_set_difference_base;
  compact_tuple_sketch(bool is_empty, bool is_ordered, uint16_t seed_hash, uint64_t theta, std::vector<Entry, AllocEntry>&& entries);
};

template<typename Derived, typename Policy, typename Allocator>
class tuple_base_builder: public theta_base_builder<Derived, Allocator> {
public:
  tuple_base_builder(const Policy& policy, const Allocator& allocator);
 
protected:
  Policy policy_;
};

template<typename S, typename U, typename P, typename A>
class update_tuple_sketch<S, U, P, A>::builder: public tuple_base_builder<builder, P, A> {
public:
  builder(const P& policy = P(), const A& allocator = A());

  update_tuple_sketch<S, U, P, A> build() const;
};
 
} /* namespace datasketches */
 
#include "tuple_sketch_impl.hpp"
 
#endif