kll_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 KLL_SKETCH_HPP_
#define KLL_SKETCH_HPP_
 
#include <memory>
#include <vector>
 
#include "common_defs.hpp"
#include "serde.hpp"
#include "quantiles_sorted_view.hpp"
#include "optional.hpp"
 
namespace datasketches {
 
namespace kll_constants {
  const uint16_t DEFAULT_K = 200;
  const uint8_t DEFAULT_M = 8;
  const uint16_t MIN_K = DEFAULT_M;
  const uint16_t MAX_K = (1 << 16) - 1;
}
 
template <
  typename T,
  typename C = std::less<T>, // strict weak ordering function (see C++ named requirements: Compare)
  typename A = std::allocator<T>
>
class kll_sketch {
  public:
    using value_type = T;
    using comparator = C;
    using allocator_type = A;
    using vector_u32 = std::vector<uint32_t, typename std::allocator_traits<A>::template rebind_alloc<uint32_t>>;
    using vector_double = typename quantiles_sorted_view<T, C, A>::vector_double;
 
    using quantile_return_type = typename quantiles_sorted_view<T, C, A>::quantile_return_type;
 
    explicit kll_sketch(uint16_t k = kll_constants::DEFAULT_K, const C& comparator = C(), const A& allocator = A());
 
    kll_sketch(const kll_sketch& other);
 
    kll_sketch(kll_sketch&& other) noexcept;
 
 
    ~kll_sketch();
 
    kll_sketch& operator=(const kll_sketch& other);
 
    kll_sketch& operator=(kll_sketch&& other);
 
    /*
     * Type converting constructor.
     * @param other sketch of a different type
     * @param comparator instance of a Comparator
     * @param allocator instance of an Allocator
     */
    template<typename TT, typename CC, typename AA>
    explicit kll_sketch(const kll_sketch<TT, CC, AA>& other, const C& comparator = C(), const A& allocator = A());
 
    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;
 
    T get_min_item() const;
 
    T get_max_item() const;
 
    C get_comparator() const;
 
    A 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;
 
    double get_normalized_rank_error(bool pmf) const;
 
    template<typename TT = T, typename SerDe = serde<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 TT = T, typename SerDe = serde<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 TT = T, typename std::enable_if<std::is_arithmetic<TT>::value, int>::type = 0>
    static size_t get_max_serialized_size_bytes(uint16_t k, uint64_t n);
 
    template<typename TT = T, typename std::enable_if<!std::is_arithmetic<TT>::value, int>::type = 0>
    static size_t get_max_serialized_size_bytes(uint16_t k, uint64_t n, size_t max_item_size_bytes);
 
    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<A>::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 kll_sketch deserialize(std::istream& is, const SerDe& sd = SerDe(),
        const C& comparator = C(), const A& allocator = A());
 
    template<typename SerDe = serde<T>>
    static kll_sketch deserialize(const void* bytes, size_t size, const SerDe& sd = SerDe(),
        const C& comparator = C(), const A& allocator = A());
 
    /*
     * Gets the normalized rank error given k and pmf.
     * k - the configuration parameter
     * pmf - if true, returns the "double-sided" normalized rank error for the get_PMF() function.
     * Otherwise, it is the "single-sided" normalized rank error for all the other queries.
     * Constants were derived as the best fit to 99 percentile empirically measured max error in thousands of trials
     */
    static double get_normalized_rank_error(uint16_t k, bool pmf);
 
    string<A> 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, C, A> get_sorted_view() const;
 
  private:
    /* Serialized sketch layout:
     *  Addr:
     *      ||    7    |   6   |    5   |    4   |    3   |    2    |    1   |      0       |
     *  0   || unused  |   M   |--------K--------|  Flags |  FamID  | SerVer | PreambleInts |
     *      ||   15    |   14  |   13   |   12   |   11   |   10    |    9   |      8       |
     *  1   ||-----------------------------------N------------------------------------------|
     *      ||   23    |   22  |   21   |   20   |   19   |    18   |   17   |      16      |
     *  2   ||---------------data----------------|-unused-|numLevels|-------min K-----------|
     */
 
    static const size_t EMPTY_SIZE_BYTES = 8;
    static const size_t DATA_START_SINGLE_ITEM = 8;
    static const size_t DATA_START = 20;
 
    static const uint8_t SERIAL_VERSION_1 = 1;
    static const uint8_t SERIAL_VERSION_2 = 2;
    static const uint8_t FAMILY = 15;
 
    enum flags { IS_EMPTY, IS_LEVEL_ZERO_SORTED, IS_SINGLE_ITEM };
 
    static const uint8_t PREAMBLE_INTS_SHORT = 2; // for empty and single item
    static const uint8_t PREAMBLE_INTS_FULL = 5;
 
    C comparator_;
    A allocator_;
    uint16_t k_;
    uint8_t m_; // minimum buffer "width"
    uint16_t min_k_; // for error estimation after merging with different k
    uint8_t num_levels_;
    bool is_level_zero_sorted_;
    uint64_t n_;
    vector_u32 levels_;
    T* items_;
    uint32_t items_size_;
    optional<T> min_item_;
    optional<T> max_item_;
    mutable quantiles_sorted_view<T, C, A>* sorted_view_;
 
    // for deserialization
    class items_deleter;
    kll_sketch(uint16_t k, uint16_t min_k, uint64_t n, uint8_t num_levels, vector_u32&& levels,
        std::unique_ptr<T, items_deleter> items, uint32_t items_size, optional<T>&& min_item,
        optional<T>&& max_item, bool is_level_zero_sorted, const C& comparator);
 
    // common update code
    inline void update_min_max(const T& item);
    inline uint32_t internal_update();
 
    // The following code is only valid in the special case of exactly reaching capacity while updating.
    // It cannot be used while merging, while reducing k, or anything else.
    void compress_while_updating(void);
 
    uint8_t find_level_to_compact() const;
    void add_empty_top_level_to_completely_full_sketch();
    void sort_level_zero();
 
    template<typename O> void merge_higher_levels(O&& other, uint64_t final_n);
 
    template<typename FwdSk>
    void populate_work_arrays(FwdSk&& other, T* workbuf, uint32_t* worklevels, uint8_t provisional_num_levels);
 
    void assert_correct_total_weight() const;
    uint32_t safe_level_size(uint8_t level) const;
    uint32_t get_num_retained_above_level_zero() const;
 
    static void check_m(uint8_t m);
    static void check_preamble_ints(uint8_t preamble_ints, uint8_t flags_byte);
    static void check_serial_version(uint8_t serial_version);
    static void check_family_id(uint8_t family_id);
 
    void check_sorting() const;
 
    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 TT, typename CC, typename AA> friend class kll_sketch;
 
    void setup_sorted_view() const; // modifies mutable state
    void reset_sorted_view();
};
 
template<typename T, typename C, typename A>
class kll_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;
 
  friend class kll_sketch<T, C, A>;
  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:
  const T* items;
  const uint32_t* levels;
  const uint8_t num_levels;
  uint32_t index;
  uint8_t level;
  uint64_t weight;
  const_iterator(const T* items, const uint32_t* levels, const uint8_t num_levels);
};
 
} /* namespace datasketches */
 
#include "kll_sketch_impl.hpp"
 
#endif