done throwing error handling in execution

Author: Tianyi Chen

src/common/init.cpp

@@ -12,14 +12,14 @@
 
 #include "common/init.h"
 
-#include <google/protobuf/stubs/common.h>
 #include <gflags/gflags.h>
+#include <google/protobuf/stubs/common.h>
 
 #include "brain/index_tuner.h"
 #include "brain/layout_tuner.h"
 #include "catalog/catalog.h"
-#include "common/thread_pool.h"
 #include "common/statement_cache_manager.h"
+#include "common/thread_pool.h"
 #include "concurrency/transaction_manager_factory.h"
 #include "gc/gc_manager_factory.h"
 #include "settings/settings_manager.h"
@@ -83,7 +83,7 @@ void PelotonInit::Initialize() {
   // initialize the catalog and add the default database, so we don't do this on
   // the first query
   pg_catalog->CreateDatabase(DEFAULT_DB_NAME, txn);
-
+  
   txn_manager.CommitTransaction(txn);
 
   // Initialize the Statement Cache Manager

src/executor/plan_executor.cpp

@@ -15,7 +15,6 @@
 #include "codegen/buffering_consumer.h"
 #include "codegen/query.h"
 #include "codegen/query_cache.h"
-#include "concurrency/transaction_manager_factory.h"
 #include "codegen/query_compiler.h"
 #include "common/logger.h"
 #include "concurrency/transaction_manager_factory.h"
@@ -64,19 +63,19 @@ static void CompileAndExecutePlan(
     codegen::QueryCache::Instance().Add(plan, std::move(compiled_query));
   }
 
-  auto on_query_result =
-      [&on_complete, &consumer](executor::ExecutionResult result) {
-        std::vector<ResultValue> values;
-        for (const auto &tuple : consumer.GetOutputTuples()) {
-          for (uint32_t i = 0; i < tuple.tuple_.size(); i++) {
-            auto column_val = tuple.GetValue(i);
-            auto str = column_val.IsNull() ? "" : column_val.ToString();
-            LOG_TRACE("column content: [%s]", str.c_str());
-            values.push_back(std::move(str));
-          }
-        }
-        on_complete(result, std::move(values));
-      };
+  auto on_query_result = [&on_complete,
+                          &consumer](executor::ExecutionResult result) {
+    std::vector<ResultValue> values;
+    for (const auto &tuple : consumer.GetOutputTuples()) {
+      for (uint32_t i = 0; i < tuple.tuple_.size(); i++) {
+        auto column_val = tuple.GetValue(i);
+        auto str = column_val.IsNull() ? "" : column_val.ToString();
+        LOG_TRACE("column content: [%s]", str.c_str());
+        values.push_back(std::move(str));
+      }
+    }
+    on_complete(result, std::move(values));
+  };
 
   query->Execute(std::move(executor_context), consumer, on_query_result);
 }
