query_execution/WorkOrdersContainer.hpp - 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.
  **/

 #ifndef QUICKSTEP_QUERY_EXECUTION_WORKORDERS_CONTAINER_HPP_
 #define QUICKSTEP_QUERY_EXECUTION_WORKORDERS_CONTAINER_HPP_

 #include <cstddef>
 #include <list>
 #include <memory>
 #include <queue>
 #include <vector>

 #include "query_execution/WorkOrderSelectionPolicy.hpp"
 #include "relational_operators/WorkOrder.hpp"
 #include "utility/Macros.hpp"
 #include "utility/PtrVector.hpp"

 #include "glog/logging.h"

 namespace quickstep {

 /** \addtogroup QueryExecution
  *  @{
  */

  /**
   * @brief A container to hold the WorkOrders for a given query.
   * @note This container stays alive during the lifetime of the query.
   * @note The NUMA node indexing is assumed to start from 0.
   **/
 class WorkOrdersContainer {
  public:
   /**
    * @brief Constructor
    *
    * @param num_operators Number of operators in the query DAG.
    * @param num_numa_nodes Number of NUMA nodes in the system.
    **/
   WorkOrdersContainer(const std::size_t num_operators,
                       const std::size_t num_numa_nodes)
       : num_operators_(num_operators), num_numa_nodes_(num_numa_nodes) {
     DEBUG_ASSERT(num_operators != 0);
     for (std::size_t op = 0; op < num_operators; ++op) {
       normal_workorders_.push_back(
           new OperatorWorkOrdersContainer(num_numa_nodes_));
       rebuild_workorders_.push_back(
           new OperatorWorkOrdersContainer(num_numa_nodes_));
     }
   }

   /**
    * @brief Destructor.
    *
    * @note If the query is executed normally, we should never encounter a
    *       situation where at the time of deletion the WorkOrdersContainer has
    *       pending WorkOrders.
    **/
   ~WorkOrdersContainer();

   /**
    * @brief Check if there are some pending WorkOrders for the given operator.
    *
    * @param operator_index Index of the operator.
    *
    * @return If there are pending WorkOrders.
    **/
   inline bool hasNormalWorkOrder(const std::size_t operator_index) const {
     DCHECK_LT(operator_index, num_operators_);
     return normal_workorders_[operator_index].hasWorkOrder();
   }

   /**
    * @brief Check if there are some pending normal (i.e. non-rebuild) WorkOrders
    *        for the given operator which prefer the specified NUMA node.
    *
    * @param operator_index Index of the operator.
    * @param numa_node_id The NUMA node.
    *
    * @return If there are pending WorkOrders for the given operator which prefer
    *         numa_node_id.
    **/
   inline bool hasNormalWorkOrderForNUMANode(
       const std::size_t operator_index, const int numa_node_id) const {
     DCHECK_LT(operator_index, num_operators_);
     DCHECK_GE(numa_node_id, 0);
     DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
     return normal_workorders_[operator_index].hasWorkOrderForNUMANode(
         numa_node_id);
   }

   /**
    * @brief Check if there are some pending rebuild WorkOrders for the given
    *        operator.
    *
    * @param operator_index Index of the operator.
    *
    * @return If there are pending rebuild WorkOrders.
    **/
   inline bool hasRebuildWorkOrder(const std::size_t operator_index) const {
     DCHECK_LT(operator_index, num_operators_);
     return rebuild_workorders_[operator_index].hasWorkOrder();
   }

   /**
    * @brief Check if there are some pending rebuild WorkOrders for the given
    *        operator which prefer the specified NUMA node.
    *
    * @param operator_index Index of the operator.
    * @param numa_node_id The NUMA node.
    *
    * @return If there are pending rebuild WorkOrders for the given operator which
    *         prefer numa_node_id.
    **/
   inline bool hasRebuildWorkOrderForNUMANode(
       const std::size_t operator_index, const int numa_node_id) const {
     DCHECK_LT(operator_index, num_operators_);
     DCHECK_GE(numa_node_id, 0);
     DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
     return rebuild_workorders_[operator_index].hasWorkOrderForNUMANode(
         numa_node_id);
   }

   /**
    * @brief Get a normal (non-rebuild) WorkOrder for a given operator which
    *        prefer the given NUMA node.
    *
    * @param operator_index The index of the operator.
    * @param numa_node_id The NUMA node.
    *
    * @return A WorkOrder which prefers numa_node_id. If no such WorkOrder is
    *         available, return nullptr. The caller is responsible for taking the
    *         ownership.
    **/
   WorkOrder* getNormalWorkOrderForNUMANode(const std::size_t operator_index,
                                            const int numa_node_id) {
     DCHECK_LT(operator_index, num_operators_);
     DCHECK_GE(numa_node_id, 0);
     DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
     return normal_workorders_[operator_index].getWorkOrderForNUMANode(
         numa_node_id);
   }

   /**
    * @brief Get a normal (non-rebuild) WorkOrder for a given operator.
    *
    * @param operator_index The index of the operator.
    * @param prefer_single_NUMA_node If true, first try to get workorders which
    *        prefer exactly one NUMA node over workorders which list more than
    *        one NUMA node as their preference.
    *
    * @return A WorkOrder. If no WorkOrder is available, returns nullptr. The
    *         caller is responsible for taking the ownership.
    **/
   WorkOrder* getNormalWorkOrder(const std::size_t operator_index,
                                 const bool prefer_single_NUMA_node = true) {
     DCHECK_LT(operator_index, num_operators_);
     return normal_workorders_[operator_index].getWorkOrder(
         prefer_single_NUMA_node);
   }

   WorkOrder* getNextWorkOrder(std::size_t *operator_index, bool *is_rebuild) {
     if (rebuild_work_orders_policy_.hasWorkOrder()) {
       *operator_index = rebuild_work_orders_policy_.getOperatorIndexForNextWorkOrder();
       *is_rebuild = true;
       return rebuild_workorders_[*operator_index].getWorkOrder();
     }

     if (normal_work_orders_policy_.hasWorkOrder()) {
       *operator_index = normal_work_orders_policy_.getOperatorIndexForNextWorkOrder();
       *is_rebuild = false;
       return normal_workorders_[*operator_index].getWorkOrder();
     }

     return nullptr;
   }

   /**
    * @brief Get a rebuild WorkOrder for a given operator whch prefer the
    *        specified NUMA node.
    *
    * @param operator_index The index of the operator.
    * @param numa_node_id The NUMA node.
    *
    * @return A WorkOrder that prefers numa_node_id. If no such WorkOrder is
    *         available, return nullptr. The caller is responsible for taking the
    *         ownership.
    **/
   WorkOrder* getRebuildWorkOrderForNUMANode(const std::size_t operator_index,
                                             const int numa_node_id) {
     DCHECK_LT(operator_index, num_operators_);
     DCHECK_GE(numa_node_id, 0);
     DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
     return rebuild_workorders_[operator_index].getWorkOrderForNUMANode(
         numa_node_id);
   }

   /**
    * @brief Get a rebuild WorkOrder for the given operator
    *
    * @param operator_index The index of the operator.
    * @param prefer_single_NUMA_node If true, first try to get workorders which
    *        prefer exactly one NUMA node over workorders which list more than
    *        one NUMA node as their preference.
    *
    * @return A WorkOrder. If no WorkOrder is available, returns nullptr. The
    *         caller is responsible for taking the ownership.
    **/
   WorkOrder* getRebuildWorkOrder(const std::size_t operator_index,
                                  const bool prefer_single_NUMA_node = true) {
     DCHECK_LT(operator_index, num_operators_);
     return rebuild_workorders_[operator_index].getWorkOrder(
         prefer_single_NUMA_node);
   }

   /**
    * @brief Add a normal (non-rebuild) WorkOrder generated from a given
    *        operator.
    *
    * @note Take the ownership of \p workorder.
    * @note The workorder to be added contains information about its preferred
    *       NUMA nodes. This information is used to insert the WorkOrder
    *       appropriately.
    *
    * @param workorder A pointer to the WorkOrder to be added.
    * @param operator_index The index of the operator in the query DAG.
    **/
   void addNormalWorkOrder(WorkOrder *workorder, const std::size_t operator_index) {
     DCHECK(workorder != nullptr);
     DCHECK_LT(operator_index, num_operators_);
     normal_workorders_[operator_index].addWorkOrder(workorder);
     normal_work_orders_policy_.addWorkOrder(operator_index);
   }

   /**
    * @brief Add a rebuild WorkOrder generated from a given operator.
    *
    * @note Take the ownership of \p workorder.
    * @note The workorder to be added contains information about its preferred
    *       NUMA nodes. This information is used to insert the WorkOrder
    *       appropriately.
    *
    * @param workorder A pointer to the WorkOrder to be added.
    * @param operator_index The index of the operator in the query DAG.
    **/
   void addRebuildWorkOrder(WorkOrder *workorder,
                            const std::size_t operator_index) {
     DCHECK(workorder != nullptr);
     DCHECK_LT(operator_index, num_operators_);
     rebuild_workorders_[operator_index].addWorkOrder(workorder);
     rebuild_work_orders_policy_.addWorkOrder(operator_index);
   }

   /**
    * @brief Get the number of pending normal WorkOrders for a given operator
    *        which prefer the specified NUMA node.
    *
    * @param operator_index The index of the operator.
    * @param numa_node_id The NUMA node.
    *
    * @return The number of pending WorkOrders which prefer numa_node_id.
    **/
   inline std::size_t getNumNormalWorkOrdersForNUMANode(
       const std::size_t operator_index, const int numa_node_id) const {
     DCHECK_LT(operator_index, num_operators_);
     DCHECK_GE(numa_node_id, 0);
     DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
     return normal_workorders_[operator_index].getNumWorkOrdersForNUMANode(
         numa_node_id);
   }

   /**
    * @brief Get the number of all pending normal (i.e. non-rebuild) WorkOrders
    *        for a given operator.
    *
    * @param operator_index The index of the operator.
    *
    * @return The number of pending normal WorkOrders.
    **/
   inline std::size_t getNumNormalWorkOrders(
       const std::size_t operator_index) const {
     DCHECK_LT(operator_index, num_operators_);
     return normal_workorders_[operator_index].getNumWorkOrders();
   }

   /**
    * @brief Get the number of pending rebuild WorkOrders for a given operator
    *        which prefer the specified NUMA node.
    *
    * @param operator_index The index of the operator.
    * @param numa_node_id The NUMA node.
    *
    * @return The number of pending WorkOrders which prefer numa_node_id.
    **/
   inline std::size_t getNumRebuildWorkOrdersForNUMANode(
       const std::size_t operator_index, const int numa_node_id) const {
     DCHECK_LT(operator_index, num_operators_);
     DCHECK_GE(numa_node_id, 0);
     DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
     return rebuild_workorders_[operator_index].getNumWorkOrdersForNUMANode(
         numa_node_id);
   }

   /**
    * @brief Get the number of all pending rebuild WorkOrders for a given
    *        operator.
    *
    * @param operator_index The index of the operator.
    *
    * @return The number of pending rebuild WorkOrders.
    **/
   inline std::size_t getNumRebuildWorkOrders(
       const std::size_t operator_index) const {
     DCHECK_LT(operator_index, num_operators_);
     return rebuild_workorders_[operator_index].getNumWorkOrders();
   }

   /**
    * @brief Get the total number of work orders for the given operator.
    *
    * @param operator_index The index of the operator.
    *
    * @return The total number of WorkOrders (normal + rebuild).
    **/
   inline std::size_t getNumTotalWorkOrders(
       const std::size_t operator_index) const {
     return getNumNormalWorkOrders(operator_index) +
            getNumRebuildWorkOrders(operator_index);
   }

  private:
   /**
    * @brief An internal queue-based container structure to hold the WorkOrders.
    **/
   class InternalQueueContainer {
    public:
     InternalQueueContainer() {
     }

     inline void addWorkOrder(WorkOrder *workorder) {
       workorders_.emplace(std::unique_ptr<WorkOrder>(workorder));
     }

     inline WorkOrder* getWorkOrder() {
       if (workorders_.empty()) {
         return nullptr;
       }

       WorkOrder *work_order = workorders_.front().release();
       workorders_.pop();
       return work_order;
     }

     inline bool hasWorkOrder() const {
       return !workorders_.empty();
     }

     inline std::size_t getNumWorkOrders() const {
       return workorders_.size();
     }

    private:
     std::queue<std::unique_ptr<WorkOrder>> workorders_;

     DISALLOW_COPY_AND_ASSIGN(InternalQueueContainer);
   };

   /**
    * @brief An internal list-based container structure to hold the WorkOrders.
    *
    * @note We use this class for WorkOrders whose inputs come from multiple NUMA
    *       nodes e.g. a HashJoinWorkOrder whose probe block and the hash table
    *       may reside on two different NUMA nodes.
    **/
   class InternalListContainer {
    public:
     InternalListContainer() {
     }

     inline void addWorkOrder(WorkOrder *workorder) {
       workorders_.emplace_back(std::unique_ptr<WorkOrder>(workorder));
     }

     inline WorkOrder* getWorkOrder() {
       if (workorders_.empty()) {
         return nullptr;
       }

       WorkOrder *work_order = workorders_.front().release();
       workorders_.pop_front();
       return work_order;
     }

     /**
      * @note This method has O(N) complexity.
      **/
     WorkOrder* getWorkOrderForNUMANode(const int numa_node);

     inline bool hasWorkOrder() const {
       return !workorders_.empty();
     }

     /**
      * @brief Check if numa_node is in the set of preferred NUMA nodes for at
      *        least one WorkOrder in this container.
      *
      * @note This method has O(N) complexity.
      **/
     bool hasWorkOrderForNUMANode(const int numa_node) const;

     inline std::size_t getNumWorkOrders() const {
       return workorders_.size();
     }

     /**
      * @brief Return the number of WorkOrders that list the numa_node as one of
      *        the preferred nodes of execution.
      **/
     std::size_t getNumWorkOrdersForNUMANode(const int numa_node) const;

    private:
     std::list<std::unique_ptr<WorkOrder>> workorders_;

     DISALLOW_COPY_AND_ASSIGN(InternalListContainer);
   };

   /**
    * @brief A container to hold all the WorkOrders generated by one operator.
    **/
   class OperatorWorkOrdersContainer {
    public:
     explicit OperatorWorkOrdersContainer(const std::size_t num_numa_nodes)
         : num_numa_nodes_(num_numa_nodes) {
       for (std::size_t numa_node = 0; numa_node < num_numa_nodes; ++numa_node) {
         single_numa_node_workorders_.push_back(new InternalQueueContainer());
       }
     }

     void addWorkOrder(WorkOrder *workorder);

     bool hasWorkOrderForNUMANode(const int numa_node_id) const {
       DCHECK_GE(numa_node_id, 0);
       DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
       return single_numa_node_workorders_[numa_node_id].hasWorkOrder() ||
              multiple_numa_nodes_workorders_.hasWorkOrderForNUMANode(
                  numa_node_id);
     }

     bool hasWorkOrder() const {
       if (!(numa_agnostic_workorders_.hasWorkOrder() ||
             multiple_numa_nodes_workorders_.hasWorkOrder())) {
         for (std::size_t i = 0; i < num_numa_nodes_; ++i) {
           if (hasWorkOrderForNUMANode(i)) {
             return true;
           }
         }
         return false;
       }
       return true;
     }

     std::size_t getNumWorkOrdersForNUMANode(
         const int numa_node_id) const {
       DCHECK_GE(numa_node_id, 0);
       DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
       return single_numa_node_workorders_[numa_node_id].getNumWorkOrders() +
              multiple_numa_nodes_workorders_.getNumWorkOrdersForNUMANode(
                  numa_node_id);
     }

     inline std::size_t getNumWorkOrders() const {
       std::size_t num_workorders = numa_agnostic_workorders_.getNumWorkOrders();
       // for each NUMA node involved ..
       for (PtrVector<InternalQueueContainer>::const_iterator it =
                single_numa_node_workorders_.begin();
            it != single_numa_node_workorders_.end();
            ++it) {
         // Add the number of workorders for the NUMA node.
         num_workorders += it->getNumWorkOrders();
       }
       // Add the number of workorders who have multiple NUMA nodes as input.
       num_workorders += multiple_numa_nodes_workorders_.getNumWorkOrders();
       return num_workorders;
     }

     WorkOrder* getWorkOrderForNUMANode(const int numa_node_id) {
       DCHECK_GE(numa_node_id, 0);
       DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
       WorkOrder *work_order = single_numa_node_workorders_[numa_node_id].getWorkOrder();
       if (work_order == nullptr) {
         work_order = multiple_numa_nodes_workorders_.getWorkOrderForNUMANode(
             numa_node_id);
       }
       return work_order;
     }

     WorkOrder* getWorkOrder(const bool prefer_single_NUMA_node = true);

    private:
     WorkOrder* getSingleNUMANodeWorkOrderHelper();

     const std::size_t num_numa_nodes_;

     // A container to store NUMA agnostic workorders.
     InternalQueueContainer numa_agnostic_workorders_;

     // A vector of containers to store workorders which prefer exactly one NUMA
     // node for execution.
     PtrVector<InternalQueueContainer> single_numa_node_workorders_;

     // A container to store workorders which prefer more than one NUMA node for
     // execution.
     InternalListContainer multiple_numa_nodes_workorders_;
     // Note that no workorder should be shared among the *_workorders_ structures.

     DISALLOW_COPY_AND_ASSIGN(OperatorWorkOrdersContainer);
   };

   const std::size_t num_operators_;
   const std::size_t num_numa_nodes_;

   PtrVector<OperatorWorkOrdersContainer> normal_workorders_;
   PtrVector<OperatorWorkOrdersContainer> rebuild_workorders_;

   LifoWorkOrderSelectionPolicy normal_work_orders_policy_;
   FifoWorkOrderSelectionPolicy rebuild_work_orders_policy_;

   DISALLOW_COPY_AND_ASSIGN(WorkOrdersContainer);
 };
 /** @} */

 }  // namespace quickstep


 #endif  // QUICKSTEP_QUERY_EXECUTION_WORKORDERS_CONTAINER_HPP_
	/**
	* 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.
	**/

	#ifndef QUICKSTEP_QUERY_EXECUTION_WORKORDERS_CONTAINER_HPP_
	#define QUICKSTEP_QUERY_EXECUTION_WORKORDERS_CONTAINER_HPP_

	#include <cstddef>
	#include <list>
	#include <memory>
	#include <queue>
	#include <vector>

	#include "query_execution/WorkOrderSelectionPolicy.hpp"
	#include "relational_operators/WorkOrder.hpp"
	#include "utility/Macros.hpp"
	#include "utility/PtrVector.hpp"

	#include "glog/logging.h"

	namespace quickstep {

	/** \addtogroup QueryExecution
	* @{
	*/

	/**
	* @brief A container to hold the WorkOrders for a given query.
	* @note This container stays alive during the lifetime of the query.
	* @note The NUMA node indexing is assumed to start from 0.
	**/
	class WorkOrdersContainer {
	public:
	/**
	* @brief Constructor
	*
	* @param num_operators Number of operators in the query DAG.
	* @param num_numa_nodes Number of NUMA nodes in the system.
	**/
	WorkOrdersContainer(const std::size_t num_operators,
	const std::size_t num_numa_nodes)
	: num_operators_(num_operators), num_numa_nodes_(num_numa_nodes) {
	DEBUG_ASSERT(num_operators != 0);
	for (std::size_t op = 0; op < num_operators; ++op) {
	normal_workorders_.push_back(
	new OperatorWorkOrdersContainer(num_numa_nodes_));
	rebuild_workorders_.push_back(
	new OperatorWorkOrdersContainer(num_numa_nodes_));
	}
	}

	/**
	* @brief Destructor.
	*
	* @note If the query is executed normally, we should never encounter a
	* situation where at the time of deletion the WorkOrdersContainer has
	* pending WorkOrders.
	**/
	~WorkOrdersContainer();

	/**
	* @brief Check if there are some pending WorkOrders for the given operator.
	*
	* @param operator_index Index of the operator.
	*
	* @return If there are pending WorkOrders.
	**/
	inline bool hasNormalWorkOrder(const std::size_t operator_index) const {
	DCHECK_LT(operator_index, num_operators_);
	return normal_workorders_[operator_index].hasWorkOrder();
	}

	/**
	* @brief Check if there are some pending normal (i.e. non-rebuild) WorkOrders
	* for the given operator which prefer the specified NUMA node.
	*
	* @param operator_index Index of the operator.
	* @param numa_node_id The NUMA node.
	*
	* @return If there are pending WorkOrders for the given operator which prefer
	* numa_node_id.
	**/
	inline bool hasNormalWorkOrderForNUMANode(
	const std::size_t operator_index, const int numa_node_id) const {
	DCHECK_LT(operator_index, num_operators_);
	DCHECK_GE(numa_node_id, 0);
	DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
	return normal_workorders_[operator_index].hasWorkOrderForNUMANode(
	numa_node_id);
	}

	/**
	* @brief Check if there are some pending rebuild WorkOrders for the given
	* operator.
	*
	* @param operator_index Index of the operator.
	*
	* @return If there are pending rebuild WorkOrders.
	**/
	inline bool hasRebuildWorkOrder(const std::size_t operator_index) const {
	DCHECK_LT(operator_index, num_operators_);
	return rebuild_workorders_[operator_index].hasWorkOrder();
	}

	/**
	* @brief Check if there are some pending rebuild WorkOrders for the given
	* operator which prefer the specified NUMA node.
	*
	* @param operator_index Index of the operator.
	* @param numa_node_id The NUMA node.
	*
	* @return If there are pending rebuild WorkOrders for the given operator which
	* prefer numa_node_id.
	**/
	inline bool hasRebuildWorkOrderForNUMANode(
	const std::size_t operator_index, const int numa_node_id) const {
	DCHECK_LT(operator_index, num_operators_);
	DCHECK_GE(numa_node_id, 0);
	DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
	return rebuild_workorders_[operator_index].hasWorkOrderForNUMANode(
	numa_node_id);
	}

	/**
	* @brief Get a normal (non-rebuild) WorkOrder for a given operator which
	* prefer the given NUMA node.
	*
	* @param operator_index The index of the operator.
	* @param numa_node_id The NUMA node.
	*
	* @return A WorkOrder which prefers numa_node_id. If no such WorkOrder is
	* available, return nullptr. The caller is responsible for taking the
	* ownership.
	**/
	WorkOrder* getNormalWorkOrderForNUMANode(const std::size_t operator_index,
	const int numa_node_id) {
	DCHECK_LT(operator_index, num_operators_);
	DCHECK_GE(numa_node_id, 0);
	DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
	return normal_workorders_[operator_index].getWorkOrderForNUMANode(
	numa_node_id);
	}

	/**
	* @brief Get a normal (non-rebuild) WorkOrder for a given operator.
	*
	* @param operator_index The index of the operator.
	* @param prefer_single_NUMA_node If true, first try to get workorders which
	* prefer exactly one NUMA node over workorders which list more than
	* one NUMA node as their preference.
	*
	* @return A WorkOrder. If no WorkOrder is available, returns nullptr. The
	* caller is responsible for taking the ownership.
	**/
	WorkOrder* getNormalWorkOrder(const std::size_t operator_index,
	const bool prefer_single_NUMA_node = true) {
	DCHECK_LT(operator_index, num_operators_);
	return normal_workorders_[operator_index].getWorkOrder(
	prefer_single_NUMA_node);
	}

	WorkOrder* getNextWorkOrder(std::size_t operator_index, bool is_rebuild) {
	if (rebuild_work_orders_policy_.hasWorkOrder()) {
	*operator_index = rebuild_work_orders_policy_.getOperatorIndexForNextWorkOrder();
	*is_rebuild = true;
	return rebuild_workorders_[*operator_index].getWorkOrder();
	}

	if (normal_work_orders_policy_.hasWorkOrder()) {
	*operator_index = normal_work_orders_policy_.getOperatorIndexForNextWorkOrder();
	*is_rebuild = false;
	return normal_workorders_[*operator_index].getWorkOrder();
	}

	return nullptr;
	}

	/**
	* @brief Get a rebuild WorkOrder for a given operator whch prefer the
	* specified NUMA node.
	*
	* @param operator_index The index of the operator.
	* @param numa_node_id The NUMA node.
	*
	* @return A WorkOrder that prefers numa_node_id. If no such WorkOrder is
	* available, return nullptr. The caller is responsible for taking the
	* ownership.
	**/
	WorkOrder* getRebuildWorkOrderForNUMANode(const std::size_t operator_index,
	const int numa_node_id) {
	DCHECK_LT(operator_index, num_operators_);
	DCHECK_GE(numa_node_id, 0);
	DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
	return rebuild_workorders_[operator_index].getWorkOrderForNUMANode(
	numa_node_id);
	}

	/**
	* @brief Get a rebuild WorkOrder for the given operator
	*
	* @param operator_index The index of the operator.
	* @param prefer_single_NUMA_node If true, first try to get workorders which
	* prefer exactly one NUMA node over workorders which list more than
	* one NUMA node as their preference.
	*
	* @return A WorkOrder. If no WorkOrder is available, returns nullptr. The
	* caller is responsible for taking the ownership.
	**/
	WorkOrder* getRebuildWorkOrder(const std::size_t operator_index,
	const bool prefer_single_NUMA_node = true) {
	DCHECK_LT(operator_index, num_operators_);
	return rebuild_workorders_[operator_index].getWorkOrder(
	prefer_single_NUMA_node);
	}

	/**
	* @brief Add a normal (non-rebuild) WorkOrder generated from a given
	* operator.
	*
	* @note Take the ownership of \p workorder.
	* @note The workorder to be added contains information about its preferred
	* NUMA nodes. This information is used to insert the WorkOrder
	* appropriately.
	*
	* @param workorder A pointer to the WorkOrder to be added.
	* @param operator_index The index of the operator in the query DAG.
	**/
	void addNormalWorkOrder(WorkOrder *workorder, const std::size_t operator_index) {
	DCHECK(workorder != nullptr);
	DCHECK_LT(operator_index, num_operators_);
	normal_workorders_[operator_index].addWorkOrder(workorder);
	normal_work_orders_policy_.addWorkOrder(operator_index);
	}

	/**
	* @brief Add a rebuild WorkOrder generated from a given operator.
	*
	* @note Take the ownership of \p workorder.
	* @note The workorder to be added contains information about its preferred
	* NUMA nodes. This information is used to insert the WorkOrder
	* appropriately.
	*
	* @param workorder A pointer to the WorkOrder to be added.
	* @param operator_index The index of the operator in the query DAG.
	**/
	void addRebuildWorkOrder(WorkOrder *workorder,
	const std::size_t operator_index) {
	DCHECK(workorder != nullptr);
	DCHECK_LT(operator_index, num_operators_);
	rebuild_workorders_[operator_index].addWorkOrder(workorder);
	rebuild_work_orders_policy_.addWorkOrder(operator_index);
	}

	/**
	* @brief Get the number of pending normal WorkOrders for a given operator
	* which prefer the specified NUMA node.
	*
	* @param operator_index The index of the operator.
	* @param numa_node_id The NUMA node.
	*
	* @return The number of pending WorkOrders which prefer numa_node_id.
	**/
	inline std::size_t getNumNormalWorkOrdersForNUMANode(
	const std::size_t operator_index, const int numa_node_id) const {
	DCHECK_LT(operator_index, num_operators_);
	DCHECK_GE(numa_node_id, 0);
	DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
	return normal_workorders_[operator_index].getNumWorkOrdersForNUMANode(
	numa_node_id);
	}

	/**
	* @brief Get the number of all pending normal (i.e. non-rebuild) WorkOrders
	* for a given operator.
	*
	* @param operator_index The index of the operator.
	*
	* @return The number of pending normal WorkOrders.
	**/
	inline std::size_t getNumNormalWorkOrders(
	const std::size_t operator_index) const {
	DCHECK_LT(operator_index, num_operators_);
	return normal_workorders_[operator_index].getNumWorkOrders();
	}

	/**
	* @brief Get the number of pending rebuild WorkOrders for a given operator
	* which prefer the specified NUMA node.
	*
	* @param operator_index The index of the operator.
	* @param numa_node_id The NUMA node.
	*
	* @return The number of pending WorkOrders which prefer numa_node_id.
	**/
	inline std::size_t getNumRebuildWorkOrdersForNUMANode(
	const std::size_t operator_index, const int numa_node_id) const {
	DCHECK_LT(operator_index, num_operators_);
	DCHECK_GE(numa_node_id, 0);
	DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
	return rebuild_workorders_[operator_index].getNumWorkOrdersForNUMANode(
	numa_node_id);
	}

	/**
	* @brief Get the number of all pending rebuild WorkOrders for a given
	* operator.
	*
	* @param operator_index The index of the operator.
	*
	* @return The number of pending rebuild WorkOrders.
	**/
	inline std::size_t getNumRebuildWorkOrders(
	const std::size_t operator_index) const {
	DCHECK_LT(operator_index, num_operators_);
	return rebuild_workorders_[operator_index].getNumWorkOrders();
	}

	/**
	* @brief Get the total number of work orders for the given operator.
	*
	* @param operator_index The index of the operator.
	*
	* @return The total number of WorkOrders (normal + rebuild).
	**/
	inline std::size_t getNumTotalWorkOrders(
	const std::size_t operator_index) const {
	return getNumNormalWorkOrders(operator_index) +
	getNumRebuildWorkOrders(operator_index);
	}

	private:
	/**
	* @brief An internal queue-based container structure to hold the WorkOrders.
	**/
	class InternalQueueContainer {
	public:
	InternalQueueContainer() {
	}

	inline void addWorkOrder(WorkOrder *workorder) {
	workorders_.emplace(std::unique_ptr<WorkOrder>(workorder));
	}

	inline WorkOrder* getWorkOrder() {
	if (workorders_.empty()) {
	return nullptr;
	}

	WorkOrder *work_order = workorders_.front().release();
	workorders_.pop();
	return work_order;
	}

	inline bool hasWorkOrder() const {
	return !workorders_.empty();
	}

	inline std::size_t getNumWorkOrders() const {
	return workorders_.size();
	}

	private:
	std::queue<std::unique_ptr<WorkOrder>> workorders_;

	DISALLOW_COPY_AND_ASSIGN(InternalQueueContainer);
	};

	/**
	* @brief An internal list-based container structure to hold the WorkOrders.
	*
	* @note We use this class for WorkOrders whose inputs come from multiple NUMA
	* nodes e.g. a HashJoinWorkOrder whose probe block and the hash table
	* may reside on two different NUMA nodes.
	**/
	class InternalListContainer {
	public:
	InternalListContainer() {
	}

	inline void addWorkOrder(WorkOrder *workorder) {
	workorders_.emplace_back(std::unique_ptr<WorkOrder>(workorder));
	}

	inline WorkOrder* getWorkOrder() {
	if (workorders_.empty()) {
	return nullptr;
	}

	WorkOrder *work_order = workorders_.front().release();
	workorders_.pop_front();
	return work_order;
	}

	/**
	* @note This method has O(N) complexity.
	**/
	WorkOrder* getWorkOrderForNUMANode(const int numa_node);

	inline bool hasWorkOrder() const {
	return !workorders_.empty();
	}

	/**
	* @brief Check if numa_node is in the set of preferred NUMA nodes for at
	* least one WorkOrder in this container.
	*
	* @note This method has O(N) complexity.
	**/
	bool hasWorkOrderForNUMANode(const int numa_node) const;

	inline std::size_t getNumWorkOrders() const {
	return workorders_.size();
	}

	/**
	* @brief Return the number of WorkOrders that list the numa_node as one of
	* the preferred nodes of execution.
	**/
	std::size_t getNumWorkOrdersForNUMANode(const int numa_node) const;

	private:
	std::list<std::unique_ptr<WorkOrder>> workorders_;

	DISALLOW_COPY_AND_ASSIGN(InternalListContainer);
	};

	/**
	* @brief A container to hold all the WorkOrders generated by one operator.
	**/
	class OperatorWorkOrdersContainer {
	public:
	explicit OperatorWorkOrdersContainer(const std::size_t num_numa_nodes)
	: num_numa_nodes_(num_numa_nodes) {
	for (std::size_t numa_node = 0; numa_node < num_numa_nodes; ++numa_node) {
	single_numa_node_workorders_.push_back(new InternalQueueContainer());
	}
	}

	void addWorkOrder(WorkOrder *workorder);

	bool hasWorkOrderForNUMANode(const int numa_node_id) const {
	DCHECK_GE(numa_node_id, 0);
	DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
	return single_numa_node_workorders_[numa_node_id].hasWorkOrder() \|\|
	multiple_numa_nodes_workorders_.hasWorkOrderForNUMANode(
	numa_node_id);
	}

	bool hasWorkOrder() const {
	if (!(numa_agnostic_workorders_.hasWorkOrder() \|\|
	multiple_numa_nodes_workorders_.hasWorkOrder())) {
	for (std::size_t i = 0; i < num_numa_nodes_; ++i) {
	if (hasWorkOrderForNUMANode(i)) {
	return true;
	}
	}
	return false;
	}
	return true;
	}

	std::size_t getNumWorkOrdersForNUMANode(
	const int numa_node_id) const {
	DCHECK_GE(numa_node_id, 0);
	DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
	return single_numa_node_workorders_[numa_node_id].getNumWorkOrders() +
	multiple_numa_nodes_workorders_.getNumWorkOrdersForNUMANode(
	numa_node_id);
	}

	inline std::size_t getNumWorkOrders() const {
	std::size_t num_workorders = numa_agnostic_workorders_.getNumWorkOrders();
	// for each NUMA node involved ..
	for (PtrVector<InternalQueueContainer>::const_iterator it =
	single_numa_node_workorders_.begin();
	it != single_numa_node_workorders_.end();
	++it) {
	// Add the number of workorders for the NUMA node.
	num_workorders += it->getNumWorkOrders();
	}
	// Add the number of workorders who have multiple NUMA nodes as input.
	num_workorders += multiple_numa_nodes_workorders_.getNumWorkOrders();
	return num_workorders;
	}

	WorkOrder* getWorkOrderForNUMANode(const int numa_node_id) {
	DCHECK_GE(numa_node_id, 0);
	DCHECK_LT(static_cast<std::size_t>(numa_node_id), num_numa_nodes_);
	WorkOrder *work_order = single_numa_node_workorders_[numa_node_id].getWorkOrder();
	if (work_order == nullptr) {
	work_order = multiple_numa_nodes_workorders_.getWorkOrderForNUMANode(
	numa_node_id);
	}
	return work_order;
	}

	WorkOrder* getWorkOrder(const bool prefer_single_NUMA_node = true);

	private:
	WorkOrder* getSingleNUMANodeWorkOrderHelper();

	const std::size_t num_numa_nodes_;

	// A container to store NUMA agnostic workorders.
	InternalQueueContainer numa_agnostic_workorders_;

	// A vector of containers to store workorders which prefer exactly one NUMA
	// node for execution.
	PtrVector<InternalQueueContainer> single_numa_node_workorders_;

	// A container to store workorders which prefer more than one NUMA node for
	// execution.
	InternalListContainer multiple_numa_nodes_workorders_;
	// Note that no workorder should be shared among the *_workorders_ structures.

	DISALLOW_COPY_AND_ASSIGN(OperatorWorkOrdersContainer);
	};

	const std::size_t num_operators_;
	const std::size_t num_numa_nodes_;

	PtrVector<OperatorWorkOrdersContainer> normal_workorders_;
	PtrVector<OperatorWorkOrdersContainer> rebuild_workorders_;

	LifoWorkOrderSelectionPolicy normal_work_orders_policy_;
	FifoWorkOrderSelectionPolicy rebuild_work_orders_policy_;

	DISALLOW_COPY_AND_ASSIGN(WorkOrdersContainer);
	};
	/** @} */

	} // namespace quickstep


	#endif // QUICKSTEP_QUERY_EXECUTION_WORKORDERS_CONTAINER_HPP_