quantiles_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 _QUANTILES_SKETCH_HPP_
#define _QUANTILES_SKETCH_HPP_
 
#include <functional>
#include <memory>
#include <vector>
 
#include "quantiles_sorted_view.hpp"
#include "common_defs.hpp"
#include "serde.hpp"
#include "optional.hpp"
 
namespace datasketches {
 
namespace quantiles_constants {
  const uint16_t DEFAULT_K = 128;
  const uint16_t MIN_K = 2;
  const uint16_t MAX_K = 1 << 15;
}
 
template <typename T,
          typename Comparator = std::less<T>, // strict weak ordering function (see C++ named requirements: Compare)
          typename Allocator = std::allocator<T>>
class quantiles_sketch {
public:
  using value_type = T;
  using allocator_type = Allocator;
  using comparator = Comparator;
  using quantile_return_type = typename quantiles_sorted_view<T, Comparator, Allocator>::quantile_return_type;
  using vector_double = typename quantiles_sorted_view<T, Comparator, Allocator>::vector_double;
 
  explicit quantiles_sketch(uint16_t k = quantiles_constants::DEFAULT_K,
      const Comparator& comparator = Comparator(), const Allocator& allocator = Allocator());
 
  quantiles_sketch(const quantiles_sketch& other);
 
  quantiles_sketch(quantiles_sketch&& other) noexcept;
 
  ~quantiles_sketch();
 
  quantiles_sketch& operator=(const quantiles_sketch& other);
 
  quantiles_sketch& operator=(quantiles_sketch&& other) noexcept;
 
  template<typename From, typename FC, typename FA>
  explicit quantiles_sketch(const quantiles_sketch<From, FC, FA>& other,
      const Comparator& comparator = Comparator(), const Allocator& allocator = Allocator());
 
  template<typename FwdT>
  void update(FwdT&& item);
 
  template<typename FwdSk>
  void merge(FwdSk&& other);
 
  bool is_empty() const;
 
  uint16_t get_k() const;
 
  uint64_t get_n() const;
 
  uint32_t get_num_retained() const;
 
  bool is_estimation_mode() const;
 
  const T& get_min_item() const;
 
  const T& get_max_item() const;
 
  Comparator get_comparator() const;
 
  allocator_type get_allocator() const;
 
  quantile_return_type get_quantile(double rank, bool inclusive = true) const;
 
  double get_rank(const T& item, bool inclusive = true) const;
 
  vector_double get_PMF(const T* split_points, uint32_t size, bool inclusive = true) const;
 
  vector_double get_CDF(const T* split_points, uint32_t size, bool inclusive = true) const;
 
  template<typename SerDe = serde<T>, typename TT = T, typename std::enable_if<std::is_arithmetic<TT>::value, int>::type = 0>
  size_t get_serialized_size_bytes(const SerDe& sd = SerDe()) const;
 
  template<typename SerDe = serde<T>, typename TT = T, typename std::enable_if<!std::is_arithmetic<TT>::value, int>::type = 0>
  size_t get_serialized_size_bytes(const SerDe& sd = SerDe()) const;
 
  template<typename SerDe = serde<T>>
  void serialize(std::ostream& os, const SerDe& sd = SerDe()) const;
 
  // This is a convenience alias for users
  // The type returned by the following serialize method
  using vector_bytes = std::vector<uint8_t, typename std::allocator_traits<Allocator>::template rebind_alloc<uint8_t>>;
 
  template<typename SerDe = serde<T>>
  vector_bytes serialize(unsigned header_size_bytes = 0, const SerDe& sd = SerDe()) const;
 
  template<typename SerDe = serde<T>>
  static quantiles_sketch deserialize(std::istream& is, const SerDe& sd = SerDe(),
      const Comparator& comparator = Comparator(), const Allocator& allocator = Allocator());
 
  template<typename SerDe = serde<T>>
  static quantiles_sketch deserialize(const void* bytes, size_t size, const SerDe& sd = SerDe(),
      const Comparator& comparator = Comparator(), const Allocator& allocator = Allocator());
 
  double get_normalized_rank_error(bool is_pmf) const;
 
  static double get_normalized_rank_error(uint16_t k, bool is_pmf);
 
  string<Allocator> to_string(bool print_levels = false, bool print_items = false) const;
 
  class const_iterator;
 
  const_iterator begin() const;
 
  const_iterator end() const;
 
  quantiles_sorted_view<T, Comparator, Allocator> get_sorted_view() const;
 
private:
  using Level = std::vector<T, Allocator>;
  using VectorLevels = std::vector<Level, typename std::allocator_traits<Allocator>::template rebind_alloc<Level>>;
 
  /* Serialized sketch layout:
   * Long || Start Byte Addr:
   * Addr:
   *      ||       0        |    1   |    2   |    3   |    4   |    5   |    6   |    7   |
   *  0   || Preamble_Longs | SerVer | FamID  |  Flags |----- K ---------|---- unused -----|
   *
   *      ||       8        |    9   |   10   |   11   |   12   |   13   |   14   |   15   |
   *  1   ||---------------------------Items Seen Count (N)--------------------------------|
   *
   * Long 3 is the start of data, beginning with serialized min and max item, followed by
   * the sketch data buffers.
   */
 
