Cleaned up query interface to be synchronous.

Author: pmenon

src/codegen/query.cpp

@@ -24,15 +24,14 @@ namespace codegen {
 Query::Query(const planner::AbstractPlan &query_plan)
     : query_plan_(query_plan) {}
 
-void Query::Execute(std::unique_ptr<executor::ExecutorContext> executor_context,
-                    ExecutionConsumer &consumer,
-                    std::function<void(executor::ExecutionResult)> on_complete,
-                    RuntimeStats *stats) {
+void Query::Execute(executor::ExecutorContext &executor_context,
+                    ExecutionConsumer &consumer, RuntimeStats *stats) {
   CodeGen codegen{code_context_};
 
   llvm::Type *query_state_type = query_state_.GetType();
   size_t parameter_size = codegen.SizeOf(query_state_type);
-  PELOTON_ASSERT((parameter_size % 8 == 0) && "parameter size not multiple of 8");
+  PELOTON_ASSERT((parameter_size % 8 == 0) &&
+                 "parameter size not multiple of 8");
 
   // Allocate some space for the function arguments
   std::unique_ptr<char[]> param_data{new char[parameter_size]};
@@ -41,18 +40,14 @@ void Query::Execute(std::unique_ptr<executor::ExecutorContext> executor_context,
 
   // We use this handy class to avoid complex casting and pointer manipulation
   struct FunctionArguments {
-    // storage::StorageManager *storage_manager;
     executor::ExecutorContext *executor_context;
-    // QueryParameters *query_parameters;
     char *consumer_arg;
     char rest[0];
   } PACKED;
 
   // Set up the function arguments
   auto *func_args = reinterpret_cast<FunctionArguments *>(param_data.get());
-  // func_args->storage_manager = storage::StorageManager::GetInstance();
-  func_args->executor_context = executor_context.get();
-  // func_args->query_parameters = &executor_context->GetParams();
+  func_args->executor_context = &executor_context;
   func_args->consumer_arg = consumer.GetConsumerState();
 
   // Timer
@@ -104,11 +99,6 @@ void Query::Execute(std::unique_ptr<executor::ExecutorContext> executor_context,
     timer.Stop();
     stats->tear_down_ms = timer.GetDuration();
   }
-
-  executor::ExecutionResult result;
-  result.m_result = ResultType::SUCCESS;
-  result.m_processed = executor_context->num_processed;
-  on_complete(result);
 }
 
 bool Query::Prepare(const QueryFunctions &query_funcs) {

src/executor/plan_executor.cpp

@@ -49,36 +49,47 @@ static void CompileAndExecutePlan(
   plan->GetOutputColumns(columns);
   codegen::BufferingConsumer consumer{columns, context};
 
-  std::unique_ptr<executor::ExecutorContext> executor_context(
-      new executor::ExecutorContext(txn,
-                                    codegen::QueryParameters(*plan, params)));
+  // The executor context for this execution
+  executor::ExecutorContext executor_context{
+      txn, codegen::QueryParameters(*plan, params)};
 
-  // Compile the query
+  // Check if we have a cached compiled plan already
   codegen::Query *query = codegen::QueryCache::Instance().Find(plan);
   if (query == nullptr) {
+    // Cached plan doesn't exist, let's compile the query
     codegen::QueryCompiler compiler;
     auto compiled_query = compiler.Compile(
-        *plan, executor_context->GetParams().GetQueryParametersMap(), consumer);
+        *plan, executor_context.GetParams().GetQueryParametersMap(), consumer);
+
+    // Grab an instance to the plan
     query = compiled_query.get();
+
+    // Insert the compiled plan into the cache
     codegen::QueryCache::Instance().Add(plan, std::move(compiled_query));
   }
 
-  auto on_query_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++) {
-            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));
-          }
-        }
-        plan->ClearParameterValues();
-        on_complete(result, std::move(values));
-      };
+  // Execute the query!
+  query->Execute(executor_context, consumer);
+
+  // Execution complete, setup the results
+  executor::ExecutionResult result;
+  result.m_processed = executor_context.num_processed;
+  result.m_result = ResultType::SUCCESS;
 
