JSONStorage.h

Go to the documentation of this file.

//
// Created by Nico Schäfer on 9/7/17.
//
 
#ifndef JODA_JSONSTORAGE_H
#define JODA_JSONSTORAGE_H
 
#include <joda/container/ContainerFlags.h>
#include <joda/container/JSONContainer.h>
 
#include <joda/container/ContainerFlags.h>
#include <joda/container/JSONContainer.h>
#include <future>
#include <mutex>
#include <unordered_map>
class QueryPlan;
 
class JSONStorage {
 public:
  virtual ~JSONStorage();
  void preparePurge();
  explicit JSONStorage(std::string query_string);
  void insertDocumentsQueue(JsonContainerQueue::queue_t *queue);
  void insertDocumentsQueue(JsonContainerQueue::queue_t *queue,
                            size_t &insertedDocs, size_t &insertedConts);
 
  void insertDocuments(
      moodycamel::ConcurrentQueue<std::unique_ptr<JSONContainer>> &queue,
      const std::atomic_bool &cf, std::atomic_uint &cs);
 
  void insertDocuments(std::unique_ptr<JSONContainer> &&cont);
 
  void getDocumentsQueue(JsonContainerRefQueue::queue_t *queue);
 
  unsigned long size() const;
  unsigned long contSize() const;
 
  // Memory
 
  size_t estimatedSize() const;
  size_t estimatedCapacity() const;
 
  void freeAllMemory();
 
  size_t parsedSize() const;
 
  const std::string &getName() const;
 
  void writeFile(const std::string &file);
  void writeFiles(const std::string &file);
 
  std::vector<std::string> stringify(unsigned long start = 0,
                                     unsigned long end = ULONG_MAX);
 
  std::vector<std::shared_ptr<RJDocument>> getRaw(
      unsigned long start = 0, unsigned long end = ULONG_MAX);
 
  template <class Handler>
  std::vector<bool> AcceptDocuments(Handler &handler, unsigned long start = 0,
                                    unsigned long end = ULONG_MAX) {
    std::lock_guard<std::mutex> lock(documentMutex);
    std::vector<bool> ret;
    if (docCount == 0) return ret;
    end = std::min(end, docCount - 1);
    if (start > end) return ret;
    auto count = (end - start) + 1;
    ret.reserve(count);
    auto it = container.begin();
 
    unsigned long skippedSize = 0;
    // Browse container until designated is reached
    while (start >= (*it)->size() + skippedSize) {
      skippedSize += (*it)->size();
      it++;
      DCHECK(it != container.end());
    }
 
    while (ret.size() < count) {
      size_t contStart = start - skippedSize;
      contStart = std::max(contStart, (size_t)0);
      unsigned long contEnd = contStart + count - ret.size() - 1;
      if (it == container.end()) {
        DCHECK(false) << "If all Container contributed to the raws, this "
                         "should not be possible";
        return ret;
      }
 
      if ((*it)->size() != 0) {
        auto tmp = (*it)->AcceptDocuments(handler, contStart, contEnd);
        DCHECK_EQ(tmp.size(), contEnd - contStart + 1);
        std::move(tmp.begin(), tmp.end(), std::back_inserter(ret));
        it++;
      } else {
        DLOG(INFO) << "Skipping empty container";
      }
    }
    return ret;
  }
 
  const std::string &getRegtmpdir() const;
 
  unsigned long getLastUsed() const;
 
  void addQueryString(const std::string &query_string);
 
  const std::string &getQueryString() const;
 
 
 
 protected:
  friend QueryPlan;
  const std::vector<std::unique_ptr<JSONContainer>> &getContainer() const;
 
  std::string name;
  std::string regtmpdir;
 
  // Stored Queries
  std::string query;
 
  std::mutex documentMutex;
  std::vector<std::unique_ptr<JSONContainer>> container;
  unsigned long docCount = 0;
 
  // Multithread
  size_t threadCount(size_t containerSize) const;
 
  std::string getStorageID() const;
};
 
#endif  // JODA_JSONSTORAGE_H