You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@quickstep.apache.org by hb...@apache.org on 2016/07/16 22:17:02 UTC
[1/8] incubator-quickstep git commit: addfeedback now supports
static/dynamic cases. [Forced Update!]
Repository: incubator-quickstep
Updated Branches:
refs/heads/memory-estimate 0b973b38e -> 21be79baf (forced update)
addfeedback now supports static/dynamic cases.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/a8405445
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/a8405445
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/a8405445
Branch: refs/heads/memory-estimate
Commit: a8405445b3928a3e4ee341ccdc86276f34655857
Parents: 03c35b9
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Thu Jul 14 00:26:09 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Thu Jul 14 00:26:09 2016 -0500
----------------------------------------------------------------------
query_execution/Learner.cpp | 22 ----------------------
query_execution/Learner.hpp | 24 ++++++++++++++++++++++--
query_execution/PriorityPolicyEnforcer.cpp | 6 +++++-
3 files changed, 27 insertions(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a8405445/query_execution/Learner.cpp
----------------------------------------------------------------------
diff --git a/query_execution/Learner.cpp b/query_execution/Learner.cpp
index 1908f5a..0bedb73 100644
--- a/query_execution/Learner.cpp
+++ b/query_execution/Learner.cpp
@@ -47,28 +47,6 @@ Learner::Learner()
LOG(INFO) << "Query ID|Operator ID|Worker ID|Time in microseconds|Workorder end timestamp";
}
-void Learner::addCompletionFeedback(
- const serialization::NormalWorkOrderCompletionMessage
- &workorder_completion_proto) {
- const std::size_t query_id = workorder_completion_proto.query_id();
- DCHECK(isQueryPresent(query_id));
- ExecutionStats *execution_stats = getExecutionStats(query_id);
- DCHECK(execution_stats != nullptr);
- execution_stats->addEntry(
- workorder_completion_proto.execution_time_in_microseconds(),
- workorder_completion_proto.operator_index());
-
- const std::size_t priority_level = getQueryPriorityUnsafe(query_id);
- if (!hasFeedbackFromAllQueriesInPriorityLevel(priority_level)) {
- updateFeedbackFromQueriesInPriorityLevel(priority_level);
- }
- updateProbabilitiesForQueriesInPriorityLevel(priority_level, query_id);
- updateProbabilitiesOfAllPriorityLevels();
- // printProbabilitiesForPriorityLevel(priority_level);
- printWorkOrderDetails(workorder_completion_proto);
- printPredictedWorkOrderTimes();
-}
-
void Learner::updateProbabilitiesForQueriesInPriorityLevel(
const std::size_t priority_level, const std::size_t query_id) {
DCHECK(isPriorityLevelPresent(priority_level));
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a8405445/query_execution/Learner.hpp
----------------------------------------------------------------------
diff --git a/query_execution/Learner.hpp b/query_execution/Learner.hpp
index d90bd63..26daf6e 100644
--- a/query_execution/Learner.hpp
+++ b/query_execution/Learner.hpp
@@ -33,7 +33,7 @@
#include "glog/logging.h"
-namespace serialization { class NormalWorkOrderCompletionMessage; }
+// namespace serialization { class NormalWorkOrderCompletionMessage; }
namespace quickstep {
@@ -47,9 +47,29 @@ class Learner {
**/
Learner();
+ template <bool dynamic_probabilities>
void addCompletionFeedback(
const serialization::NormalWorkOrderCompletionMessage
- &workorder_completion_proto);
+ &workorder_completion_proto) {
+ if (dynamic_probabilities) {
+ const std::size_t query_id = workorder_completion_proto.query_id();
+ DCHECK(isQueryPresent(query_id));
+ ExecutionStats *execution_stats = getExecutionStats(query_id);
+ DCHECK(execution_stats != nullptr);
+ execution_stats->addEntry(
+ workorder_completion_proto.execution_time_in_microseconds(),
+ workorder_completion_proto.operator_index());
+
+ const std::size_t priority_level = getQueryPriorityUnsafe(query_id);
+ if (!hasFeedbackFromAllQueriesInPriorityLevel(priority_level)) {
+ updateFeedbackFromQueriesInPriorityLevel(priority_level);
+ }
+ updateProbabilitiesForQueriesInPriorityLevel(priority_level, query_id);
+ updateProbabilitiesOfAllPriorityLevels();
+ }
+ printWorkOrderDetails(workorder_completion_proto);
+ // printPredictedWorkOrderTimes();
+ }
/**
* @brief Add a query to the Learner.
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/a8405445/query_execution/PriorityPolicyEnforcer.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PriorityPolicyEnforcer.cpp b/query_execution/PriorityPolicyEnforcer.cpp
index 36e2624..479a257 100644
--- a/query_execution/PriorityPolicyEnforcer.cpp
+++ b/query_execution/PriorityPolicyEnforcer.cpp
@@ -107,7 +107,11 @@ void PriorityPolicyEnforcer::processMessage(const TaggedMessage &tagged_message)
operator_id = proto.operator_index();
worker_directory_->decrementNumQueuedWorkOrders(
proto.worker_thread_index());
- learner_->addCompletionFeedback(proto);
+ if (FLAGS_dynamic_probabilities_in_learner) {
+ learner_->addCompletionFeedback<true>(proto);
+ } else {
+ learner_->addCompletionFeedback<false>(proto);
+ }
if (profile_individual_workorders_) {
recordTimeForWorkOrder(proto);
}
[2/8] incubator-quickstep git commit: Fixed bugs about float division
Posted by hb...@apache.org.
Fixed bugs about float division
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/eecdc5d8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/eecdc5d8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/eecdc5d8
Branch: refs/heads/memory-estimate
Commit: eecdc5d87ed73044e4ce45e4a76c985fbf17f091
Parents: a840544
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Thu Jul 14 23:33:14 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Thu Jul 14 23:33:14 2016 -0500
----------------------------------------------------------------------
query_execution/Learner.cpp | 10 +++++-----
query_execution/Learner.hpp | 8 ++++----
2 files changed, 9 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/eecdc5d8/query_execution/Learner.cpp
----------------------------------------------------------------------
diff --git a/query_execution/Learner.cpp b/query_execution/Learner.cpp
index 0bedb73..264e84e 100644
--- a/query_execution/Learner.cpp
+++ b/query_execution/Learner.cpp
@@ -76,7 +76,7 @@ void Learner::updateProbabilitiesForQueriesInPriorityLevel(
return;
}
// Else, there are more than one queries for the given priority level.
- std::unordered_map<std::size_t, std::size_t>
+ std::unordered_map<std::size_t, float>
mean_workorders_per_query =
getMeanWorkOrderTimesForQueriesInPriorityLevel(priority_level);
const float denominator = calculateDenominator(mean_workorders_per_query);
@@ -108,16 +108,16 @@ void Learner::updateProbabilitiesOfAllPriorityLevels() {
return;
}
// Compute the predicted work order execution times for all the level.
- std::unordered_map<std::size_t, std::size_t> predicted_time_for_level;
+ std::unordered_map<std::size_t, float> predicted_time_for_level;
std::size_t sum_active_priorities = 0;
for (auto priority_iter = execution_stats_.begin();
priority_iter != execution_stats_.end();
++priority_iter) {
- std::size_t total_time_curr_level = 0;
+ float total_time_curr_level = 0;
const std::size_t curr_priority_level = priority_iter->first;
sum_active_priorities += curr_priority_level;
// For each query, find its predicted work order execution time.
- const std::unordered_map<std::size_t, std::size_t>
+ const std::unordered_map<std::size_t, float>
mean_workorders_all_queries_curr_level =
getMeanWorkOrderTimesForQueriesInPriorityLevel(
curr_priority_level);
@@ -138,7 +138,7 @@ void Learner::updateProbabilitiesOfAllPriorityLevels() {
float total_num_workorders = 0;
for (auto predicted_time_iter : predicted_time_for_level) {
const std::size_t curr_priority_level = predicted_time_iter.first;
- const std::size_t num_workorders_for_curr_level =
+ const float num_workorders_for_curr_level =
(predicted_time_iter.second == 0)
? 0
: static_cast<float>(curr_priority_level) /
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/eecdc5d8/query_execution/Learner.hpp
----------------------------------------------------------------------
diff --git a/query_execution/Learner.hpp b/query_execution/Learner.hpp
index 26daf6e..f61e4f9 100644
--- a/query_execution/Learner.hpp
+++ b/query_execution/Learner.hpp
@@ -467,18 +467,18 @@ class Learner {
* @return An unordered_map in which: Key = query ID.
* Value = Mean time per work order for that query.
**/
- inline std::unordered_map<std::size_t, std::size_t>
+ inline std::unordered_map<std::size_t, float>
getMeanWorkOrderTimesForQueriesInPriorityLevel(
const std::size_t priority_level) {
DCHECK(isPriorityLevelPresent(priority_level));
- std::unordered_map<std::size_t, std::size_t> result;
+ std::unordered_map<std::size_t, float> result;
for (auto it = execution_stats_[priority_level].begin();
it != execution_stats_[priority_level].end();
++it) {
DCHECK(it->second.get() != nullptr);
auto query_stats = it->second->getCurrentStats();
result[it->first] =
- query_stats.second == 0 ? 0 : query_stats.first / query_stats.second;
+ query_stats.second == 0 ? 0 : query_stats.first / float(query_stats.second);
}
return result;
}
@@ -494,7 +494,7 @@ class Learner {
* @return The denominator to be used for probability calculations.
**/
inline float calculateDenominator(
- const std::unordered_map<std::size_t, std::size_t>
+ const std::unordered_map<std::size_t, float>
&mean_workorder_per_query) const {
float denominator = 0;
for (const auto &element : mean_workorder_per_query) {
[4/8] incubator-quickstep git commit: Learner keeps a STL set of
priority levels.
Posted by hb...@apache.org.
Learner keeps a STL set of priority levels.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/8caae5a0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/8caae5a0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/8caae5a0
Branch: refs/heads/memory-estimate
Commit: 8caae5a0f87b8e3dfc19c08af6a1595d5dd24048
Parents: 884a50e
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Fri Jul 15 16:30:46 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Fri Jul 15 16:30:46 2016 -0500
----------------------------------------------------------------------
query_execution/Learner.cpp | 21 ++-------------------
query_execution/Learner.hpp | 12 +++++++-----
2 files changed, 9 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8caae5a0/query_execution/Learner.cpp
----------------------------------------------------------------------
diff --git a/query_execution/Learner.cpp b/query_execution/Learner.cpp
index 264e84e..272d582 100644
--- a/query_execution/Learner.cpp
+++ b/query_execution/Learner.cpp
@@ -40,8 +40,7 @@ DEFINE_uint64(max_past_entries_learner,
"The maximum number of past WorkOrder execution statistics"
" entries for a query");
-Learner::Learner()
- : highest_priority_level_(kInvalidPriorityLevel) {
+Learner::Learner() {
probabilities_of_priority_levels_.reset(new ProbabilityStore());
// Format: Query ID, Operator ID, Worker ID, Time in micros, WO execution end timestamp.
LOG(INFO) << "Query ID|Operator ID|Worker ID|Time in microseconds|Workorder end timestamp";
@@ -232,23 +231,7 @@ void Learner::checkAndRemovePriorityLevel(const std::size_t priority_level) {
default_probabilities_.erase(priority_level);
// NOTE(harshad) : Not using this cache as it gets confusing.
// has_feedback_from_all_queries_.erase(priority_level);
- if (hasActiveQueries()) {
- if (static_cast<int>(priority_level) == highest_priority_level_) {
- // The priority level to be removed is the highest priority level.
- std::size_t new_highest_priority_level = 0;
- // Find the new highest priority level.
- for (auto priority_level_it = execution_stats_.cbegin();
- priority_level_it != execution_stats_.cend();
- ++priority_level_it) {
- if (priority_level_it->first > new_highest_priority_level) {
- new_highest_priority_level = priority_level_it->first;
- }
- }
- highest_priority_level_ = static_cast<int>(new_highest_priority_level);
- }
- } else {
- highest_priority_level_ = kInvalidPriorityLevel;
- }
+ priority_levels_set_.erase(priority_level);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8caae5a0/query_execution/Learner.hpp
----------------------------------------------------------------------
diff --git a/query_execution/Learner.hpp b/query_execution/Learner.hpp
index f61e4f9..36f1509 100644
--- a/query_execution/Learner.hpp
+++ b/query_execution/Learner.hpp
@@ -162,7 +162,11 @@ class Learner {
* kInvalidPriorityLevel.
**/
inline const int getHighestPriorityLevel() const {
- return highest_priority_level_;
+ if (!priority_levels_set_.empty()) {
+ return static_cast<int>(*priority_levels_set_.rbegin());
+ } else {
+ return kInvalidPriorityLevel;
+ }
}
/**
@@ -325,9 +329,7 @@ class Learner {
if (!isPriorityLevelPresent(priority_level)) {
current_probabilities_[priority_level].reset(new ProbabilityStore());
execution_stats_.emplace(priority_level, std::vector<std::pair<std::size_t, std::unique_ptr<ExecutionStats>>>());
- if (static_cast<int>(priority_level) > highest_priority_level_) {
- highest_priority_level_ = priority_level;
- }
+ priority_levels_set_.insert(priority_level);
}
}
@@ -555,7 +557,7 @@ class Learner {
// feedback from all the queries in the given priority level.
// std::unordered_map<std::size_t, bool> has_feedback_from_all_queries_;
- int highest_priority_level_;
+ std::set<std::size_t> priority_levels_set_;
DISALLOW_COPY_AND_ASSIGN(Learner);
};
[3/8] incubator-quickstep git commit: Added code for HPF.
Posted by hb...@apache.org.
Added code for HPF.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/884a50e1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/884a50e1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/884a50e1
Branch: refs/heads/memory-estimate
Commit: 884a50e18c76b9e3714a2acf142a5300c51bc09e
Parents: eecdc5d
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Fri Jul 15 15:54:19 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Fri Jul 15 15:54:30 2016 -0500
----------------------------------------------------------------------
query_execution/PriorityPolicyEnforcer.cpp | 86 +++++++++++++++++--------
query_execution/PriorityPolicyEnforcer.hpp | 2 +
2 files changed, 61 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/884a50e1/query_execution/PriorityPolicyEnforcer.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PriorityPolicyEnforcer.cpp b/query_execution/PriorityPolicyEnforcer.cpp
index 479a257..9bfe979 100644
--- a/query_execution/PriorityPolicyEnforcer.cpp
+++ b/query_execution/PriorityPolicyEnforcer.cpp
@@ -42,6 +42,8 @@ DEFINE_uint64(max_msgs_per_dispatch_round, 40, "Maximum number of messages that"
" can be allocated in a single round of dispatch of messages to"
" the workers.");
+DEFINE_bool(highest_priority_first, false, "Pick queries from the highest priority level first");
+
DEFINE_bool(dynamic_probabilities_in_learner, true, "Whether the learner should have dynamic probabilities or static probabilities");
PriorityPolicyEnforcer::PriorityPolicyEnforcer(const tmb::client_id foreman_client_id,
@@ -179,41 +181,71 @@ void PriorityPolicyEnforcer::processMessage(const TaggedMessage &tagged_message)
void PriorityPolicyEnforcer::getWorkerMessages(
std::vector<std::unique_ptr<WorkerMessage>> *worker_messages) {
+ if (!FLAGS_highest_priority_first) {
+ // Iterate over admitted queries until either there are no more
+ // messages available, or the maximum number of messages have
+ // been collected.
+ DCHECK(worker_messages->empty());
+ std::unordered_map<std::size_t, bool> finished_queries_ids;
+
+ if (learner_->hasActiveQueries()) {
+ // Key = priority level. Value = Whether we have already checked the
+ std::unordered_map<std::size_t, bool> checked_priority_levels;
+ // While there are more priority levels to be checked ..
+ while (checked_priority_levels.size() < priority_query_ids_.size() && worker_messages->size() < FLAGS_max_msgs_per_dispatch_round) {
+ const int chosen_priority_level = learner_->pickRandomPriorityLevel();
+ if (chosen_priority_level == kInvalidPriorityLevel) {
+ DLOG(INFO) << "No valid priority level chosen";
+ break;
+ } else if (checked_priority_levels.find(static_cast<std::size_t>(
+ chosen_priority_level)) != checked_priority_levels.end()) {
+ continue;
+ } else {
+ WorkerMessage *next_worker_message =
+ getNextWorkerMessageFromPriorityLevel(chosen_priority_level,
+ &finished_queries_ids);
+ if (next_worker_message != nullptr) {
+ worker_messages->push_back(std::unique_ptr<WorkerMessage>(next_worker_message));
+ } else {
+ checked_priority_levels[static_cast<std::size_t>(chosen_priority_level)] = true;
+ }
+ }
+ }
+ } else {
+ DLOG(INFO) << "No active queries in the learner at this point.";
+ return;
+ }
+ for (auto finished_qid_pair : finished_queries_ids) {
+ removeQuery(finished_qid_pair.first);
+ }
+ } else {
+ getWorkerMessagesHPF(worker_messages);
+ }
+}
+
+void PriorityPolicyEnforcer::getWorkerMessagesHPF(std::vector<std::unique_ptr<WorkerMessage>> *worker_messages) {
// Iterate over admitted queries until either there are no more
// messages available, or the maximum number of messages have
// been collected.
DCHECK(worker_messages->empty());
- std::unordered_map<std::size_t, bool> finished_queries_ids;
-
if (learner_->hasActiveQueries()) {
- // Key = priority level. Value = Whether we have already checked the
- std::unordered_map<std::size_t, bool> checked_priority_levels;
- // While there are more priority levels to be checked ..
- while (checked_priority_levels.size() < priority_query_ids_.size() && worker_messages->size() < FLAGS_max_msgs_per_dispatch_round) {
- const int chosen_priority_level = learner_->pickRandomPriorityLevel();
- if (chosen_priority_level == kInvalidPriorityLevel) {
- DLOG(INFO) << "No valid priority level chosen";
- break;
- } else if (checked_priority_levels.find(static_cast<std::size_t>(
- chosen_priority_level)) != checked_priority_levels.end()) {
- continue;
+ std::unordered_map<std::size_t, bool> finished_queries_ids;
+ const int chosen_priority_level = learner_->getHighestPriorityLevel();
+ if (chosen_priority_level == kInvalidPriorityLevel) {
+ DLOG(INFO) << "No valid priority level chosen";
+ return;
+ }
+ while (worker_messages->size() < FLAGS_max_msgs_per_dispatch_round) {
+ WorkerMessage *next_worker_message = getNextWorkerMessageFromPriorityLevel(chosen_priority_level, &finished_queries_ids);
+ if (next_worker_message != nullptr) {
+ worker_messages->push_back(std::unique_ptr<WorkerMessage>(next_worker_message));
} else {
- WorkerMessage *next_worker_message =
- getNextWorkerMessageFromPriorityLevel(chosen_priority_level,
- &finished_queries_ids);
- if (next_worker_message != nullptr) {
- worker_messages->push_back(std::unique_ptr<WorkerMessage>(next_worker_message));
- } else {
- checked_priority_levels[static_cast<std::size_t>(chosen_priority_level)] = true;
- }
+ break;
}
}
- } else {
- DLOG(INFO) << "No active queries in the learner at this point.";
- return;
- }
- for (auto finished_qid_pair : finished_queries_ids) {
- removeQuery(finished_qid_pair.first);
+ for (auto finished_qid_pair : finished_queries_ids) {
+ removeQuery(finished_qid_pair.first);
+ }
}
}
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/884a50e1/query_execution/PriorityPolicyEnforcer.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PriorityPolicyEnforcer.hpp b/query_execution/PriorityPolicyEnforcer.hpp
index b20d831..702b976 100644
--- a/query_execution/PriorityPolicyEnforcer.hpp
+++ b/query_execution/PriorityPolicyEnforcer.hpp
@@ -189,6 +189,8 @@ class PriorityPolicyEnforcer {
const std::size_t priority_level,
std::unordered_map<std::size_t, bool> *finished_queries_ids);
+ void getWorkerMessagesHPF(std::vector<std::unique_ptr<WorkerMessage>> *worker_messages);
+
const tmb::client_id foreman_client_id_;
const std::size_t num_numa_nodes_;
[7/8] incubator-quickstep git commit: Memory estimate from selection
output.
Posted by hb...@apache.org.
Memory estimate from selection output.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/8895a7de
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/8895a7de
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/8895a7de
Branch: refs/heads/memory-estimate
Commit: 8895a7def57a1d780a98c13b7195da890a728b43
Parents: 5f58d92
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Wed Jul 13 14:33:43 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Sat Jul 16 16:31:21 2016 -0500
----------------------------------------------------------------------
query_execution/PriorityPolicyEnforcer.cpp | 1 +
query_optimizer/CMakeLists.txt | 1 +
query_optimizer/ExecutionGenerator.cpp | 7 +++++++
query_optimizer/ExecutionGenerator.hpp | 2 ++
4 files changed, 11 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8895a7de/query_execution/PriorityPolicyEnforcer.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PriorityPolicyEnforcer.cpp b/query_execution/PriorityPolicyEnforcer.cpp
index e0290f1..30aa39e 100644
--- a/query_execution/PriorityPolicyEnforcer.cpp
+++ b/query_execution/PriorityPolicyEnforcer.cpp
@@ -78,6 +78,7 @@ bool PriorityPolicyEnforcer::admitQuery(QueryHandle *query_handle) {
learner_->addQuery(*query_handle);
query_handle->setAdmissionTime();
query_id_to_handle_[query_handle->query_id()] = query_handle;
+ LOG(INFO) << "Query " << query_handle->query_id() << " mem estimate: " << query_handle->getEstimatedMaxMemoryInBytes() << " bytes";
return true;
} else {
LOG(ERROR) << "Query with the same ID " << query_id << " exists";
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8895a7de/query_optimizer/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_optimizer/CMakeLists.txt b/query_optimizer/CMakeLists.txt
index 8912414..fdc607a 100644
--- a/query_optimizer/CMakeLists.txt
+++ b/query_optimizer/CMakeLists.txt
@@ -77,6 +77,7 @@ target_link_libraries(quickstep_queryoptimizer_ExecutionGenerator
quickstep_queryoptimizer_QueryPlan
quickstep_queryoptimizer_costmodel_CostModel
quickstep_queryoptimizer_costmodel_SimpleCostModel
+ quickstep_queryoptimizer_costmodel_StarSchemaSimpleCostModel
quickstep_queryoptimizer_expressions_AggregateFunction
quickstep_queryoptimizer_expressions_Alias
quickstep_queryoptimizer_expressions_AttributeReference
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8895a7de/query_optimizer/ExecutionGenerator.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/ExecutionGenerator.cpp b/query_optimizer/ExecutionGenerator.cpp
index 33708f0..9bb6174 100644
--- a/query_optimizer/ExecutionGenerator.cpp
+++ b/query_optimizer/ExecutionGenerator.cpp
@@ -55,6 +55,7 @@
#include "query_optimizer/QueryHandle.hpp"
#include "query_optimizer/QueryPlan.hpp"
#include "query_optimizer/cost_model/SimpleCostModel.hpp"
+#include "query_optimizer/cost_model/StarSchemaSimpleCostModel.hpp"
#include "query_optimizer/expressions/AggregateFunction.hpp"
#include "query_optimizer/expressions/Alias.hpp"
#include "query_optimizer/expressions/AttributeReference.hpp"
@@ -161,6 +162,8 @@ void ExecutionGenerator::generatePlan(const P::PhysicalPtr &physical_plan) {
cost_model_.reset(
new cost::SimpleCostModel(top_level_physical_plan_->shared_subplans()));
+ ss_cost_model_.reset(new cost::StarSchemaSimpleCostModel(
+ top_level_physical_plan_->shared_subplans()));
const CatalogRelation *result_relation = nullptr;
@@ -329,6 +332,10 @@ void ExecutionGenerator::createTemporaryCatalogRelation(
const relation_id output_rel_id = optimizer_context_->catalog_database()->addRelation(
catalog_relation.release());
+ const std::size_t estimated_cardinality = ss_cost_model_->estimateCardinality(physical);
+ const std::size_t estimated_tuple_size = (*catalog_relation_output)->getEstimatedByteLength();
+ query_handle_->addMemoryToEstimate(estimated_cardinality * estimated_tuple_size);
+
#ifdef QUICKSTEP_DISTRIBUTED
referenced_relation_ids_.insert(output_rel_id);
#endif
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/8895a7de/query_optimizer/ExecutionGenerator.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/ExecutionGenerator.hpp b/query_optimizer/ExecutionGenerator.hpp
index c7fd018..8cd8283 100644
--- a/query_optimizer/ExecutionGenerator.hpp
+++ b/query_optimizer/ExecutionGenerator.hpp
@@ -415,6 +415,8 @@ class ExecutionGenerator {
* @brief The cost model to use for creating the execution plan.
*/
std::unique_ptr<cost::CostModel> cost_model_;
+ // The star schema cost model.
+ std::unique_ptr<cost::CostModel> ss_cost_model_;
physical::TopLevelPlanPtr top_level_physical_plan_;
[6/8] incubator-quickstep git commit: Add memory estimate for a query
in handle.
Posted by hb...@apache.org.
Add memory estimate for a query in handle.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/5f58d927
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/5f58d927
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/5f58d927
Branch: refs/heads/memory-estimate
Commit: 5f58d927ef18e5700c28e5965bd1f9cdd6cca773
Parents: eea5603
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Mon Jul 11 16:32:28 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Sat Jul 16 16:31:21 2016 -0500
----------------------------------------------------------------------
query_optimizer/ExecutionGenerator.cpp | 7 +++++
query_optimizer/QueryHandle.hpp | 10 +++++++
storage/HashTableFactory.hpp | 31 ++++++++++++++++++++
storage/LinearOpenAddressingHashTable.hpp | 12 ++++++++
storage/SeparateChainingHashTable.hpp | 13 ++++++++
.../SimpleScalarSeparateChainingHashTable.hpp | 8 +++++
6 files changed, 81 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/5f58d927/query_optimizer/ExecutionGenerator.cpp
----------------------------------------------------------------------
diff --git a/query_optimizer/ExecutionGenerator.cpp b/query_optimizer/ExecutionGenerator.cpp
index 45f5f78..33708f0 100644
--- a/query_optimizer/ExecutionGenerator.cpp
+++ b/query_optimizer/ExecutionGenerator.cpp
@@ -741,6 +741,13 @@ void ExecutionGenerator::convertHashJoin(const P::HashJoinPtr &physical_plan) {
}
hash_table_proto->set_estimated_num_entries(build_cardinality);
+ // Add the memory estimate to the query handle.
+ // TODO(harshad) - Convert the build_cardinality to bytes.
+ const std::size_t hash_table_estimated_memory_in_bytes =
+ JoinHashTableFactory::GetEstimatedMemoryInBytes(
+ build_cardinality,
+ HashTableImplTypeProtoFromString(FLAGS_join_hashtable_type));
+ query_handle_->addMemoryToEstimate(hash_table_estimated_memory_in_bytes);
// Create three operators.
const QueryPlan::DAGNodeIndex build_operator_index =
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/5f58d927/query_optimizer/QueryHandle.hpp
----------------------------------------------------------------------
diff --git a/query_optimizer/QueryHandle.hpp b/query_optimizer/QueryHandle.hpp
index 04f672e..a07313f 100644
--- a/query_optimizer/QueryHandle.hpp
+++ b/query_optimizer/QueryHandle.hpp
@@ -50,6 +50,7 @@ class QueryHandle {
const std::uint64_t query_priority = 1)
: query_id_(query_id),
query_priority_(query_priority),
+ estimated_max_memory_in_bytes_(0),
query_plan_(new QueryPlan()),
query_result_relation_(nullptr) {}
@@ -153,9 +154,18 @@ class QueryHandle {
return std::chrono::duration<double, std::milli>(admission_time_ - entry_time_).count();
}
+ std::size_t getEstimatedMaxMemoryInBytes() const {
+ return estimated_max_memory_in_bytes_;
+ }
+
+ void addMemoryToEstimate(const std::size_t memory_bytes) {
+ estimated_max_memory_in_bytes_ += memory_bytes;
+ }
+
private:
const std::size_t query_id_;
const std::uint64_t query_priority_;
+ std::size_t estimated_max_memory_in_bytes_;
std::unique_ptr<QueryPlan> query_plan_;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/5f58d927/storage/HashTableFactory.hpp
----------------------------------------------------------------------
diff --git a/storage/HashTableFactory.hpp b/storage/HashTableFactory.hpp
index 34baaeb..47e3a24 100644
--- a/storage/HashTableFactory.hpp
+++ b/storage/HashTableFactory.hpp
@@ -135,6 +135,37 @@ template <typename ValueT,
bool allow_duplicate_keys>
class HashTableFactory {
public:
+ static std::size_t GetEstimatedMemoryInBytes(
+ const std::size_t num_estimated_entries,
+ const serialization::HashTableImplType hash_table_type) {
+ switch (hash_table_type) {
+ case serialization::HashTableImplType::LINEAR_OPEN_ADDRESSING:
+ return LinearOpenAddressingHashTable<ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>::
+ GetEstimatedMemoryInBytes(num_estimated_entries);
+ case serialization::HashTableImplType::SEPARATE_CHAINING:
+ return SeparateChainingHashTable<ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>::
+ GetEstimatedMemoryInBytes(num_estimated_entries);
+ case serialization::HashTableImplType::SIMPLE_SCALAR_SEPARATE_CHAINING:
+ return SimpleScalarSeparateChainingHashTable<ValueT,
+ resizable,
+ serializable,
+ force_key_copy,
+ allow_duplicate_keys>::
+ GetEstimatedMemoryInBytes(num_estimated_entries);
+ default: {
+ LOG(FATAL) << "Unrecognized HashTableImplType\n";
+ }
+ }
+ }
+
/**
* @brief Create a new resizable HashTable, with the type selected by
* hash_table_type. Other parameters are forwarded to the HashTable's
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/5f58d927/storage/LinearOpenAddressingHashTable.hpp
----------------------------------------------------------------------
diff --git a/storage/LinearOpenAddressingHashTable.hpp b/storage/LinearOpenAddressingHashTable.hpp
index 438a2d1..9f841cd 100644
--- a/storage/LinearOpenAddressingHashTable.hpp
+++ b/storage/LinearOpenAddressingHashTable.hpp
@@ -61,6 +61,18 @@ class LinearOpenAddressingHashTable : public HashTable<ValueT,
force_key_copy,
allow_duplicate_keys> {
public:
+ static std::size_t GetEstimatedMemoryInBytes(const std::size_t num_entries) {
+ const std::size_t num_slots_tmp = get_next_prime_number(num_entries * kHashTableLoadFactor);
+ const std::size_t bucket_size = LinearOpenAddressingHashTable::ComputeBucketSize(sizeof(std::size_t) * 2);
+ // TODO(harshad) - Refine the bucket_size below with the help of KeyManager.
+ // NOTE(harshad) - We ignore the estimated variable key size term below
+ // assuming that we don't have variable length keys in SSB queries.
+ const std::size_t required_memory = LinearOpenAddressingHashTable::ComputeBucketSize(sizeof(std::size_t) * 2)
+ + (num_slots_tmp / kHashTableLoadFactor)
+ * (bucket_size /*+ *key_manager_.getEstimatedVariableKeySize()*/);
+ return required_memory;
+ }
+
// Bring in constants from HashTable.
static constexpr unsigned char kEmptyHashByte
= HashTable<ValueT, resizable, serializable, force_key_copy, allow_duplicate_keys>::kEmptyHashByte;
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/5f58d927/storage/SeparateChainingHashTable.hpp
----------------------------------------------------------------------
diff --git a/storage/SeparateChainingHashTable.hpp b/storage/SeparateChainingHashTable.hpp
index c096b1b..e3dbff3 100644
--- a/storage/SeparateChainingHashTable.hpp
+++ b/storage/SeparateChainingHashTable.hpp
@@ -61,6 +61,19 @@ class SeparateChainingHashTable : public HashTable<ValueT,
force_key_copy,
allow_duplicate_keys> {
public:
+ static std::size_t GetEstimatedMemoryInBytes(const std::size_t num_entries) {
+ const std::size_t num_slots_tmp = get_next_prime_number(num_entries * kHashTableLoadFactor);
+ // TODO(harshad) - Refine the bucket_size below with the help of KeyManager.
+ const std::size_t bucket_size = SeparateChainingHashTable::ComputeBucketSize(sizeof(std::size_t) * 2);
+ // NOTE(harshad) - We ignore the estimated variable key size term below
+ // assuming that we don't have variable length keys in SSB queries.
+ const std::size_t required_memory =
+ sizeof(Header) + num_slots_tmp * sizeof(std::atomic<std::size_t>) +
+ (num_slots_tmp / kHashTableLoadFactor) *
+ (bucket_size /*+ *key_manager_.getEstimatedVariableKeySize()*/);
+ return required_memory;
+ }
+
SeparateChainingHashTable(const std::vector<const Type*> &key_types,
const std::size_t num_entries,
StorageManager *storage_manager);
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/5f58d927/storage/SimpleScalarSeparateChainingHashTable.hpp
----------------------------------------------------------------------
diff --git a/storage/SimpleScalarSeparateChainingHashTable.hpp b/storage/SimpleScalarSeparateChainingHashTable.hpp
index 962a66c..c6e5b4e 100644
--- a/storage/SimpleScalarSeparateChainingHashTable.hpp
+++ b/storage/SimpleScalarSeparateChainingHashTable.hpp
@@ -74,6 +74,14 @@ class SimpleScalarSeparateChainingHashTable : public HashTable<ValueT,
force_key_copy,
allow_duplicate_keys> {
public:
+ static std::size_t GetEstimatedMemoryInBytes(const std::size_t num_entries) {
+ const std::size_t num_slots_tmp = get_next_prime_number(num_entries * kHashTableLoadFactor);
+ const std::size_t required_memory = sizeof(Header)
+ + num_slots_tmp * sizeof(std::atomic<std::size_t>)
+ + (num_slots_tmp / kHashTableLoadFactor) * sizeof(Bucket);
+ return required_memory;
+ }
+
SimpleScalarSeparateChainingHashTable(const std::vector<const Type*> &key_types,
const std::size_t num_entries,
StorageManager *storage_manager);
[5/8] incubator-quickstep git commit: HPF can schedule work from
lower priority levels.
Posted by hb...@apache.org.
HPF can schedule work from lower priority levels.
- If highest priority level doesn't have work and we have some free
cycles, we will schedule work from lower priority levels.
- We will probe priority levels in decreasing order to see if there is
any work to be scheduled.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/eea56033
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/eea56033
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/eea56033
Branch: refs/heads/memory-estimate
Commit: eea560336e29d15a4611f7ae4910c878a2e6bd0d
Parents: 8caae5a
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Fri Jul 15 16:38:46 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Fri Jul 15 16:38:46 2016 -0500
----------------------------------------------------------------------
query_execution/Learner.hpp | 4 ++++
query_execution/PriorityPolicyEnforcer.cpp | 15 +++++++++------
2 files changed, 13 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/eea56033/query_execution/Learner.hpp
----------------------------------------------------------------------
diff --git a/query_execution/Learner.hpp b/query_execution/Learner.hpp
index 36f1509..8f63c19 100644
--- a/query_execution/Learner.hpp
+++ b/query_execution/Learner.hpp
@@ -169,6 +169,10 @@ class Learner {
}
}
+ inline const std::set<std::size_t>& getSetOfPriorityLevels() const {
+ return priority_levels_set_;
+ }
+
/**
* @brief Randomly pick a priority level.
*
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/eea56033/query_execution/PriorityPolicyEnforcer.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PriorityPolicyEnforcer.cpp b/query_execution/PriorityPolicyEnforcer.cpp
index 9bfe979..e0290f1 100644
--- a/query_execution/PriorityPolicyEnforcer.cpp
+++ b/query_execution/PriorityPolicyEnforcer.cpp
@@ -229,18 +229,21 @@ void PriorityPolicyEnforcer::getWorkerMessagesHPF(std::vector<std::unique_ptr<Wo
// been collected.
DCHECK(worker_messages->empty());
if (learner_->hasActiveQueries()) {
+ const std::set<std::size_t> &priority_set = learner_->getSetOfPriorityLevels();
std::unordered_map<std::size_t, bool> finished_queries_ids;
- const int chosen_priority_level = learner_->getHighestPriorityLevel();
- if (chosen_priority_level == kInvalidPriorityLevel) {
- DLOG(INFO) << "No valid priority level chosen";
+ auto priority_set_reverse_it = priority_set.rbegin();
+ if (priority_set.empty()) {
+ DLOG(INFO) << "No priority level available";
return;
}
- while (worker_messages->size() < FLAGS_max_msgs_per_dispatch_round) {
- WorkerMessage *next_worker_message = getNextWorkerMessageFromPriorityLevel(chosen_priority_level, &finished_queries_ids);
+ while (worker_messages->size() < FLAGS_max_msgs_per_dispatch_round && priority_set_reverse_it != priority_set.rend()) {
+ WorkerMessage *next_worker_message = getNextWorkerMessageFromPriorityLevel(*priority_set_reverse_it, &finished_queries_ids);
if (next_worker_message != nullptr) {
worker_messages->push_back(std::unique_ptr<WorkerMessage>(next_worker_message));
} else {
- break;
+ // No work orders available in the current priority level, check the
+ // next largest priority level.
+ ++priority_set_reverse_it;
}
}
for (auto finished_qid_pair : finished_queries_ids) {
[8/8] incubator-quickstep git commit: Check memory availability
before admitting a query.
Posted by hb...@apache.org.
Check memory availability before admitting a query.
Project: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/commit/21be79ba
Tree: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/tree/21be79ba
Diff: http://git-wip-us.apache.org/repos/asf/incubator-quickstep/diff/21be79ba
Branch: refs/heads/memory-estimate
Commit: 21be79bafcf51f0302ea2ab672898c9028392a69
Parents: 8895a7d
Author: Harshad Deshmukh <hb...@apache.org>
Authored: Wed Jul 13 23:19:58 2016 -0500
Committer: Harshad Deshmukh <hb...@apache.org>
Committed: Sat Jul 16 16:32:21 2016 -0500
----------------------------------------------------------------------
query_execution/CMakeLists.txt | 1 +
query_execution/PriorityPolicyEnforcer.cpp | 51 +++++++++++++++++++------
query_execution/PriorityPolicyEnforcer.hpp | 4 ++
3 files changed, 45 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/21be79ba/query_execution/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/query_execution/CMakeLists.txt b/query_execution/CMakeLists.txt
index 9ab86b2..11e0e1d 100644
--- a/query_execution/CMakeLists.txt
+++ b/query_execution/CMakeLists.txt
@@ -116,6 +116,7 @@ target_link_libraries(quickstep_queryexecution_PolicyEnforcer
quickstep_queryexecution_WorkerMessage
quickstep_queryoptimizer_QueryHandle
quickstep_relationaloperators_WorkOrder
+ quickstep_storage_StorageManager
quickstep_utility_Macros
tmb)
target_link_libraries(quickstep_queryexecution_PriorityPolicyEnforcer
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/21be79ba/query_execution/PriorityPolicyEnforcer.cpp
----------------------------------------------------------------------
diff --git a/query_execution/PriorityPolicyEnforcer.cpp b/query_execution/PriorityPolicyEnforcer.cpp
index 30aa39e..2496b74 100644
--- a/query_execution/PriorityPolicyEnforcer.cpp
+++ b/query_execution/PriorityPolicyEnforcer.cpp
@@ -17,6 +17,7 @@
#include "query_execution/PriorityPolicyEnforcer.hpp"
+#include <algorithm>
#include <cstddef>
#include <memory>
#include <queue>
@@ -32,6 +33,7 @@
#include "query_execution/WorkerDirectory.hpp"
#include "query_optimizer/QueryHandle.hpp"
#include "relational_operators/WorkOrder.hpp"
+#include "storage/StorageManager.hpp"
#include "gflags/gflags.h"
#include "glog/logging.h"
@@ -59,12 +61,13 @@ PriorityPolicyEnforcer::PriorityPolicyEnforcer(const tmb::client_id foreman_clie
storage_manager_(storage_manager),
worker_directory_(worker_directory),
bus_(bus),
- profile_individual_workorders_(profile_individual_workorders) {
+ profile_individual_workorders_(profile_individual_workorders),
+ committed_memory_(0) {
learner_.reset(new Learner());
}
bool PriorityPolicyEnforcer::admitQuery(QueryHandle *query_handle) {
- if (admitted_queries_.size() < kMaxConcurrentQueries) {
+ if (admissionMemoryCheck(query_handle)) { //admitted_queries_.size() < kMaxConcurrentQueries) {
// Ok to admit the query.
const std::size_t query_id = query_handle->query_id();
if (admitted_queries_.find(query_id) == admitted_queries_.end()) {
@@ -72,13 +75,14 @@ bool PriorityPolicyEnforcer::admitQuery(QueryHandle *query_handle) {
admitted_queries_[query_id].reset(
new QueryManager(foreman_client_id_, num_numa_nodes_, query_handle,
catalog_database_, storage_manager_, bus_));
- DLOG(INFO) << "Admitted query with ID: " << query_handle->query_id()
- << " priority: " << query_handle->query_priority();
+ std::cout << "Admitted query with ID: " << query_handle->query_id()
+ << " priority: " << query_handle->query_priority() << "\n";
priority_query_ids_[query_handle->query_priority()].emplace_back(query_id);
learner_->addQuery(*query_handle);
query_handle->setAdmissionTime();
query_id_to_handle_[query_handle->query_id()] = query_handle;
LOG(INFO) << "Query " << query_handle->query_id() << " mem estimate: " << query_handle->getEstimatedMaxMemoryInBytes() << " bytes";
+ LOG(INFO) << "Query " << query_handle->query_id() << " memory admissible? " << admissionMemoryCheck(query_handle);
return true;
} else {
LOG(ERROR) << "Query with the same ID " << query_id << " exists";
@@ -86,9 +90,12 @@ bool PriorityPolicyEnforcer::admitQuery(QueryHandle *query_handle) {
}
} else {
// This query will have to wait.
- DLOG(INFO) << "Query " << query_handle->query_id() << " waitlisted";
- query_id_to_handle_[query_handle->query_id()] = query_handle;
- waiting_queries_.push(query_handle);
+ std::cout << "Query " << query_handle->query_id() << " waitlisted\n";
+ if (query_id_to_handle_.find(query_handle->query_id()) == query_id_to_handle_.end()) {
+ // This query was not waitlisted earlier.
+ query_id_to_handle_[query_handle->query_id()] = query_handle;
+ waiting_queries_.push(query_handle);
+ }
return false;
}
}
@@ -172,8 +179,11 @@ void PriorityPolicyEnforcer::processMessage(const TaggedMessage &tagged_message)
if (!waiting_queries_.empty()) {
// Admit the earliest waiting query.
QueryHandle *new_query = waiting_queries_.front();
- waiting_queries_.pop();
- admitQuery(new_query);
+ // waiting_queries_.pop();
+ if (admitQuery(new_query)) {
+ std::cout << "Removing Q " << new_query->query_id() << " from waitlist\n";
+ waiting_queries_.pop();
+ }
}
} else if (return_code == QueryManager::QueryStatusCode::kOperatorExecuted) {
learner_->removeOperator(query_id, operator_id);
@@ -276,6 +286,8 @@ void PriorityPolicyEnforcer::removeQuery(const std::size_t query_id) {
priority_query_ids_.erase(query_priority_unsigned);
}
query_id_to_handle_[query_id]->setCompletionTime();
+ const std::size_t estimated_memory_bytes = query_id_to_handle_[query_id]->getEstimatedMaxMemoryInBytes();
+ committed_memory_ -= estimated_memory_bytes;
// Remove the query from the learner.
learner_->removeQuery(query_id);
// TODO(harshad) - Admit waiting queries, if any.
@@ -284,12 +296,13 @@ void PriorityPolicyEnforcer::removeQuery(const std::size_t query_id) {
bool PriorityPolicyEnforcer::admitQueries(
const std::vector<QueryHandle*> &query_handles) {
+ bool result = true;
for (QueryHandle *curr_query : query_handles) {
if (!admitQuery(curr_query)) {
- return false;
+ result = false;
}
}
- return true;
+ return result;
}
void PriorityPolicyEnforcer::recordTimeForWorkOrder(
@@ -363,4 +376,20 @@ WorkerMessage* PriorityPolicyEnforcer::getNextWorkerMessageFromPriorityLevel(
return nullptr;
}
+bool PriorityPolicyEnforcer::admissionMemoryCheck(const QueryHandle *query_handle) {
+ if (admitted_queries_.empty()) {
+ // No query running in the system, let the query in.
+ return true;
+ }
+ const std::size_t estimated_memory_requirement_bytes = query_handle->getEstimatedMaxMemoryInBytes();
+ const std::size_t estimated_slots = StorageManager::SlotsNeededForBytes(estimated_memory_requirement_bytes);
+ const std::size_t current_slots = StorageManager::SlotsNeededForBytes(storage_manager_->getMemorySize());
+ const std::size_t committed_slots = StorageManager::SlotsNeededForBytes(committed_memory_);
+ if (std::max(committed_slots, current_slots) + estimated_slots < storage_manager_->getMaxBufferPoolSlots()) {
+ committed_memory_ += estimated_memory_requirement_bytes;
+ return true;
+ }
+ return false;
+}
+
} // namespace quickstep
http://git-wip-us.apache.org/repos/asf/incubator-quickstep/blob/21be79ba/query_execution/PriorityPolicyEnforcer.hpp
----------------------------------------------------------------------
diff --git a/query_execution/PriorityPolicyEnforcer.hpp b/query_execution/PriorityPolicyEnforcer.hpp
index 702b976..9d727ed 100644
--- a/query_execution/PriorityPolicyEnforcer.hpp
+++ b/query_execution/PriorityPolicyEnforcer.hpp
@@ -191,6 +191,8 @@ class PriorityPolicyEnforcer {
void getWorkerMessagesHPF(std::vector<std::unique_ptr<WorkerMessage>> *worker_messages);
+ bool admissionMemoryCheck(const QueryHandle *query_handle);
+
const tmb::client_id foreman_client_id_;
const std::size_t num_numa_nodes_;
@@ -228,6 +230,8 @@ class PriorityPolicyEnforcer {
std::unique_ptr<Learner> learner_;
+ std::size_t committed_memory_;
+
DISALLOW_COPY_AND_ASSIGN(PriorityPolicyEnforcer);
};