Merge pull request #1158 from nappelson/plan_timeout

Support for optimizer task execution timeouts

Author: mengranwo

src/include/common/exception.h

@@ -57,8 +57,9 @@ enum class ExceptionType {
   CONNECTION = 21,        // connection related
   SYNTAX = 22,            // syntax related
   SETTINGS = 23,          // settings related
-  BINDER = 24,            // settings related
-  NETWORK = 25
+  BINDER = 24,            // binder related
+  NETWORK = 25,           // network related
+  OPTIMIZER = 26          // optimizer related
 };
 
 class Exception : public std::runtime_error {
@@ -125,6 +126,8 @@ class Exception : public std::runtime_error {
         return "Syntax";
       case ExceptionType::SETTINGS:
         return "Settings";
+      case ExceptionType::OPTIMIZER:
+        return "Optimizer";
       default:
         return "Unknown";
     }
@@ -452,4 +455,12 @@ class BinderException : public Exception {
   BinderException(std::string msg) : Exception(ExceptionType::BINDER, msg) {}
 };
 
+class OptimizerException : public Exception {
+  OptimizerException() = delete;
+
+ public:
+  OptimizerException(std::string msg)
+      : Exception(ExceptionType::OPTIMIZER, msg) {}
+};
+
 }  // namespace peloton

src/include/optimizer/group.h

