query_execution/PolicyEnforcerDistributed.cpp - incubator-retired-quickstep - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you 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.
  **/

 #include "query_execution/PolicyEnforcerDistributed.hpp"

 #include <cstddef>
 #include <cstdio>
 #include <cstdlib>
 #include <memory>
 #include <queue>
 #include <sstream>
 #include <string>
 #include <unordered_map>
 #include <utility>
 #include <vector>

 #include "catalog/Catalog.pb.h"
 #include "catalog/CatalogDatabase.hpp"
 #include "catalog/CatalogRelation.hpp"
 #include "catalog/CatalogRelationSchema.hpp"
 #include "catalog/CatalogRelationStatistics.hpp"
 #include "query_execution/QueryContext.pb.h"
 #include "query_execution/QueryExecutionMessages.pb.h"
 #include "query_execution/QueryExecutionState.hpp"
 #include "query_execution/QueryExecutionTypedefs.hpp"
 #include "query_execution/QueryExecutionUtil.hpp"
 #include "query_execution/QueryManagerBase.hpp"
 #include "query_execution/QueryManagerDistributed.hpp"
 #include "query_optimizer/QueryHandle.hpp"
 #include "query_optimizer/QueryProcessor.hpp"
 #include "storage/StorageBlock.hpp"
 #include "storage/StorageBlockInfo.hpp"
 #include "storage/StorageManager.hpp"
 #include "storage/TupleIdSequence.hpp"
 #include "storage/TupleStorageSubBlock.hpp"
 #include "types/TypedValue.hpp"
 #include "utility/ExecutionDAGVisualizer.hpp"

 #include "gflags/gflags.h"
 #include "glog/logging.h"

 #include "tmb/address.h"
 #include "tmb/id_typedefs.h"
 #include "tmb/message_bus.h"
 #include "tmb/tagged_message.h"

 using std::free;
 using std::malloc;
 using std::move;
 using std::ostringstream;
 using std::size_t;
 using std::string;
 using std::unique_ptr;
 using std::vector;

 using tmb::MessageBus;
 using tmb::TaggedMessage;

 namespace quickstep {

 DECLARE_bool(visualize_execution_dag);

 namespace S = serialization;

 DEFINE_uint64(max_msgs_per_dispatch_round, 20, "Maximum number of messages that"
               " can be allocated in a single round of dispatch of messages to"
               " the workers.");

 void PolicyEnforcerDistributed::getWorkOrderProtoMessages(
     vector<unique_ptr<S::WorkOrderMessage>> *work_order_proto_messages) {
   // Iterate over admitted queries until either there are no more
   // messages available, or the maximum number of messages have
   // been collected.
   DCHECK(work_order_proto_messages->empty());
   // TODO(harshad) - Make this function generic enough so that it
   // works well when multiple queries are getting executed.
   if (admitted_queries_.empty()) {
     if (waiting_queries_.empty()) {
       return;
     }

     // Admit the earliest waiting query.
     QueryHandle *new_query = waiting_queries_.front();
     waiting_queries_.pop();
     admitQuery(new_query);
   }

   const std::size_t per_query_share =
       FLAGS_max_msgs_per_dispatch_round / admitted_queries_.size();
   DCHECK_GT(per_query_share, 0u);

   vector<std::size_t> finished_queries_ids;

   for (const auto &admitted_query_info : admitted_queries_) {
     QueryManagerBase *curr_query_manager = admitted_query_info.second.get();
     DCHECK(curr_query_manager != nullptr);
     std::size_t messages_collected_curr_query = 0;
     while (messages_collected_curr_query < per_query_share) {
       unique_ptr<S::WorkOrderMessage> next_work_order_message(
           static_cast<QueryManagerDistributed*>(curr_query_manager)->getNextWorkOrderMessage());
       if (next_work_order_message) {
         ++messages_collected_curr_query;
         work_order_proto_messages->push_back(move(next_work_order_message));
       } else {
         // No more work ordes from the current query at this time.
         // Check if the query's execution is over.
         if (curr_query_manager->getQueryExecutionState().hasQueryExecutionFinished()) {
           // If the query has been executed, remove it.
           finished_queries_ids.push_back(admitted_query_info.first);
         }
         break;
       }
     }
   }
   for (const std::size_t finished_qid : finished_queries_ids) {
     onQueryCompletion(admitted_queries_[finished_qid].get());
     removeQuery(finished_qid);
   }
 }

 bool PolicyEnforcerDistributed::admitQuery(QueryHandle *query_handle) {
   const std::size_t num_shiftbosses = shiftboss_directory_->size();
   // We only allow the number of concurrent queries that equals to
   // that of Shiftbosses.
   if (admitted_queries_.size() > num_shiftbosses) {
     // This query will have to wait.
     waiting_queries_.push(query_handle);
     return false;
   }

   const std::size_t query_id = query_handle->query_id();
   if (admitted_queries_.find(query_id) != admitted_queries_.end()) {
     LOG(ERROR) << "Query with the same ID " << query_id << " exists";
     return false;
   }

   // Ok to admit the query.
   // NOTE(zuyu): Should call before constructing a 'QueryManager'.
   // Otherwise, an InitiateRebuildMessage may be sent before 'QueryContext'
   // initializes.
   initiateQueryInShiftboss(query_handle);

   tmb::Address shiftboss_addresses;
   for (std::size_t i = 0; i < num_shiftbosses; ++i) {
     shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
   }

   // Query with the same ID not present, ok to admit.
   admitted_queries_.emplace(query_id,
                             std::make_unique<QueryManagerDistributed>(
                                 query_handle, foreman_client_id_, num_shiftbosses, move(shiftboss_addresses), bus_));
   return true;
 }

 void PolicyEnforcerDistributed::processInitiateRebuildResponseMessage(const tmb::TaggedMessage &tagged_message) {
   S::InitiateRebuildResponseMessage proto;
   CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

   const std::size_t query_id = proto.query_id();
   DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());

   QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());

   const std::size_t num_rebuild_work_orders = proto.num_rebuild_work_orders();
   const size_t shiftboss_index = proto.shiftboss_index();
   query_manager->processInitiateRebuildResponseMessage(
       proto.operator_index(), num_rebuild_work_orders, shiftboss_index);
   shiftboss_directory_->addNumQueuedWorkOrders(shiftboss_index, num_rebuild_work_orders);

   if (query_manager->getQueryExecutionState().hasQueryExecutionFinished()) {
     onQueryCompletion(query_manager);
     removeQuery(query_id);
   }
 }

 void PolicyEnforcerDistributed::getShiftbossIndexForAggregation(
     const std::size_t query_id,
     const QueryContext::aggregation_state_id aggr_state_index,
     const partition_id part_id,
     const vector<QueryContext::lip_filter_id> &lip_filter_indexes,
     const BlockLocator &block_locator,
     const block_id block,
     const std::size_t next_shiftboss_index_to_schedule,
     std::size_t *shiftboss_index) {
   DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
   QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
   query_manager->getShiftbossIndexForAggregation(aggr_state_index,
                                                  part_id,
                                                  lip_filter_indexes,
                                                  block_locator,
                                                  block,
                                                  next_shiftboss_index_to_schedule,
                                                  shiftboss_index);
 }

 void PolicyEnforcerDistributed::getShiftbossIndexForHashJoin(
     const std::size_t query_id,
     const QueryContext::join_hash_table_id join_hash_table_index,
     const partition_id part_id,
     const vector<QueryContext::lip_filter_id> &lip_filter_indexes,
     const BlockLocator &block_locator,
     const block_id block,
     const std::size_t next_shiftboss_index_to_schedule,
     std::size_t *shiftboss_index) {
   DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
   QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
   query_manager->getShiftbossIndexForHashJoin(join_hash_table_index,
                                               part_id,
                                               lip_filter_indexes,
                                               block_locator,
                                               block,
                                               next_shiftboss_index_to_schedule,
                                               shiftboss_index);
 }

 void PolicyEnforcerDistributed::getShiftbossIndexForLip(
     const std::size_t query_id,
     const vector<QueryContext::lip_filter_id> &lip_filter_indexes,
     const BlockLocator &block_locator,
     const block_id block,
     const std::size_t next_shiftboss_index_to_schedule,
     std::size_t *shiftboss_index) {
   DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
   QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
   query_manager->getShiftbossIndexForLip(lip_filter_indexes,
                                          block_locator,
                                          block,
                                          next_shiftboss_index_to_schedule,
                                          shiftboss_index);
 }

 void PolicyEnforcerDistributed::getShiftbossIndexForNestedLoopsJoin(
     const std::size_t query_id,
     const std::size_t nested_loops_join_index,
     const partition_id part_id,
     const BlockLocator &block_locator,
     const block_id left_block,
     const block_id right_block,
     const std::size_t next_shiftboss_index_to_schedule,
     std::size_t *shiftboss_index_for_nested_loops_join) {
   DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
   QueryManagerDistributed *query_manager = static_cast<QueryManagerDistributed*>(admitted_queries_[query_id].get());
   query_manager->getShiftbossIndexForNestedLoopsJoin(nested_loops_join_index,
                                                      part_id,
                                                      block_locator,
                                                      left_block,
                                                      right_block,
                                                      next_shiftboss_index_to_schedule,
                                                      shiftboss_index_for_nested_loops_join);
 }

 void PolicyEnforcerDistributed::initiateQueryInShiftboss(QueryHandle *query_handle) {
   S::QueryInitiateMessage proto;
   proto.set_query_id(query_handle->query_id());
   proto.mutable_catalog_database_cache()->MergeFrom(query_handle->getCatalogDatabaseCacheProto());
   proto.mutable_query_context()->MergeFrom(query_handle->getQueryContextProto());

   const size_t proto_length = proto.ByteSize();
   char *proto_bytes = static_cast<char*>(malloc(proto_length));
   CHECK(proto.SerializeToArray(proto_bytes, proto_length));

   TaggedMessage message(static_cast<const void*>(proto_bytes),
                         proto_length,
                         kQueryInitiateMessage);
   free(proto_bytes);

   // TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
   tmb::Address shiftboss_addresses;
   for (std::size_t i = 0; i < shiftboss_directory_->size(); ++i) {
     shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
   }

   DLOG(INFO) << "PolicyEnforcerDistributed sent QueryInitiateMessage to all Shiftbosses";
   QueryExecutionUtil::BroadcastMessage(foreman_client_id_,
                                        shiftboss_addresses,
                                        move(message),
                                        bus_);
 }

 void PolicyEnforcerDistributed::onQueryCompletion(QueryManagerBase *query_manager) {
   const QueryHandle *query_handle = query_manager->query_handle();

   const CatalogRelation *query_result_relation = query_handle->getQueryResultRelation();
   const tmb::client_id cli_id = query_handle->getClientId();
   const std::size_t query_id = query_handle->query_id();

   if (FLAGS_visualize_execution_dag && hasProfilingResults(query_id)) {
     ExecutionDAGVisualizer* dag_visualizer = query_manager->dag_visualizer();
     dag_visualizer->bindProfilingStats(getProfilingResults(query_id));

     std::ostringstream dot_filename;
     dot_filename << query_id << ".dot";

     FILE *fp = std::fopen(dot_filename.str().c_str(), "w");
     CHECK_NOTNULL(fp);
     const std::string dot_file_content(dag_visualizer->toDOT());
     const std::size_t dot_file_length = dot_file_content.length();

     CHECK_EQ(dot_file_length,
              std::fwrite(dot_file_content.c_str(), sizeof(char), dot_file_length, fp));

     std::fclose(fp);
   }

   // TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
   tmb::Address shiftboss_addresses;
   for (std::size_t i = 0; i < shiftboss_directory_->size(); ++i) {
     shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
   }


   if (query_result_relation) {
     const QueryHandle::AnalyzeQueryInfo *analyze_query_info = query_handle->analyze_query_info();
     if (analyze_query_info) {
       processAnalyzeQueryResult(cli_id, query_result_relation, analyze_query_info);
     } else {
       S::QueryExecutionSuccessMessage proto;
       proto.mutable_result_relation()->MergeFrom(query_result_relation->getProto());

       const size_t proto_length = proto.ByteSize();
       char *proto_bytes = static_cast<char*>(malloc(proto_length));
       CHECK(proto.SerializeToArray(proto_bytes, proto_length));

       TaggedMessage message(static_cast<const void*>(proto_bytes), proto_length, kQueryExecutionSuccessMessage);
       free(proto_bytes);

       // Notify the CLI regarding the query result.
       DLOG(INFO)
         << "PolicyEnforcerDistributed sent QueryExecutionSuccessMessage to DistributedCLI with Client "
         << cli_id;
       const MessageBus::SendStatus send_status =
           QueryExecutionUtil::SendTMBMessage(bus_, foreman_client_id_, cli_id, move(message));
       CHECK(send_status == MessageBus::SendStatus::kOK);
     }
   } else {
     if (query_processor_) {
       query_processor_->saveCatalog();
     }

     // Notify the CLI query execution successfully.
     DLOG(INFO)
       << "PolicyEnforcerDistributed sent QueryExecutionSuccessMessage to DistributedCLI with Client "
       << cli_id;
     const MessageBus::SendStatus send_status =
         QueryExecutionUtil::SendTMBMessage(bus_, foreman_client_id_, cli_id,
                                            TaggedMessage(kQueryExecutionSuccessMessage));
     CHECK(send_status == MessageBus::SendStatus::kOK);
   }

   // Clean up query execution states, i.e., QueryContext, in Shiftbosses.
   S::QueryTeardownMessage proto;
   proto.set_query_id(query_id);

   const size_t proto_length = proto.ByteSize();
   char *proto_bytes = static_cast<char*>(malloc(proto_length));
   CHECK(proto.SerializeToArray(proto_bytes, proto_length));

   TaggedMessage message(static_cast<const void*>(proto_bytes), proto_length, kQueryTeardownMessage);
   free(proto_bytes);

   DLOG(INFO) << "PolicyEnforcerDistributed sent QueryTeardownMessage to all Shiftbosses";
   QueryExecutionUtil::BroadcastMessage(foreman_client_id_, shiftboss_addresses, move(message), bus_);
 }

 void PolicyEnforcerDistributed::processAnalyzeQueryResult(const tmb::client_id cli_id,
                                                           const CatalogRelation *query_result_relation,
                                                           const QueryHandle::AnalyzeQueryInfo *analyze_query_info) {
   const relation_id rel_id = analyze_query_info->rel_id;
   CatalogRelation *mutable_relation =
       static_cast<CatalogDatabase*>(catalog_database_)->getRelationByIdMutable(rel_id);
   CatalogRelationStatistics *mutable_stat = mutable_relation->getStatisticsMutable();

   const auto analyze_query_result = [this, &query_result_relation]() {
     const vector<block_id> blocks = query_result_relation->getBlocksSnapshot();
     DCHECK_EQ(1u, blocks.size());

     vector<TypedValue> query_result;
     {
       BlockReference block = storage_manager_->getBlock(blocks.front(), *query_result_relation);
       const TupleStorageSubBlock &tuple_store = block->getTupleStorageSubBlock();
       DCHECK_EQ(1, tuple_store.numTuples());

       const std::size_t num_columns = tuple_store.getRelation().size();
       if (tuple_store.isPacked()) {
         for (std::size_t i = 0; i < num_columns; ++i) {
           query_result.emplace_back(tuple_store.getAttributeValueTyped(0, i));
         }
       } else {
         std::unique_ptr<TupleIdSequence> existence_map(tuple_store.getExistenceMap());
         for (std::size_t i = 0; i < num_columns; ++i) {
           query_result.emplace_back(
               tuple_store.getAttributeValueTyped(*existence_map->begin(), i));
         }
       }
     }

     // Clean up the query result relation.
     catalog_database_->dropRelationById(query_result_relation->getID());

     return query_result;
   }();

   if (analyze_query_info->is_analyze_attribute_query) {
     const attribute_id attr_id = analyze_query_info->attr_id;

     auto cit = analyze_query_result.begin();
     DCHECK_EQ(TypeID::kLong, cit->getTypeID());
     mutable_stat->setNumDistinctValues(attr_id, cit->getLiteral<std::int64_t>());

     if (analyze_query_info->is_min_applicable) {
       ++cit;
       mutable_stat->setMinValue(attr_id, *cit);
     }

     if (analyze_query_info->is_max_applicable) {
       ++cit;
       mutable_stat->setMaxValue(attr_id, *cit);
     }
   } else {
     completed_analyze_relations_[cli_id].push_back(rel_id);

     DCHECK_EQ(1u, analyze_query_result.size());
     const TypedValue &num_tuples = analyze_query_result.front();
     DCHECK_EQ(TypeID::kLong, num_tuples.getTypeID());
     mutable_stat->setNumTuples(num_tuples.getLiteral<std::int64_t>());

     mutable_stat->setExactness(true);

     if (completed_analyze_relations_[cli_id].size() == analyze_query_info->num_relations) {
       query_processor_->markCatalogAltered();
       query_processor_->saveCatalog();

       ostringstream analyze_command_response;
       for (const relation_id rel_id : completed_analyze_relations_[cli_id]) {
         analyze_command_response << "Analyzing " << catalog_database_->getRelationSchemaById(rel_id).getName()
                                  << " ... done\n";
       }

       S::CommandResponseMessage proto;
       proto.set_command_response(analyze_command_response.str());

       const size_t proto_length = proto.ByteSize();
       char *proto_bytes = static_cast<char*>(malloc(proto_length));
       CHECK(proto.SerializeToArray(proto_bytes, proto_length));

       TaggedMessage message(static_cast<const void*>(proto_bytes), proto_length, kCommandResponseMessage);
       free(proto_bytes);

       DLOG(INFO) << "PolicyEnforcerDistributed sent CommandResponseMessage to DistributedCLI with Client " << cli_id;
       const MessageBus::SendStatus send_status =
           QueryExecutionUtil::SendTMBMessage(bus_, foreman_client_id_, cli_id, move(message));
       CHECK(send_status == MessageBus::SendStatus::kOK);

       completed_analyze_relations_.erase(cli_id);
     }
   }
 }

 }  // namespace quickstep
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you 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.
	**/

	#include "query_execution/PolicyEnforcerDistributed.hpp"

	#include <cstddef>
	#include <cstdio>
	#include <cstdlib>
	#include <memory>
	#include <queue>
	#include <sstream>
	#include <string>
	#include <unordered_map>
	#include <utility>
	#include <vector>

	#include "catalog/Catalog.pb.h"
	#include "catalog/CatalogDatabase.hpp"
	#include "catalog/CatalogRelation.hpp"
	#include "catalog/CatalogRelationSchema.hpp"
	#include "catalog/CatalogRelationStatistics.hpp"
	#include "query_execution/QueryContext.pb.h"
	#include "query_execution/QueryExecutionMessages.pb.h"
	#include "query_execution/QueryExecutionState.hpp"
	#include "query_execution/QueryExecutionTypedefs.hpp"
	#include "query_execution/QueryExecutionUtil.hpp"
	#include "query_execution/QueryManagerBase.hpp"
	#include "query_execution/QueryManagerDistributed.hpp"
	#include "query_optimizer/QueryHandle.hpp"
	#include "query_optimizer/QueryProcessor.hpp"
	#include "storage/StorageBlock.hpp"
	#include "storage/StorageBlockInfo.hpp"
	#include "storage/StorageManager.hpp"
	#include "storage/TupleIdSequence.hpp"
	#include "storage/TupleStorageSubBlock.hpp"
	#include "types/TypedValue.hpp"
	#include "utility/ExecutionDAGVisualizer.hpp"

	#include "gflags/gflags.h"
	#include "glog/logging.h"

	#include "tmb/address.h"
	#include "tmb/id_typedefs.h"
	#include "tmb/message_bus.h"
	#include "tmb/tagged_message.h"

	using std::free;
	using std::malloc;
	using std::move;
	using std::ostringstream;
	using std::size_t;
	using std::string;
	using std::unique_ptr;
	using std::vector;

	using tmb::MessageBus;
	using tmb::TaggedMessage;

	namespace quickstep {

	DECLARE_bool(visualize_execution_dag);

	namespace S = serialization;

	DEFINE_uint64(max_msgs_per_dispatch_round, 20, "Maximum number of messages that"
	" can be allocated in a single round of dispatch of messages to"
	" the workers.");

	void PolicyEnforcerDistributed::getWorkOrderProtoMessages(
	vector<unique_ptr<S::WorkOrderMessage>> *work_order_proto_messages) {
	// Iterate over admitted queries until either there are no more
	// messages available, or the maximum number of messages have
	// been collected.
	DCHECK(work_order_proto_messages->empty());
	// TODO(harshad) - Make this function generic enough so that it
	// works well when multiple queries are getting executed.
	if (admitted_queries_.empty()) {
	if (waiting_queries_.empty()) {
	return;
	}

	// Admit the earliest waiting query.
	QueryHandle *new_query = waiting_queries_.front();
	waiting_queries_.pop();
	admitQuery(new_query);
	}

	const std::size_t per_query_share =
	FLAGS_max_msgs_per_dispatch_round / admitted_queries_.size();
	DCHECK_GT(per_query_share, 0u);

	vector<std::size_t> finished_queries_ids;

	for (const auto &admitted_query_info : admitted_queries_) {
	QueryManagerBase *curr_query_manager = admitted_query_info.second.get();
	DCHECK(curr_query_manager != nullptr);
	std::size_t messages_collected_curr_query = 0;
	while (messages_collected_curr_query < per_query_share) {
	unique_ptr<S::WorkOrderMessage> next_work_order_message(
	static_cast<QueryManagerDistributed*>(curr_query_manager)->getNextWorkOrderMessage());
	if (next_work_order_message) {
	++messages_collected_curr_query;
	work_order_proto_messages->push_back(move(next_work_order_message));
	} else {
	// No more work ordes from the current query at this time.
	// Check if the query's execution is over.
	if (curr_query_manager->getQueryExecutionState().hasQueryExecutionFinished()) {
	// If the query has been executed, remove it.
	finished_queries_ids.push_back(admitted_query_info.first);
	}
	break;
	}
	}
	}
	for (const std::size_t finished_qid : finished_queries_ids) {
	onQueryCompletion(admitted_queries_[finished_qid].get());
	removeQuery(finished_qid);
	}
	}

	bool PolicyEnforcerDistributed::admitQuery(QueryHandle *query_handle) {
	const std::size_t num_shiftbosses = shiftboss_directory_->size();
	// We only allow the number of concurrent queries that equals to
	// that of Shiftbosses.
	if (admitted_queries_.size() > num_shiftbosses) {
	// This query will have to wait.
	waiting_queries_.push(query_handle);
	return false;
	}

	const std::size_t query_id = query_handle->query_id();
	if (admitted_queries_.find(query_id) != admitted_queries_.end()) {
	LOG(ERROR) << "Query with the same ID " << query_id << " exists";
	return false;
	}

	// Ok to admit the query.
	// NOTE(zuyu): Should call before constructing a 'QueryManager'.
	// Otherwise, an InitiateRebuildMessage may be sent before 'QueryContext'
	// initializes.
	initiateQueryInShiftboss(query_handle);

	tmb::Address shiftboss_addresses;
	for (std::size_t i = 0; i < num_shiftbosses; ++i) {
	shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
	}

	// Query with the same ID not present, ok to admit.
	admitted_queries_.emplace(query_id,
	std::make_unique<QueryManagerDistributed>(
	query_handle, foreman_client_id_, num_shiftbosses, move(shiftboss_addresses), bus_));
	return true;
	}

	void PolicyEnforcerDistributed::processInitiateRebuildResponseMessage(const tmb::TaggedMessage &tagged_message) {
	S::InitiateRebuildResponseMessage proto;
	CHECK(proto.ParseFromArray(tagged_message.message(), tagged_message.message_bytes()));

	const std::size_t query_id = proto.query_id();
	DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());

	QueryManagerDistributed query_manager = static_cast<QueryManagerDistributed>(admitted_queries_[query_id].get());

	const std::size_t num_rebuild_work_orders = proto.num_rebuild_work_orders();
	const size_t shiftboss_index = proto.shiftboss_index();
	query_manager->processInitiateRebuildResponseMessage(
	proto.operator_index(), num_rebuild_work_orders, shiftboss_index);
	shiftboss_directory_->addNumQueuedWorkOrders(shiftboss_index, num_rebuild_work_orders);

	if (query_manager->getQueryExecutionState().hasQueryExecutionFinished()) {
	onQueryCompletion(query_manager);
	removeQuery(query_id);
	}
	}

	void PolicyEnforcerDistributed::getShiftbossIndexForAggregation(
	const std::size_t query_id,
	const QueryContext::aggregation_state_id aggr_state_index,
	const partition_id part_id,
	const vector<QueryContext::lip_filter_id> &lip_filter_indexes,
	const BlockLocator &block_locator,
	const block_id block,
	const std::size_t next_shiftboss_index_to_schedule,
	std::size_t *shiftboss_index) {
	DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
	QueryManagerDistributed query_manager = static_cast<QueryManagerDistributed>(admitted_queries_[query_id].get());
	query_manager->getShiftbossIndexForAggregation(aggr_state_index,
	part_id,
	lip_filter_indexes,
	block_locator,
	block,
	next_shiftboss_index_to_schedule,
	shiftboss_index);
	}

	void PolicyEnforcerDistributed::getShiftbossIndexForHashJoin(
	const std::size_t query_id,
	const QueryContext::join_hash_table_id join_hash_table_index,
	const partition_id part_id,
	const vector<QueryContext::lip_filter_id> &lip_filter_indexes,
	const BlockLocator &block_locator,
	const block_id block,
	const std::size_t next_shiftboss_index_to_schedule,
	std::size_t *shiftboss_index) {
	DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
	QueryManagerDistributed query_manager = static_cast<QueryManagerDistributed>(admitted_queries_[query_id].get());
	query_manager->getShiftbossIndexForHashJoin(join_hash_table_index,
	part_id,
	lip_filter_indexes,
	block_locator,
	block,
	next_shiftboss_index_to_schedule,
	shiftboss_index);
	}

	void PolicyEnforcerDistributed::getShiftbossIndexForLip(
	const std::size_t query_id,
	const vector<QueryContext::lip_filter_id> &lip_filter_indexes,
	const BlockLocator &block_locator,
	const block_id block,
	const std::size_t next_shiftboss_index_to_schedule,
	std::size_t *shiftboss_index) {
	DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
	QueryManagerDistributed query_manager = static_cast<QueryManagerDistributed>(admitted_queries_[query_id].get());
	query_manager->getShiftbossIndexForLip(lip_filter_indexes,
	block_locator,
	block,
	next_shiftboss_index_to_schedule,
	shiftboss_index);
	}

	void PolicyEnforcerDistributed::getShiftbossIndexForNestedLoopsJoin(
	const std::size_t query_id,
	const std::size_t nested_loops_join_index,
	const partition_id part_id,
	const BlockLocator &block_locator,
	const block_id left_block,
	const block_id right_block,
	const std::size_t next_shiftboss_index_to_schedule,
	std::size_t *shiftboss_index_for_nested_loops_join) {
	DCHECK(admitted_queries_.find(query_id) != admitted_queries_.end());
	QueryManagerDistributed query_manager = static_cast<QueryManagerDistributed>(admitted_queries_[query_id].get());
	query_manager->getShiftbossIndexForNestedLoopsJoin(nested_loops_join_index,
	part_id,
	block_locator,
	left_block,
	right_block,
	next_shiftboss_index_to_schedule,
	shiftboss_index_for_nested_loops_join);
	}

	void PolicyEnforcerDistributed::initiateQueryInShiftboss(QueryHandle *query_handle) {
	S::QueryInitiateMessage proto;
	proto.set_query_id(query_handle->query_id());
	proto.mutable_catalog_database_cache()->MergeFrom(query_handle->getCatalogDatabaseCacheProto());
	proto.mutable_query_context()->MergeFrom(query_handle->getQueryContextProto());

	const size_t proto_length = proto.ByteSize();
	char proto_bytes = static_cast<char>(malloc(proto_length));
	CHECK(proto.SerializeToArray(proto_bytes, proto_length));

	TaggedMessage message(static_cast<const void*>(proto_bytes),
	proto_length,
	kQueryInitiateMessage);
	free(proto_bytes);

	// TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
	tmb::Address shiftboss_addresses;
	for (std::size_t i = 0; i < shiftboss_directory_->size(); ++i) {
	shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
	}

	DLOG(INFO) << "PolicyEnforcerDistributed sent QueryInitiateMessage to all Shiftbosses";
	QueryExecutionUtil::BroadcastMessage(foreman_client_id_,
	shiftboss_addresses,
	move(message),
	bus_);
	}

	void PolicyEnforcerDistributed::onQueryCompletion(QueryManagerBase *query_manager) {
	const QueryHandle *query_handle = query_manager->query_handle();

	const CatalogRelation *query_result_relation = query_handle->getQueryResultRelation();
	const tmb::client_id cli_id = query_handle->getClientId();
	const std::size_t query_id = query_handle->query_id();

	if (FLAGS_visualize_execution_dag && hasProfilingResults(query_id)) {
	ExecutionDAGVisualizer* dag_visualizer = query_manager->dag_visualizer();
	dag_visualizer->bindProfilingStats(getProfilingResults(query_id));

	std::ostringstream dot_filename;
	dot_filename << query_id << ".dot";

	FILE *fp = std::fopen(dot_filename.str().c_str(), "w");
	CHECK_NOTNULL(fp);
	const std::string dot_file_content(dag_visualizer->toDOT());
	const std::size_t dot_file_length = dot_file_content.length();

	CHECK_EQ(dot_file_length,
	std::fwrite(dot_file_content.c_str(), sizeof(char), dot_file_length, fp));

	std::fclose(fp);
	}

	// TODO(quickstep-team): Dynamically scale-up/down Shiftbosses.
	tmb::Address shiftboss_addresses;
	for (std::size_t i = 0; i < shiftboss_directory_->size(); ++i) {
	shiftboss_addresses.AddRecipient(shiftboss_directory_->getClientId(i));
	}


	if (query_result_relation) {
	const QueryHandle::AnalyzeQueryInfo *analyze_query_info = query_handle->analyze_query_info();
	if (analyze_query_info) {
	processAnalyzeQueryResult(cli_id, query_result_relation, analyze_query_info);
	} else {
	S::QueryExecutionSuccessMessage proto;
	proto.mutable_result_relation()->MergeFrom(query_result_relation->getProto());

	const size_t proto_length = proto.ByteSize();
	char proto_bytes = static_cast<char>(malloc(proto_length));
	CHECK(proto.SerializeToArray(proto_bytes, proto_length));

	TaggedMessage message(static_cast<const void*>(proto_bytes), proto_length, kQueryExecutionSuccessMessage);
	free(proto_bytes);

	// Notify the CLI regarding the query result.
	DLOG(INFO)
	<< "PolicyEnforcerDistributed sent QueryExecutionSuccessMessage to DistributedCLI with Client "
	<< cli_id;
	const MessageBus::SendStatus send_status =
	QueryExecutionUtil::SendTMBMessage(bus_, foreman_client_id_, cli_id, move(message));
	CHECK(send_status == MessageBus::SendStatus::kOK);
	}
	} else {
	if (query_processor_) {
	query_processor_->saveCatalog();
	}

	// Notify the CLI query execution successfully.
	DLOG(INFO)
	<< "PolicyEnforcerDistributed sent QueryExecutionSuccessMessage to DistributedCLI with Client "
	<< cli_id;
	const MessageBus::SendStatus send_status =
	QueryExecutionUtil::SendTMBMessage(bus_, foreman_client_id_, cli_id,
	TaggedMessage(kQueryExecutionSuccessMessage));
	CHECK(send_status == MessageBus::SendStatus::kOK);
	}

	// Clean up query execution states, i.e., QueryContext, in Shiftbosses.
	S::QueryTeardownMessage proto;
	proto.set_query_id(query_id);

	const size_t proto_length = proto.ByteSize();
	char proto_bytes = static_cast<char>(malloc(proto_length));
	CHECK(proto.SerializeToArray(proto_bytes, proto_length));

	TaggedMessage message(static_cast<const void*>(proto_bytes), proto_length, kQueryTeardownMessage);
	free(proto_bytes);

	DLOG(INFO) << "PolicyEnforcerDistributed sent QueryTeardownMessage to all Shiftbosses";
	QueryExecutionUtil::BroadcastMessage(foreman_client_id_, shiftboss_addresses, move(message), bus_);
	}

	void PolicyEnforcerDistributed::processAnalyzeQueryResult(const tmb::client_id cli_id,
	const CatalogRelation *query_result_relation,
	const QueryHandle::AnalyzeQueryInfo *analyze_query_info) {
	const relation_id rel_id = analyze_query_info->rel_id;
	CatalogRelation *mutable_relation =
	static_cast<CatalogDatabase*>(catalog_database_)->getRelationByIdMutable(rel_id);
	CatalogRelationStatistics *mutable_stat = mutable_relation->getStatisticsMutable();

	const auto analyze_query_result = [this, &query_result_relation]() {
	const vector<block_id> blocks = query_result_relation->getBlocksSnapshot();
	DCHECK_EQ(1u, blocks.size());

	vector<TypedValue> query_result;
	{
	BlockReference block = storage_manager_->getBlock(blocks.front(), *query_result_relation);
	const TupleStorageSubBlock &tuple_store = block->getTupleStorageSubBlock();
	DCHECK_EQ(1, tuple_store.numTuples());

	const std::size_t num_columns = tuple_store.getRelation().size();
	if (tuple_store.isPacked()) {
	for (std::size_t i = 0; i < num_columns; ++i) {
	query_result.emplace_back(tuple_store.getAttributeValueTyped(0, i));
	}
	} else {
	std::unique_ptr<TupleIdSequence> existence_map(tuple_store.getExistenceMap());
	for (std::size_t i = 0; i < num_columns; ++i) {
	query_result.emplace_back(
	tuple_store.getAttributeValueTyped(*existence_map->begin(), i));
	}
	}
	}

	// Clean up the query result relation.
	catalog_database_->dropRelationById(query_result_relation->getID());

	return query_result;
	}();

	if (analyze_query_info->is_analyze_attribute_query) {
	const attribute_id attr_id = analyze_query_info->attr_id;

	auto cit = analyze_query_result.begin();
	DCHECK_EQ(TypeID::kLong, cit->getTypeID());
	mutable_stat->setNumDistinctValues(attr_id, cit->getLiteral<std::int64_t>());

	if (analyze_query_info->is_min_applicable) {
	++cit;
	mutable_stat->setMinValue(attr_id, *cit);
	}

	if (analyze_query_info->is_max_applicable) {
	++cit;
	mutable_stat->setMaxValue(attr_id, *cit);
	}
	} else {
	completed_analyze_relations_[cli_id].push_back(rel_id);

	DCHECK_EQ(1u, analyze_query_result.size());
	const TypedValue &num_tuples = analyze_query_result.front();
	DCHECK_EQ(TypeID::kLong, num_tuples.getTypeID());
	mutable_stat->setNumTuples(num_tuples.getLiteral<std::int64_t>());

	mutable_stat->setExactness(true);

	if (completed_analyze_relations_[cli_id].size() == analyze_query_info->num_relations) {
	query_processor_->markCatalogAltered();
	query_processor_->saveCatalog();

	ostringstream analyze_command_response;
	for (const relation_id rel_id : completed_analyze_relations_[cli_id]) {
	analyze_command_response << "Analyzing " << catalog_database_->getRelationSchemaById(rel_id).getName()
	<< " ... done\n";
	}

	S::CommandResponseMessage proto;
	proto.set_command_response(analyze_command_response.str());

	const size_t proto_length = proto.ByteSize();
	char proto_bytes = static_cast<char>(malloc(proto_length));
	CHECK(proto.SerializeToArray(proto_bytes, proto_length));

	TaggedMessage message(static_cast<const void*>(proto_bytes), proto_length, kCommandResponseMessage);
	free(proto_bytes);

	DLOG(INFO) << "PolicyEnforcerDistributed sent CommandResponseMessage to DistributedCLI with Client " << cli_id;
	const MessageBus::SendStatus send_status =
	QueryExecutionUtil::SendTMBMessage(bus_, foreman_client_id_, cli_id, move(message));
	CHECK(send_status == MessageBus::SendStatus::kOK);

	completed_analyze_relations_.erase(cli_id);
	}
	}
	}

	} // namespace quickstep