Clear insert plan value_ vector after use. Fixes issue #1151.

Resolves large performance degradation of inserts, after inclusion
of statement/plan cache.

Author: pervazea

src/executor/plan_executor.cpp

@@ -65,7 +65,7 @@ static void CompileAndExecutePlan(
   }
 
   auto on_query_result =
-      [&on_complete, &consumer](executor::ExecutionResult result) {
+    [&on_complete, &consumer, plan](executor::ExecutionResult result) {
         std::vector<ResultValue> values;
         for (const auto &tuple : consumer.GetOutputTuples()) {
           for (uint32_t i = 0; i < tuple.tuple_.size(); i++) {
@@ -75,6 +75,7 @@ static void CompileAndExecutePlan(
             values.push_back(std::move(str));
           }
         }
+	plan->ClearParameterValues();
         on_complete(result, std::move(values));
       };
 
@@ -102,6 +103,7 @@ static void InterpretPlan(
   if (status != true) {
     result.m_result = ResultType::FAILURE;
     CleanExecutorTree(executor_tree.get());
+    plan->ClearParameterValues();
     on_complete(result, std::move(values));
     return;
   }
@@ -131,6 +133,7 @@ static void InterpretPlan(
   result.m_processed = executor_context->num_processed;
   result.m_result = ResultType::SUCCESS;
   CleanExecutorTree(executor_tree.get());
+  plan->ClearParameterValues();
   on_complete(result, std::move(values));
 }
 
@@ -164,7 +167,7 @@ void PlanExecutor::ExecutePlan(
  * @return number of executed tuples and logical_tile_list
  */
 int PlanExecutor::ExecutePlan(
-    const planner::AbstractPlan *plan, const std::vector<type::Value> &params,
+    planner::AbstractPlan *plan, const std::vector<type::Value> &params,
     std::vector<std::unique_ptr<executor::LogicalTile>> &logical_tile_list) {
   PL_ASSERT(plan != nullptr);
   LOG_TRACE("PlanExecutor Start with transaction");

src/include/executor/plan_executor.h

@@ -66,7 +66,7 @@ class PlanExecutor {
    */
   // FIXME This should be removed when PelotonService is removed/rewritten
   static int ExecutePlan(
-      const planner::AbstractPlan *plan, const std::vector<type::Value> &params,
+      planner::AbstractPlan *plan, const std::vector<type::Value> &params,
       std::vector<std::unique_ptr<executor::LogicalTile>> &logical_tile_list);
 };
 

src/include/planner/abstract_plan.h

@@ -78,6 +78,9 @@ class AbstractPlan : public Printable {
 
   // Setting values of the parameters in the prepare statement
   virtual void SetParameterValues(std::vector<type::Value> *values);
+
+  // FIXME. Clear the value_ vector.
+  virtual void ClearParameterValues() {};
 
   // Get the estimated cardinality of this plan
   int GetCardinality() const { return estimated_cardinality_; }

src/include/planner/insert_plan.h

@@ -69,6 +69,12 @@ class InsertPlan : public AbstractPlan {
 
   void SetParameterValues(std::vector<type::Value> *values) override;
 
+  /* 
+   * Clear the parameter values of the current insert. The plan may be 
+   * cached in the statement / plan cache and may be reused.
+   */
+  void ClearParameterValues() { values_.clear(); }
+
   storage::DataTable *GetTable() const { return target_table_; }
 
   const planner::ProjectInfo *GetProjectInfo() const {