RelAlgExecutionDescriptor.h

Go to the documentation of this file.

/*
 * Copyright 2022 HEAVY.AI, Inc.
 *
 * Licensed 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.
 */

#pragma once

#include <boost/graph/adjacency_list.hpp>

#include "QueryEngine/Descriptors/QueryMemoryDescriptor.h"
#include "QueryEngine/JoinFilterPushDown.h"
#include "QueryEngine/ResultSet.h"
#include "Shared/TargetInfo.h"
#include "Shared/toString.h"

class ResultSet;

class ExecutionResult {
 public:
  ExecutionResult();

  ExecutionResult(const std::shared_ptr<ResultSet>& rows,
                  const std::vector<TargetMetaInfo>& targets_meta);

  ExecutionResult(ResultSetPtr&& result, const std::vector<TargetMetaInfo>& targets_meta);

  ExecutionResult(const ExecutionResult& that);

  ExecutionResult(ExecutionResult&& that);

  ExecutionResult(const std::vector<PushedDownFilterInfo>& pushed_down_filter_info,
                  bool filter_push_down_enabled);

  ExecutionResult& operator=(const ExecutionResult& that);

  const std::shared_ptr<ResultSet>& getRows() const { return result_; }

  bool empty() const { return !result_; }

  const ResultSetPtr& getDataPtr() const { return result_; }

  const std::vector<TargetMetaInfo>& getTargetsMeta() const { return targets_meta_; }

  const std::vector<PushedDownFilterInfo>& getPushedDownFilterInfo() const;

  const bool isFilterPushDownEnabled() const { return filter_push_down_enabled_; }

  void setQueueTime(const int64_t queue_time_ms) {
    CHECK(result_);
    result_->setQueueTime(queue_time_ms);
  }

  std::string toString() const {
    return ::typeName(this) + "(" + ::toString(result_) + ", " +
           ::toString(targets_meta_) + ")";
  }

  enum RType { QueryResult, SimpleResult, Explanation, CalciteDdl };

  std::string getExplanation();
  void updateResultSet(const std::string& query_ra, RType type, bool success = true);
  RType getResultType() const { return type_; }
  void setResultType(RType type) { type_ = type; }
  int64_t getExecutionTime() const { return execution_time_ms_; }
  void setExecutionTime(int64_t execution_time_ms) {
    execution_time_ms_ = execution_time_ms;
  }
  void addExecutionTime(int64_t execution_time_ms) {
    execution_time_ms_ += execution_time_ms;
  }

 private:
  ResultSetPtr result_;
  std::vector<TargetMetaInfo> targets_meta_;
  // filters chosen to be pushed down
  std::vector<PushedDownFilterInfo> pushed_down_filter_info_;
  // whether or not it was allowed to look for filters to push down
  bool filter_push_down_enabled_;

  bool success_;
  uint64_t execution_time_ms_;
  RType type_;
};

class RelAlgNode;

class RaExecutionDesc {
 public:
  RaExecutionDesc(const RelAlgNode* body)
      : body_(body)
      , result_(std::make_shared<ResultSet>(std::vector<TargetInfo>{},
                                            ExecutorDeviceType::CPU,
                                            QueryMemoryDescriptor(),
                                            nullptr,
                                            nullptr,
                                            0,
                                            0),
                {}) {}

  const ExecutionResult& getResult() const { return result_; }

  void setResult(const ExecutionResult& result);

  const RelAlgNode* getBody() const;

 private:
  const RelAlgNode* body_;
  ExecutionResult result_;
};

using DAG = boost::
    adjacency_list<boost::setS, boost::vecS, boost::bidirectionalS, const RelAlgNode*>;
using Vertex = DAG::vertex_descriptor;

class RaExecutionSequence {
 public:
  RaExecutionSequence(const RelAlgNode*, Executor*, const bool build_sequence = true);
  RaExecutionSequence(std::unique_ptr<RaExecutionDesc> exec_desc);

  RaExecutionDesc* next();

  RaExecutionDesc* prev();

  std::optional<size_t> nextStepId(const bool after_broadcast) const;

  bool executionFinished() const;

  void extractQueryStepSkippingInfo();

  void skipQuerySteps();

  std::vector<Vertex> mergeSortWithInput(const std::vector<Vertex>& vertices,
                                         const DAG& graph);

  const std::unordered_map<int, QueryPlanHash> getSkippedQueryStepCacheKeys() const {
    return cached_resultset_keys_;
  }

  RaExecutionDesc* getDescriptor(size_t idx) const {
    CHECK_LT(idx, descs_.size());
    return descs_[idx].get();
  }

  RaExecutionDesc* getDescriptorByBodyId(unsigned const body_id,
                                         size_t const start_idx) const;

  size_t size() const { return descs_.size(); }
  bool empty() const { return descs_.empty(); }

  size_t totalDescriptorsCount() const;

  const bool hasQueryStepForUnion() const { return has_step_for_union_; }

 private:
  DAG graph_;
  Executor* executor_;

  std::unordered_set<Vertex> joins_;
  std::vector<Vertex> ordering_;  // reverse order topological sort of graph_
  size_t current_vertex_ = 0;
  size_t scan_count_ = 0;

  std::unordered_map<const RelAlgNode*, int> node_ptr_to_vert_idx_;
  std::unordered_map<int, std::unordered_set<int>> skippable_steps_;
  // a set of query steps that their result sets are cached
  std::unordered_set<int> cached_query_steps_;
  std::unordered_map<int, QueryPlanHash> cached_resultset_keys_;
  bool has_step_for_union_{false};
  bool has_limit_clause_{false};

  size_t stepsToNextBroadcast() const;

  // The execution descriptors hold the pointers to their results. We need to push them
  // back into this vector as they are created, so we don't lose the intermediate results
  // later.
  std::vector<std::unique_ptr<RaExecutionDesc>> descs_;
};