QueryThread.h

Go to the documentation of this file.

//
// Created by Nico Schäfer on 11/7/17.
//
 
#ifndef JODA_QUERYTHREAD_H
#define JODA_QUERYTHREAD_H
 
#include <joda/concurrency/IOThreadPool.h>
#include <joda/misc/Benchmark.h>
#include <joda/query/Query.h>
#include <joda/query/aggregation/IAggregator.h>
 
#include "executor/IQueryExecutor.h"
 
struct QueryThreadConfig {
  std::shared_ptr<joda::query::Query> q;
  Benchmark *bench;
  std::vector<std::unique_ptr<IQueryExecutor>> executors;
  std::vector<std::unique_ptr<joda::query::IAggregator>> aggregators;
  std::shared_ptr<JoinManager> joinManager;
  joda::query::AggregatorQueue::queue_t *aggQueue{};
  std::string tmpdir;
 
  QueryThreadConfig() = default;
 
  QueryThreadConfig(std::shared_ptr<joda::query::Query> &q)
      : q(q),
        bench(nullptr),
        executors(),
        aggregators(),
        joinManager(q->getStoreJoinManager()) {
    for (const auto &agg : this->q->getAggregators()) {
      aggregators.emplace_back(agg->duplicate());
    }
  }
 
  QueryThreadConfig(const QueryThreadConfig &o)
      : q(o.q),
        bench(o.bench),
        executors(),
        aggregators(),
        joinManager(o.joinManager),
        aggQueue(o.aggQueue),
        tmpdir(o.tmpdir) {
    for (const auto &exec : o.executors) {
      executors.emplace_back(exec->duplicate());
    }
    for (const auto &agg : o.aggregators) {
      aggregators.emplace_back(agg->duplicate());
    }
  }
 
  void addExecutor(std::unique_ptr<IQueryExecutor> &&exec) {
    executors.emplace_back(std::move(exec));
  }
};
 
class QueryThread
    : public IWorkerThread<JsonContainerRefQueue, JsonContainerQueue,
                           QueryThreadConfig> {
 public:
  QueryThread(IQueue *iqueue, OQueue *oqueue, WConf &conf);
  ~QueryThread() override;
 
  typedef IPayload ContRef;
  typedef OPayload OwnedCont;
 
  const std::string getThreadID() const;
 
  /*
   * Aggregation
   */
 
  bool hasAggregators() const;
 
 protected:
  bool hasToProject() const;
  bool canCreateView() const;
 
  /*
   * Functions
   */
  std::shared_ptr<const DocIndex> select(JSONContainer &cont) const;
 
  std::vector<std::unique_ptr<RJDocument>> defaultProject(
      JSONContainer &cont, const DocIndex &ids, RJMemoryPoolAlloc &alloc) const;
 
  void aggregate(const std::vector<RapidJsonDocument> &docs) const;
 
  void aggregate(ContRef pipelineCont,
                 const std::shared_ptr<const DocIndex> &selectResult,
                 bool isSelected) const;
 
  void work() override;
 
  void logTimers() const;
  RecurringTimer bloom_timer;
  RecurringTimer select_timer;
  RecurringTimer project_timer;
  RecurringTimer aggregate_timer;
  RecurringTimer copy_timer;
  RecurringTimer serialize_timer;
  RecurringTimer sample_view_cost_timer;
};
 
#endif  // JODA_QUERYTHREAD_H