RaExecutionSequence Class Reference

A container for relational algebra descriptors defining the execution order for a relational algebra query. Holds the relational algebra descriptors for executing a relational algebra query. Each descriptor holds both a top-level relational algebra node and a ResultSet ptr holding the results from the execution of the accompany node(s). The sequence can be generated on initialization or lazily with calls to the next() operator. More...

#include <RelAlgExecutionDescriptor.h>

Collaboration diagram for RaExecutionSequence:

Public Member Functions
	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
RaExecutionDesc *	getDescriptor (size_t idx) const
RaExecutionDesc *	getDescriptorByBodyId (unsigned const body_id, size_t const start_idx) const
size_t	size () const
bool	empty () const
size_t	totalDescriptorsCount () const
const bool	hasQueryStepForUnion () const

Private Member Functions
size_t	stepsToNextBroadcast () const

Private Attributes
DAG	graph_
Executor *	executor_
std::unordered_set< Vertex >	joins_
std::vector< Vertex >	ordering_
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_
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}
std::vector< std::unique_ptr < RaExecutionDesc > >	descs_

Detailed Description

A container for relational algebra descriptors defining the execution order for a relational algebra query. Holds the relational algebra descriptors for executing a relational algebra query. Each descriptor holds both a top-level relational algebra node and a ResultSet ptr holding the results from the execution of the accompany node(s). The sequence can be generated on initialization or lazily with calls to the next() operator.

Definition at line 134 of file RelAlgExecutionDescriptor.h.

Constructor & Destructor Documentation

RaExecutionSequence::RaExecutionSequence	(	const RelAlgNode *	sink,
		Executor *	executor,
		const bool	build_sequence = true
	)

Definition at line 209 of file RelAlgExecutionDescriptor.cpp.

References anonymous_namespace{RelAlgExecutionDescriptor.cpp}::build_dag(), CHECK, executor_, extractQueryStepSkippingInfo(), g_allow_query_step_skipping, g_cluster, g_enable_data_recycler, g_use_query_resultset_cache, anonymous_namespace{RelAlgExecutionDescriptor.cpp}::get_join_vertices(), graph_, has_step_for_union_, joins_, mergeSortWithInput(), next(), node_ptr_to_vert_idx_, ordering_, gpu_enabled::reverse(), and skipQuerySteps().

                                                                    {
  CHECK(sink);
  CHECK(executor);
  if (dynamic_cast<const RelScan*>(sink) || dynamic_cast<const RelJoin*>(sink)) {
    throw std::runtime_error("Query not supported yet");
  }
  executor_ = executor;
  graph_ = build_dag(sink, node_ptr_to_vert_idx_);

  boost::topological_sort(graph_, std::back_inserter(ordering_));
  std::reverse(ordering_.begin(), ordering_.end());

  ordering_ = mergeSortWithInput(ordering_, graph_);
  joins_ = get_join_vertices(ordering_, graph_);

  if (build_sequence) {
    while (next()) {
      // noop
    }
    if (g_enable_data_recycler && g_use_query_resultset_cache &&
        g_allow_query_step_skipping && !has_step_for_union_ && !g_cluster) {
      extractQueryStepSkippingInfo();
      skipQuerySteps();
    }
  }
}

Here is the call graph for this function:

RaExecutionSequence::RaExecutionSequence

(

std::unique_ptr< RaExecutionDesc >

exec_desc

)

Definition at line 238 of file RelAlgExecutionDescriptor.cpp.

References descs_.

                                                                                 {
  descs_.emplace_back(std::move(exec_desc));
}

Member Function Documentation

bool RaExecutionSequence::empty ( ) const

inline

Definition at line 190 of file RelAlgExecutionDescriptor.h.

References descs_.

Referenced by RelAlgExecutor::executeRelAlgSeq().

{ return descs_.empty(); }

Here is the caller graph for this function:

bool RaExecutionSequence::executionFinished

(

)

const

Definition at line 431 of file RelAlgExecutionDescriptor.cpp.

