<commit_before>/**
 * PRIVATE HEADER
 *
 * Data structures for main memory storage.
 *
 * Copyright (c) 2013 <NAME> <<EMAIL>>
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the Free
 * Software Foundation; either version 2 of the License, or (at your option)
 * any later version.
 */


#pragma once
#include "page.h"
#include "cursor.h"
#include "counters.h"

#include <cpp-btree/btree_map.h>

#include <tuple>
#include <vector>
#include <algorithm>
#include <deque>
#include <memory>
#include <mutex>

#include <tbb/enumerable_thread_specific.h>
#include <tbb/concurrent_hash_map.h>
#include <tbb/concurrent_queue.h>

namespace Akumuli {


struct Sequence
{
    //! Container type
    typedef std::tuple<TimeStamp, ParamId> KeyType;
    typedef btree::btree_multimap<std::tuple<TimeStamp, ParamId>, EntryOffset> MapType;
    typedef std::tuple<TimeStamp, ParamId, EntryOffset> ValueType;

    MapType                 data_;          //< Dictionary
    mutable std::mutex      obj_mtx_;       //< data_ mutex
    mutable std::mutex      tmp_mtx_;       //< temp_ mutex
    std::vector<ValueType>  temp_;          //< Temporary storage

    Sequence() {}

    Sequence(Sequence const& other) = delete;
    Sequence& operator = (Sequence const& other) = delete;

    /**  Add item to cache.
      *  @return AKU_WRITE_STATUS_OVERFLOW if sequence is full. Note that write is successful anyway.<commit_msg>Add sequencer class - replacement of the cache class<commit_after>/**
 * PRIVATE HEADER
 *
 * Data structures for main memory storage.
 *
 * Copyright (c) 2013 <NAME> <<EMAIL>>
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the Free
 * Software Foundation; either version 2 of the License, or (at your option)
 * any later version.
 */


#pragma once
#include "page.h"
#include "cursor.h"
#include "counters.h"

#include <cpp-btree/btree_map.h>

#include <tuple>
#include <vector>
#include <algorithm>
#include <deque>
#include <memory>
#include <mutex>

#include <tbb/enumerable_thread_specific.h>
#include <tbb/concurrent_hash_map.h>
#include <tbb/concurrent_queue.h>

namespace Akumuli {

template<typename RunType>
bool top_element_less(const RunType& x, const RunType& y)
{
    return x.back() < y.back();
}

template<typename RunType>
bool top_element_more(const RunType& x, const RunType& y)
{
    return top_element_less(y, x);
}

struct TimeSeriesValue {
    std::tuple<TimeStamp, ParamId> key_;
    EntryOffset value_;

    TimeSeriesValue() {}

    TimeSeriesValue(TimeStamp ts, ParamId id, EntryOffset offset)
        : key_(ts, id)
        , value_(offset)
    {
    }

    friend bool operator < (TimeSeriesValue const& lhs, TimeSeriesValue const& rhs) {
        return lhs.key_ < rhs.key_;
    }
};

/** Time-series sequencer.
  * @brief Akumuli can accept unordered time-series (this is the case when
  * clocks of the different time-series sources are slightly out of sync).
  * This component accepts all of them, filter out late writes and reorder
  * all the remaining samples by timestamp and parameter id.
  */
struct Sequencer {
    typedef std::vector<TimeSeriesValue> SortedRun;

    std::vector<SortedRun>  runs_;
    SortedRun               key_;
    const size_t            window_size_;
    const PageHeader* const page_;

    Sequencer(PageHeader const* page, size_t window_size)
        : window_size_(window_size)
        , page_(page)
    {
        key_.push_back(TimeSeriesValue());
    }

    void check_outdated_runs() {

    }

    void add(TimeSeriesValue const& value) {
        key_.pop_back();
        key_.push_back(value);
        auto begin = runs_.begin();
        auto end = runs_.end();
        if (runs_.size() > window_size_) {
            std::advance(begin, runs_.size() - window_size_);
        }
        auto insert_it = std::lower_bound(begin, end, key_, top_element_more<SortedRun>);
        if (insert_it == runs_.end()) {
            SortedRun new_pile;
            new_pile.push_back(value);
            runs_.push_back(new_pile);
            // ammortised check
            check_outdated_runs();
        } else {
            insert_it->push_back(value);
        }
    }

    template<class Iter>
    void merge(Iter out_iter) {
        size_t n = runs_.size();
        typedef typename SortedRun::const_iterator iter_t;
        iter_t iter[n], ends[n];
        int cnt = 0;
        for(auto i = runs_.begin(); i != runs_.end(); i++) {
            iter[cnt] = i->begin();
            ends[cnt] = i->end();
            cnt++;
        }

        typedef std::tuple<TimeSeriesValue, int> HeapItem;
        typedef std::vector<HeapItem> Heap;
        Heap heap;

        for(auto index = 0u; index < n; index++) {
            if (iter[index] != ends[index]) {
                auto value = *iter[index];
                iter[index]++;
                heap.push_back(std::make_tuple(value, index));
            }
        }

        std::make_heap(heap.begin(), heap.end(), std::greater<HeapItem>());

        while(!heap.empty()) {
            std::pop_heap(heap.begin(), heap.end(), std::greater<HeapItem>());
            auto item = heap.back();
            auto value = std::get<0>(item);
            int index = std::get<1>(item);
            *out_iter++ = value;
            heap.pop_back();
            if (iter[index] != ends[index]) {
                auto value = *iter[index];
                iter[index]++;
                heap.push_back(std::make_tuple(value, index));
                std::push_heap(heap.begin(), heap.end(), std::greater<HeapItem>());
            }
        }
    }
};



struct Sequence
{
    //! Container type
    typedef std::tuple<TimeStamp, ParamId> KeyType;
    typedef btree::btree_multimap<std::tuple<TimeStamp, ParamId>, EntryOffset> MapType;
    typedef std::tuple<TimeStamp, ParamId, EntryOffset> ValueType;

    MapType                 data_;          //< Dictionary
    mutable std::mutex      obj_mtx_;       //< data_ mutex
    mutable std::mutex      tmp_mtx_;       //< temp_ mutex
    std::vector<ValueType>  temp_;          //< Temporary storage

    Sequence() {}

    Sequence(Sequence const& other) = delete;
    Sequence& operator = (Sequence const& other) = delete;

    /**  Add item to cache.
      *  @return AKU_WRITE_STATUS_OVERFLOW if sequence is full. Note that write is successful anyway.
{'fraction_non_alphanumeric': 0.11821675634127594, 'fraction_numerical': 0.0021521906225980014, 'mean_word_length': 2.636668529904975, 'pattern_counts': {'":': 0, '<': 65, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2705814', 'n_tokens_mistral': 2023, 'n_tokens_neox': 1967, 'n_words': 544}