  static const size_t EMPTY_SIZE_BYTES = 8;
  static const uint8_t SERIAL_VERSION_1 = 1;
  static const uint8_t SERIAL_VERSION_2 = 2;
  static const uint8_t SERIAL_VERSION = 3;
  static const uint8_t FAMILY = 8;
 
  enum flags { RESERVED0, RESERVED1, IS_EMPTY, IS_COMPACT, IS_SORTED };
 
  static const uint8_t PREAMBLE_LONGS_SHORT = 1; // for empty
  static const uint8_t PREAMBLE_LONGS_FULL = 2;
  static const size_t DATA_START = 16;
 
  Comparator comparator_;
  Allocator allocator_;
  bool is_base_buffer_sorted_;
  uint16_t k_;
  uint64_t n_;
  uint64_t bit_pattern_;
  Level base_buffer_;
  VectorLevels levels_;
  optional<T> min_item_;
  optional<T> max_item_;
  mutable quantiles_sorted_view<T, Comparator, Allocator>* sorted_view_;
 
  void setup_sorted_view() const; // modifies mutable state
  void reset_sorted_view();
 
  // for deserialization
  class items_deleter;
  quantiles_sketch(uint16_t k, uint64_t n, uint64_t bit_pattern,
      Level&& base_buffer, VectorLevels&& levels,
      optional<T>&& min_item, optional<T>&& max_item,
      bool is_sorted, const Comparator& comparator = Comparator(), const Allocator& allocator = Allocator());
 
  void grow_base_buffer();
  void process_full_base_buffer();
 
  // returns true if size adjusted, else false
  bool grow_levels_if_needed();
 
  // buffers should be pre-sized to target capacity as appropriate
  template<typename FwdV>
  static void in_place_propagate_carry(uint8_t starting_level, FwdV&& buf_size_k,
                                       Level& buf_size_2k, bool apply_as_update,
                                       quantiles_sketch& sketch);
  static void zip_buffer(Level& buf_in, Level& buf_out);
  static void merge_two_size_k_buffers(Level& arr_in_1, Level& arr_in_2, Level& arr_out, const Comparator& comparator);
 
  template<typename SerDe>
  static Level deserialize_array(std::istream& is, uint32_t num_items, uint32_t capcacity, const SerDe& serde, const Allocator& allocator);
  
  template<typename SerDe>
  static std::pair<Level, size_t> deserialize_array(const void* bytes, size_t size, uint32_t num_items, uint32_t capcacity, const SerDe& serde, const Allocator& allocator);
 
  static void check_k(uint16_t k);
  static void check_serial_version(uint8_t serial_version);
  static void check_header_validity(uint8_t preamble_longs, uint8_t flags_byte, uint8_t serial_version);
  static void check_family_id(uint8_t family_id);
 
  static uint32_t compute_retained_items(uint16_t k, uint64_t n);
  static uint32_t compute_base_buffer_items(uint16_t k, uint64_t n);
  static uint64_t compute_bit_pattern(uint16_t k, uint64_t n);
  static uint32_t count_valid_levels(uint64_t bit_pattern);
  static uint8_t compute_levels_needed(uint16_t k, uint64_t n);
 
  template<typename FwdSk>
  static void standard_merge(quantiles_sketch& tgt, FwdSk&& src);
 
  template<typename FwdSk>
  static void downsampling_merge(quantiles_sketch& tgt, FwdSk&& src);
 
  template<typename FwdV>
  static void zip_buffer_with_stride(FwdV&& buf_in, Level& buf_out, uint16_t stride);
 
  static uint8_t lowest_zero_bit_starting_at(uint64_t bits, uint8_t starting_bit);
 
  template<typename TT = T, typename std::enable_if<std::is_floating_point<TT>::value, int>::type = 0>
  static inline bool check_update_item(TT item) {
    return !std::isnan(item);
  }
 
  template<typename TT = T, typename std::enable_if<!std::is_floating_point<TT>::value, int>::type = 0>
  static inline bool check_update_item(TT) {
    return true;
  }
 
  // for type converting constructor
  template<typename From, typename FC, typename FA> friend class quantiles_sketch;
};
 
 
template<typename T, typename C, typename A>
class quantiles_sketch<T, C, A>::const_iterator {
public:
  using iterator_category = std::input_iterator_tag;
  using value_type = std::pair<const T&, const uint64_t>;
  using difference_type = void;
  using pointer = const return_value_holder<value_type>;
  using reference = const value_type;
 
  const_iterator& operator++();
  const_iterator& operator++(int);
  bool operator==(const const_iterator& other) const;
  bool operator!=(const const_iterator& other) const;
  reference operator*() const;
  pointer operator->() const;
private:
  friend class quantiles_sketch<T, C, A>;
  using Level = std::vector<T, A>;
  using AllocLevel = typename std::allocator_traits<A>::template rebind_alloc<Level>;
  Level base_buffer_;
  std::vector<Level, AllocLevel> levels_;
  int level_;
  uint32_t index_;
  uint32_t bb_count_;
  uint64_t bit_pattern_;
  uint64_t weight_;
  uint16_t k_;
  const_iterator(const Level& base_buffer, const std::vector<Level, AllocLevel>& levels, uint16_t k, uint64_t n, bool is_end);
};
 
} /* namespace datasketches */
 
#include "quantiles_sketch_impl.hpp"
 
#endif // _QUANTILES_SKETCH_HPP_