-  query->Execute(std::move(executor_context), consumer, on_query_result);
+  // Iterate over results
+  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));
+    }
+  }
+
+  // Done, invoke callback
+  plan->ClearParameterValues();
+  on_complete(result, std::move(values));
 }
 
 static void InterpretPlan(

src/include/codegen/query.h

@@ -66,19 +66,18 @@ class Query {
   /**
    * @brief Executes the compiled query.
    *
-   * This function is **asynchronous** - it returns before execution completes,
-   * and invokes a user-provided callback on completion. It is the user's
-   * responsibility that the result consumer object has a lifetime as far as
-   * the return of the callback.
-   *
    * @param executor_context Stores transaction and parameters.
    * @param consumer Stores the result.
-   * @param on_complete The callback to be invoked when execution completes.
+   * @param stats Handy struct to collect various runtime timing statistics
    */
-  void Execute(std::unique_ptr<executor::ExecutorContext> executor_context,
-               ExecutionConsumer &consumer,
-               std::function<void(executor::ExecutionResult)> on_complete,
-               RuntimeStats *stats = nullptr);
+  void Execute(executor::ExecutorContext &executor_context,
+               ExecutionConsumer &consumer, RuntimeStats *stats = nullptr);
+
+  //////////////////////////////////////////////////////////////////////////////
+  ///
+  /// Accessors
+  ///
+  //////////////////////////////////////////////////////////////////////////////
 
   /// Return the query plan
   const planner::AbstractPlan &GetPlan() const { return query_plan_; }
@@ -92,7 +91,7 @@ class Query {
  private:
   friend class QueryCompiler;
 
-  /// Constructor
+  /// Constructor. Private so callers use the QueryCompiler class.
   explicit Query(const planner::AbstractPlan &query_plan);
 
  private:

test/codegen/bloom_filter_test.cpp

@@ -36,7 +36,7 @@
 namespace peloton {
 namespace test {
 
-class BloomFilterCodegenTest : public PelotonTest {
+class BloomFilterCodegenTest : public PelotonCodeGenTest {
  public:
   BloomFilterCodegenTest() {
     // Create test db
@@ -299,19 +299,20 @@ double BloomFilterCodegenTest::ExecuteJoin(std::string query,
     // Binding
     planner::BindingContext context;
     plan->PerformBinding(context);
+
+    executor::ExecutorContext executor_context{txn};
+
     // Use simple CountConsumer since we don't care about the result
     codegen::CountingConsumer consumer;
+
     // Compile the query
     codegen::QueryCompiler compiler;
     codegen::Query::RuntimeStats stats;
-    std::unique_ptr<executor::ExecutorContext> executor_context(
-        new executor::ExecutorContext{txn});
     auto compiled_query = compiler.Compile(
-        *plan, executor_context->GetParams().GetQueryParametersMap(), consumer);
+        *plan, executor_context.GetParams().GetQueryParametersMap(), consumer);
 
     // Run
-    PelotonCodeGenTest::ExecuteSync(*compiled_query,
-                                    std::move(executor_context), consumer);
+    compiled_query->Execute(executor_context, consumer, &stats);
 
     LOG_INFO("Execution Time: %0.0f ms", stats.plan_ms);
     total_runtime += stats.plan_ms;

test/codegen/testing_codegen_util.cpp

@@ -252,24 +252,6 @@ void PelotonCodeGenTest::CreateAndLoadTableWithLayout(
   txn_manager.CommitTransaction(txn);
 }
 
-void PelotonCodeGenTest::ExecuteSync(
-    codegen::Query &query,
-    std::unique_ptr<executor::ExecutorContext> executor_context,
-    codegen::ExecutionConsumer &consumer) {
-  std::mutex mu;
-  std::condition_variable cond;
-  bool finished = false;
-  query.Execute(std::move(executor_context), consumer,
-                [&](executor::ExecutionResult) {
-                  std::unique_lock<decltype(mu)> lock(mu);
-                  finished = true;
-                  cond.notify_one();
-                });
-
-  std::unique_lock<decltype(mu)> lock(mu);
-  cond.wait(lock, [&] { return finished; });
-}
-
 codegen::QueryCompiler::CompileStats PelotonCodeGenTest::CompileAndExecute(
     planner::AbstractPlan &plan, codegen::ExecutionConsumer &consumer) {
   codegen::QueryParameters parameters(plan, {});
@@ -280,14 +262,14 @@ codegen::QueryCompiler::CompileStats PelotonCodeGenTest::CompileAndExecute(
 
   // Compile the query.
   codegen::QueryCompiler::CompileStats stats;
-  auto compiled_query = codegen::QueryCompiler().Compile(
+  auto query = codegen::QueryCompiler().Compile(
       plan, parameters.GetQueryParametersMap(), consumer, &stats);
 
-  // Execute the query.
-  ExecuteSync(*compiled_query,
-              std::unique_ptr<executor::ExecutorContext>(
-                  new executor::ExecutorContext(txn, std::move(parameters))),
-              consumer);
+  // Executor context
+  executor::ExecutorContext exec_ctx{txn, std::move(parameters)};
+
+  // Execute the query
+  query->Execute(exec_ctx, consumer);
 
   // Commit the transaction.
   txn_manager.CommitTransaction(txn);
@@ -303,9 +285,8 @@ codegen::QueryCompiler::CompileStats PelotonCodeGenTest::CompileAndExecuteCache(
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
   auto *txn = txn_manager.BeginTransaction();
 
-  std::unique_ptr<executor::ExecutorContext> executor_context(
-      new executor::ExecutorContext(txn,
-                                    codegen::QueryParameters(*plan, params)));
+  executor::ExecutorContext exec_ctx{txn,
+                                     codegen::QueryParameters(*plan, params)};
 
   // Compile
   codegen::QueryCompiler::CompileStats stats;
@@ -314,13 +295,13 @@ codegen::QueryCompiler::CompileStats PelotonCodeGenTest::CompileAndExecuteCache(
   if (query == nullptr) {
     codegen::QueryCompiler compiler;
     auto compiled_query = compiler.Compile(
-        *plan, executor_context->GetParams().GetQueryParametersMap(), consumer);
+        *plan, exec_ctx.GetParams().GetQueryParametersMap(), consumer);
     query = compiled_query.get();
     codegen::QueryCache::Instance().Add(plan, std::move(compiled_query));
   }
 
   // Execute the query.
-  ExecuteSync(*query, std::move(executor_context), consumer);
+  query->Execute(exec_ctx, consumer);
 
   // Commit the transaction.
   txn_manager.CommitTransaction(txn);

test/include/codegen/testing_codegen_util.h

@@ -89,11 +89,6 @@ class PelotonCodeGenTest : public PelotonTest {
                                     oid_t tile_group_count, oid_t column_count,
                                     bool is_inlined);
 
-  static void ExecuteSync(
-      codegen::Query &query,
-      std::unique_ptr<executor::ExecutorContext> executor_context,
-      codegen::ExecutionConsumer &consumer);
-
   // Compile and execute the given plan
   codegen::QueryCompiler::CompileStats CompileAndExecute(
       planner::AbstractPlan &plan, codegen::ExecutionConsumer &consumer);