References current_vertex_, nextStepId(), ordering_, and totalDescriptorsCount().

                                                  {
  if (current_vertex_ == ordering_.size()) {
    // All descriptors visited, execution finished
    return true;
  } else {
    const auto next_step_id = nextStepId(true);
    if (!next_step_id || (*next_step_id == totalDescriptorsCount())) {
      // One step remains (the current vertex), or all remaining steps can be executed
      // without another broadcast (i.e. on the aggregator)
      return true;
    }
  }
  return false;
}

Here is the call graph for this function:

void RaExecutionSequence::extractQueryStepSkippingInfo

(

)

Extracts a topological information of the query plan DAG about the relationship among query steps where a parent step and its child steps can be skipped if we have a cached resultset of the parent step

Definition at line 328 of file RelAlgExecutionDescriptor.cpp.

References CHECK, descs_, graph_, anonymous_namespace{RelAlgDag.cpp}::node_id(), node_ptr_to_vert_idx_, run_benchmark_import::res, and skippable_steps_.

Referenced by RaExecutionSequence().

                                                       {
  const auto pushChildNodes = [&](auto& stack, const auto node_id) {
    auto [in_edge_iter, in_edge_end] = boost::in_edges(node_id, graph_);
    while (in_edge_iter != in_edge_end) {
      const auto child_node_id = boost::source(*in_edge_iter, graph_);
      stack.push(graph_[child_node_id]);
      std::advance(in_edge_iter, 1);
    }
  };
  auto top_node_id = descs_.size() - 1;
  auto top_res = skippable_steps_.emplace(top_node_id, std::unordered_set<int>{});
  CHECK(top_res.second);
  for (auto it = descs_.begin(); it != std::prev(descs_.end()); ++it) {
    const auto step = it->get();
    const auto body = step->getBody();
    const auto step_id = static_cast<int>(std::distance(descs_.begin(), it));
    auto res = skippable_steps_.emplace(step_id, std::unordered_set<int>{});
    CHECK(res.second);
    skippable_steps_[top_node_id].insert(step_id);  // top-desc can skip all child descs
    std::stack<const RelAlgNode*> stack;
    pushChildNodes(stack, node_ptr_to_vert_idx_[body]);
    while (!stack.empty()) {
      auto child_node = stack.top();
      stack.pop();
      // descs_ is based on the topologically sorted (flattened) DAG, so we can limit the
      // search range for child descs
      auto is_desc_body = std::find_if(
          descs_.begin(), it, [&child_node](std::unique_ptr<RaExecutionDesc>& ptr) {
            return ptr->getBody() == child_node;
          });
      if (is_desc_body != it) {
        // due to the topological sorting of query plan DAG, we can avoid visiting "all"
        // child nodes
        const auto child_step_id =
            static_cast<int>(std::distance(descs_.begin(), is_desc_body));
        skippable_steps_[step_id].insert(child_step_id);
        skippable_steps_[step_id].insert(skippable_steps_[child_step_id].begin(),
                                         skippable_steps_[child_step_id].end());
      } else {
        pushChildNodes(stack, node_ptr_to_vert_idx_[child_node]);
      }
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

RaExecutionDesc* RaExecutionSequence::getDescriptor ( size_t  idx ) const

inline

Definition at line 181 of file RelAlgExecutionDescriptor.h.

References CHECK_LT, and descs_.

Referenced by RelAlgExecutor::executeRelAlgQuerySingleStep(), RelAlgExecutor::executeRelAlgSeq(), RelAlgExecutor::executeRelAlgStep(), and RelAlgExecutor::executeRelAlgSubSeq().

                                                   {
    CHECK_LT(idx, descs_.size());
    return descs_[idx].get();
  }

Here is the caller graph for this function:

RaExecutionDesc * RaExecutionSequence::getDescriptorByBodyId	(	unsigned const	body_id,
		size_t const	start_idx
	)		const

Definition at line 456 of file RelAlgExecutionDescriptor.cpp.

References CHECK_LT, and descs_.

Referenced by RelAlgExecutor::executeRelAlgStep().

                                  {
  CHECK_LT(start_idx, descs_.size());
  auto const from_end = descs_.size() - (start_idx + 1);
  MatchBody const match_body{body_id};
  auto const itr = std::find_if(descs_.rbegin() + from_end, descs_.rend(), match_body);
  return itr == descs_.rend() ? nullptr : itr->get();
}

Here is the caller graph for this function:

const std::unordered_map<int, QueryPlanHash> RaExecutionSequence::getSkippedQueryStepCacheKeys ( ) const

inline

Definition at line 177 of file RelAlgExecutionDescriptor.h.

References cached_resultset_keys_.

Referenced by RelAlgExecutor::executeRelAlgSeq().

                                                                                  {
    return cached_resultset_keys_;
  }

Here is the caller graph for this function:

const bool RaExecutionSequence::hasQueryStepForUnion ( ) const

inline

Definition at line 194 of file RelAlgExecutionDescriptor.h.

References has_step_for_union_.

Referenced by RelAlgExecutor::executeRelAlgSeq(), and RelAlgExecutor::executeRelAlgStep().

{ return has_step_for_union_; }

Here is the caller graph for this function:

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

Definition at line 282 of file RelAlgExecutionDescriptor.cpp.

References CHECK, has_limit_clause_, and gpu_enabled::sort().

Referenced by RaExecutionSequence().

                      {
  DAG::in_edge_iterator ie_iter, ie_end;
  std::unordered_set<Vertex> inputs;
  for (const auto vert : vertices) {
    if (const auto sort = dynamic_cast<const RelSort*>(graph[vert])) {
      boost::tie(ie_iter, ie_end) = boost::in_edges(vert, graph);
      CHECK(size_t(1) == sort->inputCount() && boost::next(ie_iter) == ie_end);
      if (sort->isLimitDelivered()) {
        has_limit_clause_ = true;
      }
      const auto in_vert = boost::source(*ie_iter, graph);
      const auto input = graph[in_vert];
      if (dynamic_cast<const RelScan*>(input)) {
        throw std::runtime_error("Standalone sort not supported yet");
      }
      if (boost::out_degree(in_vert, graph) > 1) {
        throw std::runtime_error("Sort's input node used by others not supported yet");
      }
      inputs.insert(in_vert);
    }
  }

  std::vector<Vertex> new_vertices;
  for (const auto vert : vertices) {
    if (inputs.count(vert)) {
      continue;
    }
    new_vertices.push_back(vert);
  }
  return new_vertices;
}

Here is the call graph for this function:

Here is the caller graph for this function:

RaExecutionDesc * RaExecutionSequence::next

(

)

Return the next execution descriptor in the sequence. If no more execution descriptors exist, returns nullptr.

Definition at line 242 of file RelAlgExecutionDescriptor.cpp.

References cached_query_steps_, cached_resultset_keys_, CHECK, CHECK_GE, current_vertex_, descs_, EMPTY_QUERY_PLAN, executor_, QueryPlanDagExtractor::extractQueryPlanDag(), g_allow_query_step_skipping, RelAlgNode::getId(), RelAlgNode::getQueryPlanDagHash(), graph_, has_step_for_union_, joins_, ordering_, scan_count_, and RelAlgNode::setOutputMetainfo().

Referenced by RaExecutionSequence().

                                           {
  auto checkQueryStepSkippable =
      [&](RelAlgNode const* node, QueryPlanHash key, size_t step_id) {
        CHECK_GE(step_id, 0);
        if (executor_->getRecultSetRecyclerHolder().hasCachedQueryResultSet(key)) {
          cached_query_steps_.insert(step_id);
          cached_resultset_keys_.emplace(-node->getId(), key);
          const auto output_meta_info =
              executor_->getRecultSetRecyclerHolder().getOutputMetaInfo(key);
          CHECK(output_meta_info);
          node->setOutputMetainfo(*output_meta_info);
        }
      };
  while (current_vertex_ < ordering_.size()) {
    auto vert = ordering_[current_vertex_++];
    if (joins_.count(vert)) {
      continue;
    }
    auto& node = graph_[vert];
    CHECK(node);
    if (dynamic_cast<const RelScan*>(node)) {
      scan_count_++;
      continue;
    }
    descs_.emplace_back(std::make_unique<RaExecutionDesc>(node));
    auto logical_union = dynamic_cast<const RelLogicalUnion*>(node);
    if (logical_union) {
      has_step_for_union_ = true;
    }
    auto extracted_query_plan_dag =
        QueryPlanDagExtractor::extractQueryPlanDag(node, executor_);
    if (g_allow_query_step_skipping &&
        !boost::iequals(extracted_query_plan_dag.extracted_dag, EMPTY_QUERY_PLAN)) {
      checkQueryStepSkippable(node, node->getQueryPlanDagHash(), descs_.size() - 1);
    }
    return descs_.back().get();
  }
  return nullptr;
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::optional< size_t > RaExecutionSequence::nextStepId

(

const bool

after_broadcast

)

const

Returns the index of the next execution descriptor in the graph. If after_broadcast is true, returns the index of the first execution descriptor after the next global broadcast. Returns std::nullopt if no execution descriptors remain in the graph.

Definition at line 418 of file RelAlgExecutionDescriptor.cpp.

References current_vertex_, descs_, ordering_, and stepsToNextBroadcast().

Referenced by executionFinished().

                                                                                    {
  if (after_broadcast) {
    if (current_vertex_ == ordering_.size()) {
      return std::nullopt;
    }
    return descs_.size() + stepsToNextBroadcast();
  } else if (current_vertex_ == ordering_.size()) {
    return std::nullopt;
  } else {
    return descs_.size();
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

RaExecutionDesc * RaExecutionSequence::prev

(

)

Return the previous execution descriptor in the sequence. If the sequence is made up of 0 or 1 descriptors, returns nullptr.

Definition at line 316 of file RelAlgExecutionDescriptor.cpp.

References CHECK_GE, and descs_.

                                           {
  if (descs_.empty()) {
    return nullptr;
  }
  if (descs_.size() == 1) {
    return nullptr;
  }
  CHECK_GE(descs_.size(), size_t(2));
  auto itr = descs_.rbegin();
  return (++itr)->get();
}

size_t RaExecutionSequence::size ( ) const

inline

Definition at line 189 of file RelAlgExecutionDescriptor.h.

References descs_.

Referenced by RelAlgExecutor::executeRelAlgQueryWithFilterPushDown(), and RelAlgExecutor::executeRelAlgSeq().

{ return descs_.size(); }

Here is the caller graph for this function:

void RaExecutionSequence::skipQuerySteps

(

)

Optimizes the query execution step based on the information extracted from the extractQueryStepSkippingInfo function

Definition at line 373 of file RelAlgExecutionDescriptor.cpp.

References cached_query_steps_, cached_resultset_keys_, CHECK, CHECK_LE, descs_, skippable_steps_, gpu_enabled::swap(), and VLOG.

Referenced by RaExecutionSequence().

                                         {
  CHECK_LE(cached_query_steps_.size(), descs_.size());

  // extract skippable query steps`
  std::unordered_set<int> skippable_query_steps;
  for (const auto cached_step_id : cached_query_steps_) {
    auto it = skippable_steps_.find(cached_step_id);
    CHECK(it != skippable_steps_.end());
    const auto& child_steps = it->second;
    std::for_each(
        child_steps.begin(), child_steps.end(), [&](const auto& skippable_step_id) {
          if (skippable_step_id != cached_step_id) {
            skippable_query_steps.insert(skippable_step_id);
          }
        });
  }

  // modify query step sequence based on the skippable query steps
  if (!skippable_query_steps.empty()) {
    std::vector<std::unique_ptr<RaExecutionDesc>> new_descs;
    for (size_t step_id = 0; step_id < descs_.size(); ++step_id) {
      const auto body = descs_[step_id]->getBody();
      if (!skippable_query_steps.count(step_id)) {
        new_descs.push_back(std::make_unique<RaExecutionDesc>(body));
      }
    }
    const auto prev_num_steps = descs_.size();
    if (!new_descs.empty()) {
      std::swap(descs_, new_descs);
    }
    VLOG(1) << "Skip " << prev_num_steps - descs_.size() << " query steps from "
            << prev_num_steps << " steps";
  }

  for (const auto& desc : descs_) {
    // remove cached resultset info for each desc since
    // it is not skipped
    auto body = desc->getBody();
    auto it = cached_resultset_keys_.find(-body->getId());
    if (it != cached_resultset_keys_.end()) {
      cached_resultset_keys_.erase(it);
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t RaExecutionSequence::stepsToNextBroadcast ( ) const

private

Starting from the current vertex, iterate the graph counting the number of execution descriptors remaining before the next required broadcast step. The current vertex is counted as the first step before a broadcast is required; i.e. a return value of 0 indicates no additional steps in the graph can be executed without a global broadcast, and a return value of 2 indicates the current vertex and both subsequent vertices can be executed before a global broacast is needed.

Definition at line 484 of file RelAlgExecutionDescriptor.cpp.

References CHECK, CHECK_EQ, current_vertex_, graph_, joins_, ordering_, and gpu_enabled::sort().

Referenced by nextStepId().

                                                       {
  size_t steps_to_next_broadcast = 0;
  auto crt_vertex = current_vertex_;
  while (crt_vertex < ordering_.size()) {
    auto vert = ordering_[crt_vertex++];
    auto node = graph_[vert];
    CHECK(node);
    if (joins_.count(vert)) {
      auto join_node = dynamic_cast<const RelLeftDeepInnerJoin*>(node);
      CHECK(join_node);
      for (size_t i = 0; i < join_node->inputCount(); i++) {
        const auto input = join_node->getInput(i);
        if (dynamic_cast<const RelScan*>(input)) {
          return steps_to_next_broadcast;
        }
      }
      if (crt_vertex < ordering_.size() - 1) {
        // Force the parent node of the RelLeftDeepInnerJoin to run on the aggregator.
        // Note that crt_vertex has already been incremented once above for the join node
        // -- increment it again to account for the parent node of the join
        ++steps_to_next_broadcast;
        ++crt_vertex;
        continue;
      } else {
        CHECK_EQ(crt_vertex, ordering_.size() - 1);
        // If the join node parent is the last node in the tree, run all remaining steps
        // on the aggregator
        return ++steps_to_next_broadcast;
      }
    }
    if (auto sort = dynamic_cast<const RelSort*>(node)) {
      CHECK_EQ(sort->inputCount(), size_t(1));
      node = sort->getInput(0);
    }
    if (dynamic_cast<const RelScan*>(node)) {
      return steps_to_next_broadcast;
    }
    if (dynamic_cast<const RelModify*>(node)) {
      // Modify runs on the leaf automatically, run the same node as a noop on the
      // aggregator
      ++steps_to_next_broadcast;
      continue;
    }
    if (auto project = dynamic_cast<const RelProject*>(node)) {
      if (project->hasWindowFunctionExpr()) {
        ++steps_to_next_broadcast;
        continue;
      }
    }
    for (size_t input_idx = 0; input_idx < node->inputCount(); input_idx++) {
      if (dynamic_cast<const RelScan*>(node->getInput(input_idx))) {
        return steps_to_next_broadcast;
      }
    }
    ++steps_to_next_broadcast;
  }
  return steps_to_next_broadcast;
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t RaExecutionSequence::totalDescriptorsCount

(

)

const

Definition at line 466 of file RelAlgExecutionDescriptor.cpp.

References CHECK, graph_, joins_, and ordering_.

Referenced by executionFinished().

                                                        {
  size_t num_descriptors = 0;
  size_t crt_vertex = 0;
  while (crt_vertex < ordering_.size()) {
    auto vert = ordering_[crt_vertex++];
    if (joins_.count(vert)) {
      continue;
    }
    auto& node = graph_[vert];
    CHECK(node);
    if (dynamic_cast<const RelScan*>(node)) {
      continue;
    }
    ++num_descriptors;
  }
  return num_descriptors;
}

Here is the caller graph for this function:

Member Data Documentation

std::unordered_set<int> RaExecutionSequence::cached_query_steps_

private

Definition at line 208 of file RelAlgExecutionDescriptor.h.

Referenced by next(), and skipQuerySteps().

std::unordered_map<int, QueryPlanHash> RaExecutionSequence::cached_resultset_keys_

private

Definition at line 209 of file RelAlgExecutionDescriptor.h.

Referenced by getSkippedQueryStepCacheKeys(), next(), and skipQuerySteps().

size_t RaExecutionSequence::current_vertex_ = 0

private

Definition at line 202 of file RelAlgExecutionDescriptor.h.

Referenced by executionFinished(), next(), nextStepId(), and stepsToNextBroadcast().

std::vector<std::unique_ptr<RaExecutionDesc> > RaExecutionSequence::descs_

private

Definition at line 226 of file RelAlgExecutionDescriptor.h.

Referenced by empty(), extractQueryStepSkippingInfo(), getDescriptor(), getDescriptorByBodyId(), next(), nextStepId(), prev(), RaExecutionSequence(), size(), and skipQuerySteps().

Executor* RaExecutionSequence::executor_

private

Definition at line 198 of file RelAlgExecutionDescriptor.h.

Referenced by next(), and RaExecutionSequence().

DAG RaExecutionSequence::graph_

private

Definition at line 197 of file RelAlgExecutionDescriptor.h.

Referenced by extractQueryStepSkippingInfo(), next(), RaExecutionSequence(), stepsToNextBroadcast(), and totalDescriptorsCount().

bool RaExecutionSequence::has_limit_clause_ {false}

private

Definition at line 211 of file RelAlgExecutionDescriptor.h.

Referenced by mergeSortWithInput().

bool RaExecutionSequence::has_step_for_union_ {false}

private

Definition at line 210 of file RelAlgExecutionDescriptor.h.

Referenced by hasQueryStepForUnion(), next(), and RaExecutionSequence().

std::unordered_set<Vertex> RaExecutionSequence::joins_

private

Definition at line 200 of file RelAlgExecutionDescriptor.h.

Referenced by next(), RaExecutionSequence(), stepsToNextBroadcast(), and totalDescriptorsCount().

std::unordered_map<const RelAlgNode*, int> RaExecutionSequence::node_ptr_to_vert_idx_

private

Definition at line 205 of file RelAlgExecutionDescriptor.h.

Referenced by extractQueryStepSkippingInfo(), and RaExecutionSequence().

std::vector<Vertex> RaExecutionSequence::ordering_

private

Definition at line 201 of file RelAlgExecutionDescriptor.h.

Referenced by executionFinished(), next(), nextStepId(), RaExecutionSequence(), stepsToNextBroadcast(), and totalDescriptorsCount().

size_t RaExecutionSequence::scan_count_ = 0

private

Definition at line 203 of file RelAlgExecutionDescriptor.h.

Referenced by next().

std::unordered_map<int, std::unordered_set<int> > RaExecutionSequence::skippable_steps_

private

Definition at line 206 of file RelAlgExecutionDescriptor.h.

Referenced by extractQueryStepSkippingInfo(), and skipQuerySteps().

---

The documentation for this class was generated from the following files:

• /home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/Descriptors/RelAlgExecutionDescriptor.h
• /home/jenkins-slave/workspace/core-os-doxygen/QueryEngine/Descriptors/RelAlgExecutionDescriptor.cpp