@@ -101,6 +100,7 @@ static void InterpretPlan(
   status = executor_tree->Init();
   if (status != true) {
     result.m_result = ResultType::FAILURE;
+    result.m_error_message = "Executor tree init failed";
     CleanExecutorTree(executor_tree.get());
     on_complete(result, std::move(values));
     return;
@@ -146,16 +146,18 @@ void PlanExecutor::ExecutePlan(
 
   bool codegen_enabled =
       settings::SettingsManager::GetBool(settings::SettingId::codegen);
-  
+
   try {
     if (codegen_enabled && codegen::QueryCompiler::IsSupported(*plan)) {
-      CompileAndExecutePlan(plan, txn, params, std::move(on_complete));
+      CompileAndExecutePlan(plan, txn, params, on_complete);
     } else {
-      InterpretPlan(plan, txn, params, result_format, std::move(on_complete));
+      InterpretPlan(plan, txn, params, result_format, on_complete);
     }
   } catch (Exception &e) {
     ExecutionResult result;
     result.m_result = ResultType::FAILURE;
+    result.m_error_message = e.what();
+    LOG_ERROR("error thrown in Execution: %s", result.m_error_message.c_str());
     on_complete(result, {});
   }
 }

src/include/executor/plan_executor.h

@@ -12,9 +12,9 @@
 
 #pragma once
 
+#include "common/internal_types.h"
 #include "common/statement.h"
 #include "executor/abstract_executor.h"
-#include "common/internal_types.h"
 
 namespace peloton {
 
@@ -34,9 +34,13 @@ struct ExecutionResult {
   // number of tuples processed
   uint32_t m_processed;
 
+  // string of error message
+  std::string m_error_message;
+
   ExecutionResult() {
     m_processed = 0;
     m_result = ResultType::SUCCESS;
+    m_error_message = "";
   }
 };
 
@@ -51,13 +55,13 @@ class PlanExecutor {
    * Before ExecutePlan, a node first receives value list, so we should
    * pass value list directly rather than passing Postgres's ParamListInfo
    */
-  static void ExecutePlan(
-      std::shared_ptr<planner::AbstractPlan> plan,
-      concurrency::TransactionContext *txn,
-      const std::vector<type::Value> &params,
-      const std::vector<int> &result_format,
-      std::function<void(executor::ExecutionResult,
-                         std::vector<ResultValue> &&)> on_complete);
+  static void ExecutePlan(std::shared_ptr<planner::AbstractPlan> plan,
+                          concurrency::TransactionContext *txn,
+                          const std::vector<type::Value> &params,
+                          const std::vector<int> &result_format,
+                          std::function<void(executor::ExecutionResult,
+                                             std::vector<ResultValue> &&)>
+                              on_complete);
 
   /*
    * @brief When a peloton node recvs a query plan, this function is invoked

src/include/traffic_cop/traffic_cop.h

@@ -133,10 +133,6 @@ class TrafficCop {
 
   std::vector<type::Value> &GetParamVal() { return param_values_; }
 
-  void SetErrorMessage(std::string error_message) {
-    error_message_ = std::move(error_message);
-  }
-
   std::string &GetErrorMessage() { return error_message_; }
 
   void SetQueuing(bool is_queuing) { is_queuing_ = is_queuing; }

src/network/postgres_protocol_handler.cpp

@@ -306,9 +306,9 @@ ProcessResult PostgresProtocolHandler::ExecQueryMessage(
       }
       // Did not find statement with same name
       else {
-        traffic_cop_->SetErrorMessage("The prepared statement does not exist");
-        SendErrorResponse({{NetworkMessageType::HUMAN_READABLE_ERROR,
-                            traffic_cop_->GetErrorMessage()}});
+        std::string error_message = "The prepared statement does not exist";
+        SendErrorResponse(
+            {{NetworkMessageType::HUMAN_READABLE_ERROR, error_message}});
         SendReadyForQuery(NetworkTransactionStateType::IDLE);
         return ProcessResult::COMPLETE;
       }
@@ -373,11 +373,11 @@ void PostgresProtocolHandler::ExecQueryMessageGetResult(ResultType status) {
     SendReadyForQuery(NetworkTransactionStateType::IDLE);
     return;
   } else if (status == ResultType::TO_ABORT) {
-    traffic_cop_->SetErrorMessage(
+    std::string error_message =
         "current transaction is aborted, commands ignored until end of "
-        "transaction block");
-    SendErrorResponse({{NetworkMessageType::HUMAN_READABLE_ERROR,
-                        traffic_cop_->GetErrorMessage()}});
+        "transaction block";
+    SendErrorResponse(
+        {{NetworkMessageType::HUMAN_READABLE_ERROR, error_message}});
     SendReadyForQuery(NetworkTransactionStateType::IDLE);
     return;
   }

src/planner/insert_plan.cpp

@@ -37,7 +37,9 @@ InsertPlan::InsertPlan(storage::DataTable *table,
     for (uint32_t tuple_idx = 0; tuple_idx < insert_values->size();
          tuple_idx++) {
       auto &values = (*insert_values)[tuple_idx];
-      PL_ASSERT(values.size() <= schema->GetColumnCount());
+      if (values.size() > schema->GetColumnCount()) {
+        throw Exception("Column size does not match");
+      }
       uint32_t param_idx = 0;
       for (uint32_t column_id = 0; column_id < values.size(); column_id++) {
         auto &exp = values[column_id];
@@ -66,7 +68,9 @@ InsertPlan::InsertPlan(storage::DataTable *table,
   }
   // INSERT INTO table_name (col1, col2, ...) VALUES (val1, val2, ...);
   else {
-    PL_ASSERT(columns->size() <= schema->GetColumnCount());
+    if (columns->size() > schema->GetColumnCount()) {
+      throw Exception("Column size does not match");
+    }
     for (uint32_t tuple_idx = 0; tuple_idx < insert_values->size();
          tuple_idx++) {
       auto &values = (*insert_values)[tuple_idx];

src/storage/data_table.cpp

@@ -318,14 +318,14 @@ bool DataTable::InstallVersion(const AbstractTuple *tuple,
                                ItemPointer *index_entry_ptr) {
   if (CheckConstraints(tuple) == false) {
     LOG_TRACE("InsertVersion(): Constraint violated");
-    return false;
+    throw Exception("Constraint violated");
   }
 
   // Index checks and updates
   if (InsertInSecondaryIndexes(tuple, targets_ptr, transaction,
                                index_entry_ptr) == false) {
     LOG_TRACE("Index constraint violated");
-    return false;
+    throw Exception("Constraint violated");
   }
   return true;
 }

src/traffic_cop/traffic_cop.cpp

@@ -101,7 +101,7 @@ ResultType TrafficCop::CommitQueryHelper() {
   auto txn = curr_state.first;
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
   // I catch the exception (ex. table not found) explicitly,
-  // If this exception if caused by a query in a transaction,
+  // If this exception is caused by a query in a transaction,
   // I will block following queries in that transaction until 'COMMIT' or
   // 'ROLLBACK' After receive 'COMMIT', see if it is rollback or really commit.
   if (curr_state.second != ResultType::ABORTED) {
@@ -183,6 +183,9 @@ executor::ExecutionResult TrafficCop::ExecuteHelper(
   auto on_complete = [&result, this](executor::ExecutionResult p_status,
                                      std::vector<ResultValue> &&values) {
     this->p_status_ = p_status;
+    // TODO (Tianyi) I would make a decision on keeping one of p_status or
+    // error_message in my next PR
+    this->error_message_ = std::move(p_status.m_error_message);
     result = std::move(values);
     task_callback_(task_callback_arg_);
   };
@@ -201,19 +204,13 @@ executor::ExecutionResult TrafficCop::ExecuteHelper(
 }
 
 void TrafficCop::ExecuteStatementPlanGetResult() {
-  bool init_failure = false;
-  if (p_status_.m_result == ResultType::FAILURE) {
-    // only possible if init failed
-    init_failure = true;
-  }
+  if (p_status_.m_result == ResultType::FAILURE)
+    return;
 
   auto txn_result = GetCurrentTxnState().first->GetResult();
-  if (single_statement_txn_ || init_failure ||
-      txn_result == ResultType::FAILURE) {
-    LOG_TRACE(
-        "About to commit: single stmt: %d, init_failure: %d, txn_result: %s",
-        single_statement_txn_, init_failure,
-        ResultTypeToString(txn_result).c_str());
+  if (single_statement_txn_ || txn_result == ResultType::FAILURE) {
+    LOG_TRACE("About to commit/abort: single stmt: %d,txn_result: %s",
+              single_statement_txn_, ResultTypeToString(txn_result).c_str());
     switch (txn_result) {
       case ResultType::SUCCESS:
         // Commit single statement