@@ -73,9 +73,18 @@ class Group {
 
   std::shared_ptr<ColumnStats> GetStats(std::string column_name);
 
-  void AddStats(std::string column_name, std::shared_ptr<ColumnStats> stats);  
-
-  bool HasColumnStats(std::string column_name);  
+  void AddStats(std::string column_name, std::shared_ptr<ColumnStats> stats);
+
+  bool HasColumnStats(std::string column_name);
+
+  /*
+   * HasExpressions - Determines whether or not a lowest cost expression exists
+   * for this group
+   *
+   * properties: the property set used to search the lowest cost expressions
+   * return: true if an expression is found, false otherwise
+   */
+  bool HasExpressions(const std::shared_ptr<PropertySet> &properties) const;
 
   void SetNumRows(size_t num_rows) { num_rows_ = num_rows; }
 

src/include/optimizer/optimize_context.h

@@ -12,9 +12,12 @@
 
 #pragma once
 
+#include "common/timer.h"
+
 #include "optimizer/property_set.h"
 #include "optimizer/optimizer_task.h"
 #include "optimizer/optimizer_task_pool.h"
+#include "settings/settings_manager.h"
 
 namespace peloton {
 namespace optimizer {

src/include/optimizer/optimizer.h

@@ -124,6 +124,19 @@ class Optimizer : public AbstractOptimizer {
       GroupID id, std::shared_ptr<PropertySet> required_props,
       std::vector<expression::AbstractExpression *> required_cols);
 
+  /* ExecuteTaskStack - Execute elements of given optimization task stack
+   * and ensure that we do not go beyond the time limit (unless if one plan has
+   *    not been generated yet)
+   *
+   * task_stack: the optimizer's task stack to iterate through
+   * root_group_id: the root group id to check if we have generated a plan or
+   *not
+   * root_context: the OptimizerContext to use that maintains required
+   *properties
+   */
+  void ExecuteTaskStack(OptimizerTaskStack &task_stack, int root_group_id,
+                        std::shared_ptr<OptimizeContext> root_context);
+
   //////////////////////////////////////////////////////////////////////////////
   /// Metadata
   OptimizerMetadata metadata_;

src/include/optimizer/optimizer_metadata.h

@@ -11,9 +11,12 @@
 //===----------------------------------------------------------------------===//
 
 #pragma once
+
+#include "common/timer.h"
 #include "optimizer/memo.h"
 #include "optimizer/group_expression.h"
 #include "optimizer/rule.h"
+#include "settings/settings_manager.h"
 
 namespace peloton {
 namespace catalog {
@@ -26,9 +29,17 @@ class RuleSet;
 
 class OptimizerMetadata {
  public:
+  OptimizerMetadata()
+      : timeout_limit(settings::SettingsManager::GetInt(
+            settings::SettingId::task_execution_timeout)),
+        timer(Timer<std::milli>()) {}
+
   Memo memo;
   RuleSet rule_set;
   OptimizerTaskPool *task_pool;
+  catalog::CatalogCache *catalog_cache;
+  unsigned int timeout_limit;
+  Timer<std::milli> timer;
   concurrency::TransactionContext* txn;
 
   void SetTaskPool(OptimizerTaskPool *task_pool) {

src/include/settings/settings.h

@@ -10,7 +10,6 @@
 //
 //===----------------------------------------------------------------------===//
 
-
 //===----------------------------------------------------------------------===//
 // FILE LOCATIONS
 //===----------------------------------------------------------------------===//
@@ -196,9 +195,12 @@ SETTING_bool(hash_join_bloom_filter,
              true,
              true, true)
 
+SETTING_int(task_execution_timeout,
+            "Maximum allowed length of time (in ms) for task "
+                "execution step of optimizer, "
+                "assuming one plan has been found (default 5000)",
+            5000, true, true)
+
 //===----------------------------------------------------------------------===//
 // GENERAL
 //===----------------------------------------------------------------------===//
-
-
-

src/optimizer/group.cpp

@@ -62,6 +62,12 @@ GroupExpression *Group::GetBestExpression(
   return nullptr;
 }
 
+bool Group::HasExpressions(
+    const std::shared_ptr<PropertySet> &properties) const {
+  const auto &it = lowest_cost_expressions_.find(properties);
+  return (it != lowest_cost_expressions_.end());
+}
+
 std::shared_ptr<ColumnStats> Group::GetStats(std::string column_name) {
   if (!stats_.count(column_name)) {
     return nullptr;

src/optimizer/optimizer.cpp

@@ -18,6 +18,8 @@
 #include "catalog/manager.h"
 #include "catalog/table_catalog.h"
 
+#include "common/exception.h"
+
 #include "optimizer/binding.h"
 #include "optimizer/operator_visitor.h"
 #include "optimizer/properties.h"
@@ -74,24 +76,19 @@ void Optimizer::OptimizeLoop(int root_group_id,
 
   task_stack->Push(new BottomUpRewrite(
       root_group_id, root_context, RewriteRuleSetName::UNNEST_SUBQUERY, false));
-  while (!task_stack->Empty()) {
-    auto task = task_stack->Pop();
-    task->execute();
-  }
+
+  ExecuteTaskStack(*task_stack, root_group_id, root_context);
 
   // Perform optimization after the rewrite
   task_stack->Push(new OptimizeGroup(metadata_.memo.GetGroupByID(root_group_id),
                                      root_context));
+
   // Derive stats for the only one logical expression before optimizing
   task_stack->Push(new DeriveStats(
       metadata_.memo.GetGroupByID(root_group_id)->GetLogicalExpression(),
       ExprSet{}, root_context));
 
-  // TODO: Add timer for early stop
-  while (!task_stack->Empty()) {
-    auto task = task_stack->Pop();
-    task->execute();
-  }
+  ExecuteTaskStack(*task_stack, root_group_id, root_context);
 }
 
 shared_ptr<planner::AbstractPlan> Optimizer::BuildPelotonPlanTree(
@@ -124,7 +121,11 @@ shared_ptr<planner::AbstractPlan> Optimizer::BuildPelotonPlanTree(
   // Get the physical properties the final plan must output
   auto query_info = GetQueryInfo(parse_tree);
 
-  OptimizeLoop(root_id, query_info.physical_props);
+  try {
+    OptimizeLoop(root_id, query_info.physical_props);
+  } catch (OptimizerException &e) {
+    LOG_WARN("Optimize Loop ended prematurely: %s", e.what());
+  }
 
   try {
     auto best_plan = ChooseBestPlan(root_id, query_info.physical_props,
@@ -240,29 +241,29 @@ shared_ptr<GroupExpression> Optimizer::InsertQueryTree(
 }
 
 QueryInfo Optimizer::GetQueryInfo(parser::SQLStatement *tree) {
-  auto GetQueryInfoHelper = [](
-      std::vector<unique_ptr<expression::AbstractExpression>> &select_list,
-      std::unique_ptr<parser::OrderDescription> &order_info,
-      std::vector<expression::AbstractExpression *> &output_exprs,
-      std::shared_ptr<PropertySet> &physical_props) {
-    // Extract output column
-    for (auto &expr : select_list) output_exprs.push_back(expr.get());
-
-    // Extract sort property
-    if (order_info != nullptr) {
-      std::vector<expression::AbstractExpression *> sort_exprs;
-      std::vector<bool> sort_ascending;
-      for (auto &expr : order_info->exprs) {
-        sort_exprs.push_back(expr.get());
-      }
-      for (auto &type : order_info->types) {
-        sort_ascending.push_back(type == parser::kOrderAsc);
-      }
-      if (!sort_exprs.empty())
-        physical_props->AddProperty(
-            std::make_shared<PropertySort>(sort_exprs, sort_ascending));
-    }
-  };
+  auto GetQueryInfoHelper =
+      [](std::vector<unique_ptr<expression::AbstractExpression>> &select_list,
+         std::unique_ptr<parser::OrderDescription> &order_info,
+         std::vector<expression::AbstractExpression *> &output_exprs,
+         std::shared_ptr<PropertySet> &physical_props) {
+        // Extract output column
+        for (auto &expr : select_list) output_exprs.push_back(expr.get());
+
+        // Extract sort property
+        if (order_info != nullptr) {
+          std::vector<expression::AbstractExpression *> sort_exprs;
+          std::vector<bool> sort_ascending;
+          for (auto &expr : order_info->exprs) {
+            sort_exprs.push_back(expr.get());
+          }
+          for (auto &type : order_info->types) {
+            sort_ascending.push_back(type == parser::kOrderAsc);
+          }
+          if (!sort_exprs.empty())
+            physical_props->AddProperty(
+                std::make_shared<PropertySort>(sort_exprs, sort_ascending));
+        }
+      };
 
   std::vector<expression::AbstractExpression *> output_exprs;
   std::shared_ptr<PropertySet> physical_props = std::make_shared<PropertySet>();
@@ -280,7 +281,8 @@ QueryInfo Optimizer::GetQueryInfo(parser::SQLStatement *tree) {
                            output_exprs, physical_props);
       break;
     }
-    default:;
+    default:
+      ;
   }
 
   return QueryInfo(output_exprs, physical_props);
@@ -336,5 +338,35 @@ unique_ptr<planner::AbstractPlan> Optimizer::ChooseBestPlan(
   LOG_TRACE("Finish Choosing best plan for group %d", id);
   return plan;
 }
+
+void Optimizer::ExecuteTaskStack(
+    OptimizerTaskStack &task_stack, int root_group_id,
+    std::shared_ptr<OptimizeContext> root_context) {
+  auto root_group = metadata_.memo.GetGroupByID(root_group_id);
+  auto &timer = metadata_.timer;
+  const auto timeout_limit = metadata_.timeout_limit;
+  const auto &required_props = root_context->required_prop;
+
+  if (timer.GetInvocations() == 0) {
+    timer.Start();
+  }
+  // Iterate through the task stack
+  while (!task_stack.Empty()) {
+    // Check to see if we have at least one plan, and if we have exceeded our
+    // timeout limit
+    if (timer.GetDuration() >= timeout_limit &&
+        root_group->HasExpressions(required_props)) {
+      throw OptimizerException("Optimizer task execution duration " +
+                               std::to_string(timer.GetDuration()) +
+                               " exceeds timeout limit " +
+                               std::to_string(timeout_limit));
+    }
+    timer.Reset();
+    auto task = task_stack.Pop();
+    task->execute();
+    timer.Stop();
+  }
+}
+
 }  // namespace optimizer
 }  // namespace peloton

src/settings/settings_manager.cpp

@@ -92,17 +92,30 @@ const std::string SettingsManager::GetInfo() const {
   const std::string title = "PELOTON SETTINGS";
 
   std::string info;
-  info.append(StringUtil::Format("%*s\n", box_width/2 + title.length()/2, title.c_str()));
+  info.append(StringUtil::Format("%*s\n", box_width / 2 + title.length() / 2,
+                                 title.c_str()));
   info.append(StringUtil::Repeat("=", box_width)).append("\n");
 
-  info.append(StringUtil::Format("%28s:   %-28i\n", "Port", GetInt(SettingId::port)));
-  info.append(StringUtil::Format("%28s:   %-28s\n", "Socket Family", GetString(SettingId::socket_family).c_str()));
-  info.append(StringUtil::Format("%28s:   %-28s\n", "Statistics", GetInt(SettingId::stats_mode) ? "enabled" : "disabled"));
-  info.append(StringUtil::Format("%28s:   %-28i\n", "Max Connections", GetInt(SettingId::max_connections)));
-  info.append(StringUtil::Format("%28s:   %-28s\n", "Index Tuner", GetBool(SettingId::index_tuner) ? "enabled" : "disabled"));
-  info.append(StringUtil::Format("%28s:   %-28s\n", "Layout Tuner", GetBool(SettingId::layout_tuner) ? "enabled" : "disabled"));
-  info.append(StringUtil::Format("%28s:   %-28s\n", "Code-generation", GetBool(SettingId::codegen) ? "enabled" : "disabled"));
-
+  info.append(
+      StringUtil::Format("%28s:   %-28i\n", "Port", GetInt(SettingId::port)));
+  info.append(StringUtil::Format("%28s:   %-28s\n", "Socket Family",
+                                 GetString(SettingId::socket_family).c_str()));
+  info.append(StringUtil::Format(
+      "%28s:   %-28s\n", "Statistics",
+      GetInt(SettingId::stats_mode) ? "enabled" : "disabled"));
+  info.append(StringUtil::Format("%28s:   %-28i\n", "Max Connections",
+                                 GetInt(SettingId::max_connections)));
+  info.append(StringUtil::Format(
+      "%28s:   %-28s\n", "Index Tuner",
+      GetBool(SettingId::index_tuner) ? "enabled" : "disabled"));
+  info.append(StringUtil::Format(
+      "%28s:   %-28s\n", "Layout Tuner",
+      GetBool(SettingId::layout_tuner) ? "enabled" : "disabled"));
+  info.append(
+      StringUtil::Format("%28s:   %-28s\n", "Code-generation",
+                         GetBool(SettingId::codegen) ? "enabled" : "disabled"));
+  info.append(StringUtil::Format("%28s:   %-28s\n", "Optimization Timeout",
+                                 GetInt(SettingId::task_execution_timeout)));
   return StringBoxUtil::Box(info);
 }
 
@@ -164,12 +177,12 @@ SettingsManager::SettingsManager() {
   catalog_initialized_ = false;
   pool_.reset(new type::EphemeralPool());
 
-  // This will expand to invoke SettingsManager::DefineSetting on
-  // all of the settings defined in settings.h. See settings_macro.h.
-  #define __SETTING_DEFINE__
-  #include "settings/settings_macro.h"
-  #include "settings/settings.h"
-  #undef __SETTING_DEFINE__
+// This will expand to invoke SettingsManager::DefineSetting on
+// all of the settings defined in settings.h. See settings_macro.h.
+#define __SETTING_DEFINE__
+#include "settings/settings_macro.h"
+#include "settings/settings.h"
+#undef __SETTING_DEFINE__
 }
 
 }  // namespace settings

test/include/optimizer/mock_task.h

@@ -0,0 +1,33 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// mock_task.h
+//
+// Identification: test/include/optimizer/mock_task.h
+//
+// Copyright (c) 2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "gmock/gmock.h"
+
+#include "optimizer/optimizer_task.h"
+
+namespace peloton {
+namespace optimizer {
+namespace test {
+
+class MockTask : public optimizer::OptimizerTask {
+ public:
+  MockTask()
+      : optimizer::OptimizerTask(nullptr, OptimizerTaskType::OPTIMIZE_GROUP) {}
+
+  MOCK_METHOD0(execute, void());
+};
+
+}  // namespace test
+}  // namespace optimizer
+}  // namespace peloton