Updates.
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index b5e07df..78b72af 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -260,6 +260,7 @@
quickstep_relationaloperators_RelationalOperator
quickstep_storage_InsertDestination
quickstep_storage_StorageBlock
+ quickstep_storage_StorageManager
quickstep_utility_DAG
quickstep_utility_Macros
tmb)
diff --git a/query_execution/QueryContext.cpp b/query_execution/QueryContext.cpp
index 068b208..6c5776e 100644
--- a/query_execution/QueryContext.cpp
+++ b/query_execution/QueryContext.cpp
@@ -65,6 +65,10 @@
<< "Attempted to create QueryContext from an invalid proto description:\n"
<< proto.DebugString();
+ for (int i = 0; i < proto.build_hash_operator_ids_size(); ++i) {
+ build_hash_operator_ids_.emplace_back(proto.build_hash_operator_ids(i));
+ }
+
for (int i = 0; i < proto.aggregation_states_size(); ++i) {
aggregation_states_.emplace_back(
AggregationOperationState::ReconstructFromProto(proto.aggregation_states(i),
diff --git a/query_execution/QueryContext.hpp b/query_execution/QueryContext.hpp
index 07297e3..428d526 100644
--- a/query_execution/QueryContext.hpp
+++ b/query_execution/QueryContext.hpp
@@ -135,7 +135,7 @@
/**
* @brief A unique identifier of a relational operator in the DAG.
**/
- typedef std::uint32_t dag_operator_id;
+ typedef std::uint64_t dag_operator_id;
/**
* @brief Constructor.
diff --git a/query_execution/QueryContext.proto b/query_execution/QueryContext.proto
index f8831db..a1dce0a 100644
--- a/query_execution/QueryContext.proto
+++ b/query_execution/QueryContext.proto
@@ -63,5 +63,5 @@
required uint64 query_id = 13;
- repeated uint32 build_hash_operator_ids = 14;
+ repeated uint64 build_hash_operator_ids = 14;
}
diff --git a/query_execution/QueryManagerBase.hpp b/query_execution/QueryManagerBase.hpp
index a274742..77150ad 100644
--- a/query_execution/QueryManagerBase.hpp
+++ b/query_execution/QueryManagerBase.hpp
@@ -173,7 +173,7 @@
*
* @param index The index of the given relational operator in the DAG.
**/
- void markOperatorFinished(const dag_node_index index);
+ virtual void markOperatorFinished(const dag_node_index index);
/**
* @brief Check if all the dependencies of the node at specified index have
diff --git a/query_execution/QueryManagerSingleNode.cpp b/query_execution/QueryManagerSingleNode.cpp
index 71097ab..deff206 100644
--- a/query_execution/QueryManagerSingleNode.cpp
+++ b/query_execution/QueryManagerSingleNode.cpp
@@ -19,6 +19,7 @@
#include "query_execution/QueryManagerSingleNode.hpp"
+#include <algorithm>
#include <cstddef>
#include <memory>
#include <utility>
@@ -31,6 +32,7 @@
#include "relational_operators/RelationalOperator.hpp"
#include "storage/InsertDestination.hpp"
#include "storage/StorageBlock.hpp"
+#include "storage/StorageManager.hpp"
#include "utility/DAG.hpp"
#include "glog/logging.h"
@@ -58,17 +60,9 @@
foreman_client_id_,
bus_)),
workorders_container_(
- new WorkOrdersContainer(num_operators_in_dag_, num_numa_nodes)) {
- // Mark the active operators in the DAG.
- const std::size_t kNumMaxInitialActiveOperators = 2;
- for (dag_node_index index = 0; index < num_operators_in_dag_; ++index) {
- if (checkAllBlockingDependenciesMet(index) && active_operators_.size() < kNumMaxInitialActiveOperators) {
- active_operators_.push_back(index);
- activateOperator(index);
- } else {
- inactive_operators_.push_back(index);
- }
- }
+ new WorkOrdersContainer(num_operators_in_dag_, num_numa_nodes)),
+ estimated_memory_consumption_in_bytes_(0) {
+ updateActiveOperators();
}
WorkerMessage* QueryManagerSingleNode::getNextWorkerMessage(
@@ -77,50 +71,12 @@
if (worker_message != nullptr) {
return worker_message;
}
- // Check the operators that have finished execution.
- std::vector<dag_node_index> execution_finished_operator_indexes;
- for (dag_node_index active_operator_index : active_operators_) {
- if (query_exec_state_->hasExecutionFinished(active_operator_index)) {
- execution_finished_operator_indexes.push_back(active_operator_index);
- }
- }
- // Remove the "done" operators from the list of active operators.
- for (dag_node_index index : execution_finished_operator_indexes) {
- auto iter =
- std::find(active_operators_.begin(), active_operators_.end(), index);
- DCHECK(iter != active_operators_.end());
- active_operators_.erase(iter);
- }
- // Collect the list of "ready to be active" operators.
- std::vector<dag_node_index> ready_to_be_active_operators;
- for (dag_node_index index : inactive_operators_) {
- if (checkAllBlockingDependenciesMet(index)) {
- ready_to_be_active_operators.push_back(index);
- }
- }
- // Move as many inactive operators to active operators, as the size of
- // execution_finished_operator_indexes.
- std::size_t num_operators_activated = 0;
- if (!inactive_operators_.empty() &&
- !execution_finished_operator_indexes.empty() &&
- !ready_to_be_active_operators.empty()) {
- for (; num_operators_activated <
- std::min(execution_finished_operator_indexes.size(),
- inactive_operators_.size());
- ++num_operators_activated) {
- active_operators_.push_back(ready_to_be_active_operators[num_operators_activated]);
- activateOperator(active_operators_.back());
- }
- }
- // Remove the newly activated operators.
- for (std::size_t i = 0; i < num_operators_activated; ++i) {
- auto iter = std::find(inactive_operators_.begin(),
- inactive_operators_.end(),
- ready_to_be_active_operators[i]);
- DCHECK(iter != inactive_operators_.end());
- inactive_operators_.erase(iter);
- }
+ // We didn't find a work order. Try activating new operators.
+ updateActiveOperators();
worker_message = getNextWorkerMessageFromActiveOperators(numa_node);
+ if (worker_message == nullptr) {
+ std::cout << "No workorder even after activating new operators\n";
+ }
return worker_message;
}
@@ -128,36 +84,38 @@
const numa_node_id numa_node) {
WorkOrder *work_order = nullptr;
for (dag_node_index index : active_operators_) {
- if (numa_node != kAnyNUMANodeID) {
- // First try to get a normal WorkOrder from the specified NUMA node.
- work_order = workorders_container_->getNormalWorkOrderForNUMANode(index, numa_node);
+ if (checkAllBlockingDependenciesMet(index)) {
+ if (numa_node != kAnyNUMANodeID) {
+ // First try to get a normal WorkOrder from the specified NUMA node.
+ work_order = workorders_container_->getNormalWorkOrderForNUMANode(index, numa_node);
+ if (work_order != nullptr) {
+ // A WorkOrder found on the given NUMA node.
+ query_exec_state_->incrementNumQueuedWorkOrders(index);
+ return WorkerMessage::WorkOrderMessage(work_order, index);
+ } else {
+ // Normal workorder not found on this node. Look for a rebuild workorder
+ // on this NUMA node.
+ work_order = workorders_container_->getRebuildWorkOrderForNUMANode(index, numa_node);
+ if (work_order != nullptr) {
+ return WorkerMessage::RebuildWorkOrderMessage(work_order, index);
+ }
+ }
+ }
+ // Either no workorder found on the given NUMA node, or numa_node is
+ // 'kAnyNUMANodeID'.
+ // Try to get a normal WorkOrder from other NUMA nodes.
+ work_order = workorders_container_->getNormalWorkOrder(index);
if (work_order != nullptr) {
- // A WorkOrder found on the given NUMA node.
query_exec_state_->incrementNumQueuedWorkOrders(index);
return WorkerMessage::WorkOrderMessage(work_order, index);
} else {
- // Normal workorder not found on this node. Look for a rebuild workorder
- // on this NUMA node.
- work_order = workorders_container_->getRebuildWorkOrderForNUMANode(index, numa_node);
+ // Normal WorkOrder not found, look for a RebuildWorkOrder.
+ work_order = workorders_container_->getRebuildWorkOrder(index);
if (work_order != nullptr) {
return WorkerMessage::RebuildWorkOrderMessage(work_order, index);
}
}
}
- // Either no workorder found on the given NUMA node, or numa_node is
- // 'kAnyNUMANodeID'.
- // Try to get a normal WorkOrder from other NUMA nodes.
- work_order = workorders_container_->getNormalWorkOrder(index);
- if (work_order != nullptr) {
- query_exec_state_->incrementNumQueuedWorkOrders(index);
- return WorkerMessage::WorkOrderMessage(work_order, index);
- } else {
- // Normal WorkOrder not found, look for a RebuildWorkOrder.
- work_order = workorders_container_->getRebuildWorkOrder(index);
- if (work_order != nullptr) {
- return WorkerMessage::RebuildWorkOrderMessage(work_order, index);
- }
- }
}
return nullptr;
}
@@ -171,6 +129,11 @@
if (!checkAllBlockingDependenciesMet(index)) {
return false;
}
+ if (std::find(active_operators_.begin(), active_operators_.end(), index) == active_operators_.end()) {
+ // This operator is not yet active.
+ std::cout << "Operator " << index << " not active yet\n";
+ return false;
+ }
const size_t num_pending_workorders_before =
workorders_container_->getNumNormalWorkOrders(index);
const bool done_generation =
@@ -242,7 +205,7 @@
}
void QueryManagerSingleNode::activateOperator(const dag_node_index index) {
- DCHECK(checkAllBlockingDependenciesMet(index));
+ // DCHECK(checkAllBlockingDependenciesMet(index));
// It is okay to call the line below multiple times.
query_dag_->getNodePayloadMutable(index)->informAllBlockingDependenciesMet();
query_context_->activateOperator(
@@ -250,4 +213,80 @@
processOperator(index, false);
}
+void QueryManagerSingleNode::updateActiveOperators() {
+ // Mark the active operators in the DAG.
+ for (dag_node_index index = 0; index < num_operators_in_dag_; ++index) {
+ if (!query_exec_state_->hasExecutionFinished(index)) {
+ if (!isOperatorActive(index)) {
+ // This is an inactive operator.
+ const std::size_t estimated_memory_for_operator = query_context_->getHashTableSize(index);
+ // NOTE(harshad) - We might get in trouble in the following situation:
+ // If there's a build operator in the leaf of the query plan DAG, that
+ // has huge memory requirements (greater than the threshold). That
+ // operator won't be activated thereby leading the query execution in a
+ // deadlock.
+ if (canActivateOperator(estimated_memory_for_operator)) {
+ estimated_memory_consumption_in_bytes_ += estimated_memory_for_operator;
+ active_operators_.push_back(index);
+ activateOperator(index);
+ if (estimated_memory_for_operator > 0) {
+ std::cout << "Activated operator: " << index << " Total active: " << active_operators_.size() << "\n";
+ }
+ } else {
+ std::cout << "operator: " << index << " not activated Total active: " << active_operators_.size() << "\n";
+ }
+ }
+ }
+ }
+}
+
+bool QueryManagerSingleNode::isOperatorActive(const dag_node_index index) const {
+ auto iter = std::find(active_operators_.begin(), active_operators_.end(), index);
+ return iter != active_operators_.end();
+}
+
+bool QueryManagerSingleNode::canActivateOperator(const std::size_t estimated_memory_for_operator) const {
+ const std::size_t total_memory_consumption = estimated_memory_for_operator + estimated_memory_consumption_in_bytes_;
+ const std::size_t estimated_num_slots = StorageManager::SlotsNeededForBytes(total_memory_consumption);
+ const std::size_t threshold_num_slots = kAdmissionNumSlotsThreshold * storage_manager_->getMaxBufferPoolSlots();
+ if (estimated_memory_for_operator > 0) {
+ std::cout << "Memory estimate: " << estimated_memory_for_operator;
+ std::cout << " estimated slots: " << estimated_num_slots << " threshold: " << threshold_num_slots << "\n";
+ }
+ return estimated_num_slots < threshold_num_slots;
+}
+
+void QueryManagerSingleNode::markOperatorFinished(const dag_node_index index) {
+ query_exec_state_->setExecutionFinished(index);
+
+ RelationalOperator *op = query_dag_->getNodePayloadMutable(index);
+ op->updateCatalogOnCompletion();
+
+ const relation_id output_rel = op->getOutputRelationID();
+ for (const std::pair<dag_node_index, bool> &dependent_link : query_dag_->getDependents(index)) {
+ const dag_node_index dependent_op_index = dependent_link.first;
+ RelationalOperator *dependent_op = query_dag_->getNodePayloadMutable(dependent_op_index);
+ // Signal dependent operator that current operator is done feeding input blocks.
+ if (output_rel >= 0) {
+ dependent_op->doneFeedingInputBlocks(output_rel);
+ }
+ if (checkAllBlockingDependenciesMet(dependent_op_index)) {
+ dependent_op->informAllBlockingDependenciesMet();
+ }
+ }
+ // Reduce the estimate for this operator.
+ const std::size_t estimated_memory_for_operator = query_context_->getHashTableSize(index);
+ estimated_memory_consumption_in_bytes_ -= estimated_memory_for_operator;
+ std::cout << "Operator " << index << " over, decremented: " << estimated_memory_for_operator << " bytes Total active: " << active_operators_.size();
+ std::cout << " Current slots: " << StorageManager::SlotsNeededForBytes(estimated_memory_consumption_in_bytes_) << "\n";
+ // Remove this operator from the active operators list.
+ auto iter =
+ std::find(active_operators_.begin(), active_operators_.end(), index);
+ if (iter != active_operators_.end()) {
+ active_operators_.erase(iter);
+ } else {
+ DLOG(WARNING) << "Marking operator " << index << " as finished which wasn't active";
+ }
+}
+
} // namespace quickstep
diff --git a/query_execution/QueryManagerSingleNode.hpp b/query_execution/QueryManagerSingleNode.hpp
index 8e4a929..9ee6a63 100644
--- a/query_execution/QueryManagerSingleNode.hpp
+++ b/query_execution/QueryManagerSingleNode.hpp
@@ -95,7 +95,11 @@
return query_context_.get();
}
+ void markOperatorFinished(const dag_node_index index) override;
+
private:
+ static constexpr float kAdmissionNumSlotsThreshold = 0.25;
+
bool checkNormalExecutionOver(const dag_node_index index) const override {
return (checkAllDependenciesMet(index) &&
!workorders_container_->hasNormalWorkOrder(index) &&
@@ -129,6 +133,12 @@
WorkerMessage *getNextWorkerMessageFromActiveOperators(
const numa_node_id numa_node);
+ bool isOperatorActive(const dag_node_index index) const;
+
+ bool canActivateOperator(const std::size_t estimated_memory_for_operator) const;
+
+ void updateActiveOperators();
+
const tmb::client_id foreman_client_id_;
StorageManager *storage_manager_;
@@ -142,6 +152,8 @@
std::vector<dag_node_index> inactive_operators_;
+ std::size_t estimated_memory_consumption_in_bytes_;
+
DISALLOW_COPY_AND_ASSIGN(QueryManagerSingleNode);
};
diff --git a/relational_operators/SortMergeRunOperator.cpp b/relational_operators/SortMergeRunOperator.cpp
index 7de17ab..0ba5115 100644
--- a/relational_operators/SortMergeRunOperator.cpp
+++ b/relational_operators/SortMergeRunOperator.cpp
@@ -50,50 +50,13 @@
StorageManager *storage_manager,
const tmb::client_id scheduler_client_id,
tmb::MessageBus *bus) {
- if (input_relation_is_stored_) {
- // Input blocks (or runs) are from base relation. Only possible when base
- // relation is stored sorted.
- if (!started_) {
- // Initialize merge tree completely, since all input runs are known.
- merge_tree_.initializeTree(input_relation_block_ids_.size());
- started_ = true;
- initializeInputRuns();
- }
- } else {
- // Input blocks (or runs) are pipelined from the sorted run generation
- // operator.
- if (!started_ && !input_stream_done_) {
- // Initialize merge tree for first pipeline mode.
- merge_tree_.initializeForPipeline();
- started_ = true;
- initializeInputRuns();
- }
- }
+ initializeOperatorState();
// Generate runs from merge tree.
return generateWorkOrders(container, query_context, storage_manager, scheduler_client_id, bus);
}
bool SortMergeRunOperator::getAllWorkOrderProtos(WorkOrderProtosContainer *container) {
- if (input_relation_is_stored_) {
- // Input blocks (or runs) are from base relation. Only possible when base
- // relation is stored sorted.
- if (!started_) {
- // Initialize merge tree completely, since all input runs are known.
- merge_tree_.initializeTree(input_relation_block_ids_.size());
- started_ = true;
- initializeInputRuns();
- }
- } else {
- // Input blocks (or runs) are pipelined from the sorted run generation
- // operator.
- if (!started_ && !input_stream_done_) {
- // Initialize merge tree for first pipeline mode.
- merge_tree_.initializeForPipeline();
- started_ = true;
- initializeInputRuns();
- }
- }
-
+ initializeOperatorState();
// Get merge jobs from merge tree.
std::vector<MergeTree::MergeJob> jobs;
const bool done_generating = merge_tree_.getMergeJobs(&jobs);
@@ -216,6 +179,8 @@
return;
}
+ initializeOperatorState();
+
// Now we know all the input blocks; compute the merge tree.
merge_tree_.initializeTree(input_relation_block_ids_.size());
@@ -227,6 +192,28 @@
merge_tree_.checkAndFixFinalMerge();
}
+void SortMergeRunOperator::initializeOperatorState() {
+ if (input_relation_is_stored_) {
+ // Input blocks (or runs) are from base relation. Only possible when base
+ // relation is stored sorted.
+ if (!started_) {
+ // Initialize merge tree completely, since all input runs are known.
+ merge_tree_.initializeTree(input_relation_block_ids_.size());
+ started_ = true;
+ initializeInputRuns();
+ }
+ } else {
+ // Input blocks (or runs) are pipelined from the sorted run generation
+ // operator.
+ if (!started_ && !input_stream_done_) {
+ // Initialize merge tree for first pipeline mode.
+ merge_tree_.initializeForPipeline();
+ started_ = true;
+ initializeInputRuns();
+ }
+ }
+}
+
namespace {
/**
diff --git a/relational_operators/SortMergeRunOperator.hpp b/relational_operators/SortMergeRunOperator.hpp
index 0839320..8674fb9 100644
--- a/relational_operators/SortMergeRunOperator.hpp
+++ b/relational_operators/SortMergeRunOperator.hpp
@@ -201,6 +201,15 @@
**/
serialization::WorkOrder* createWorkOrderProto(merge_run_operator::MergeTree::MergeJob *job);
+ /**
+ * @brief Initialize the operator's internal state.
+ *
+ * @note This function can be called a number of times, but the changes will
+ * only be in effect because of the first call. Subsequent calls will
+ * not do anything.
+ **/
+ void initializeOperatorState();
+
const CatalogRelation &input_relation_;
const CatalogRelation &output_relation_;
diff --git a/storage/HashTableKeyManager.hpp b/storage/HashTableKeyManager.hpp
index c902cbe..8852b00 100644
--- a/storage/HashTableKeyManager.hpp
+++ b/storage/HashTableKeyManager.hpp
@@ -78,7 +78,7 @@
const std::size_t key_start_in_bucket,
std::vector<std::size_t> *key_offsets) {
DCHECK(!key_types.empty());
- DCHECK_NOTNULL(key_offsets);
+ DCHECK(key_offsets != nullptr);
std::size_t fixed_key_size = 0;
for (const Type *subkey_type : key_types) {
