[15721] Logging and Recovery

src/codegen/buffering_consumer.cpp

@@ -78,6 +78,46 @@ void BufferingConsumer::Prepare(CompilationContext &ctx) {
       runtime_state.RegisterState("consumerState", codegen.CharPtrType());
 }
 
+void BufferingConsumer::AddToTupleBuffer(Value &val, CodeGen &codegen, llvm::Value *tuple_buffer, size_t &i) {
+  const auto &sql_type = val.GetType().GetSqlType();
+
+  // Check if it's NULL
+  Value null_val;
+  lang::If val_is_null{codegen, val.IsNull(codegen)};
+  {
+    // If the value is NULL (i.e., has the NULL bit set), produce the NULL
+    // value for the given type.
+    null_val = sql_type.GetNullValue(codegen);
+  }
+  val_is_null.EndIf();
+  val = val_is_null.BuildPHI(null_val, val);
+
+  // Output the value using the type's output function
+  auto *output_func = sql_type.GetOutputFunction(codegen, val.GetType());
+
+  // Setup the function arguments
+  std::vector<llvm::Value *> args = {tuple_buffer, codegen.Const32(i),
+                                     val.GetValue()};
+  // If the value is a string, push back the length
+  if (val.GetLength() != nullptr) {
+    args.push_back(val.GetLength());
+  }
+
+  // If the value is a boolean, push back the NULL bit. We don't do that for
+  // the other data types because we have special values for NULL. Booleans
+  // in codegen are 1-bit types, as opposed to 1-byte types in the rest of the
+  // system. Since, we cannot have a special value for NULL in a 1-bit boolean
+  // system, we pass along the NULL bit during output.
+  if (sql_type.TypeId() == peloton::type::TypeId::BOOLEAN) {
+    args.push_back(val.IsNull(codegen));
+  }
+
+  // Call the function
+  codegen.CallFunc(output_func, args);
+}
+
+
+
 // For each output attribute, we write out the attribute's value into the
 // currently active output tuple. When all attributes have been written, we
 // call BufferTuple(...) to append the currently active tuple into the output.
@@ -95,41 +135,7 @@ void BufferingConsumer::ConsumeResult(ConsumerContext &ctx,
     Value val = row.DeriveValue(codegen, output_ais_[i]);
 
     PELOTON_ASSERT(output_ais_[i]->type == val.GetType());
-    const auto &sql_type = val.GetType().GetSqlType();
-
-    // Check if it's NULL
-    Value null_val;
-    lang::If val_is_null{codegen, val.IsNull(codegen)};
-    {
-      // If the value is NULL (i.e., has the NULL bit set), produce the NULL
-      // value for the given type.
-      null_val = sql_type.GetNullValue(codegen);
-    }
-    val_is_null.EndIf();
-    val = val_is_null.BuildPHI(null_val, val);
-
-    // Output the value using the type's output function
-    auto *output_func = sql_type.GetOutputFunction(codegen, val.GetType());
-
-    // Setup the function arguments
-    std::vector<llvm::Value *> args = {tuple_buffer_, codegen.Const32(i),
-                                       val.GetValue()};
-    // If the value is a string, push back the length
-    if (val.GetLength() != nullptr) {
-      args.push_back(val.GetLength());
-    }
-
-    // If the value is a boolean, push back the NULL bit. We don't do that for
-    // the other data types because we have special values for NULL. Booleans
-    // in codegen are 1-bit types, as opposed to 1-byte types in the rest of the
-    // system. Since, we cannot have a special value for NULL in a 1-bit boolean
-    // system, we pass along the NULL bit during output.
-    if (sql_type.TypeId() == peloton::type::TypeId::BOOLEAN) {
-      args.push_back(val.IsNull(codegen));
-    }
-
-    // Call the function
-    codegen.CallFunc(output_func, args);
+    AddToTupleBuffer(val, codegen, tuple_buffer_, i);
   }
 
   // Append the tuple to the output buffer (by calling BufferTuple(...))

src/codegen/inserter.cpp

@@ -14,12 +14,10 @@
 #include "codegen/transaction_runtime.h"
 #include "common/container_tuple.h"
 #include "concurrency/transaction_manager_factory.h"
-#include "executor/executor_context.h"
 #include "executor/logical_tile.h"
 #include "executor/logical_tile_factory.h"
-#include "storage/data_table.h"
-#include "storage/tile_group.h"
 #include "storage/tile.h"
+#include "threadpool/logger_queue_pool.h"
 
 namespace peloton {
 namespace codegen {
@@ -48,7 +46,7 @@ peloton::type::AbstractPool *Inserter::GetPool() {
   return tile_->GetPool();
 }
 
-void Inserter::Insert() {
+void Inserter::Insert(char *values_buf, uint32_t values_size) {
   PELOTON_ASSERT(table_ && executor_context_ && tile_);
   auto *txn = executor_context_->GetTransaction();
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
@@ -61,7 +59,7 @@ void Inserter::Insert() {
     txn_manager.SetTransactionResult(txn, ResultType::FAILURE);
     return;
   }
-  txn_manager.PerformInsert(txn, location_, index_entry_ptr);
+  txn_manager.PerformInsert(txn, location_, index_entry_ptr, values_buf, values_size);
   executor_context_->num_processed++;
 }
 

src/codegen/operator/insert_translator.cpp

@@ -10,11 +10,14 @@
 //
 //===----------------------------------------------------------------------===//
 
+#include "codegen/buffering_consumer.h"
 #include "codegen/proxy/inserter_proxy.h"
 #include "codegen/proxy/query_parameters_proxy.h"
 #include "codegen/proxy/storage_manager_proxy.h"
 #include "codegen/proxy/transaction_runtime_proxy.h"
 #include "codegen/proxy/tuple_proxy.h"
+#include "codegen/proxy/value_proxy.h"
+#include "codegen/proxy/values_runtime_proxy.h"
 #include "codegen/operator/insert_translator.h"
 #include "planner/insert_plan.h"
 #include "storage/data_table.h"
@@ -79,15 +82,23 @@ void InsertTranslator::Produce() const {
 
       // Transform into the codegen values and store values in the tuple storage
       std::vector<codegen::Value> values;
-      for (uint32_t column_id = 0; column_id < num_columns; column_id++) {
+      auto *values_buf = codegen.AllocateBuffer(
+          ValueProxy::GetType(codegen), num_columns,
+          "values");
+      values_buf =
+          codegen->CreatePointerCast(values_buf, codegen.CharPtrType());
+      llvm::Value *values_size = codegen.Const32((int32_t)num_columns);
+      for (size_t column_id = 0; column_id < num_columns; column_id++) {
         auto value =
             parameter_cache.GetValue(column_id + tuple_idx * num_columns);
         values.push_back(value);
+        peloton::codegen::BufferingConsumer::AddToTupleBuffer(value, codegen, values_buf, column_id);
       }
       table_storage_.StoreValues(codegen, tuple_ptr, values, pool);
 
+      std::vector<llvm::Value *> insert_args = {inserter, values_buf, values_size};
       // Complete the insertion
-      codegen.Call(InserterProxy::Insert, {inserter});
+      codegen.Call(InserterProxy::Insert, insert_args);
     }
   }
 }
@@ -103,14 +114,25 @@ void InsertTranslator::Consume(ConsumerContext &, RowBatch::Row &row) const {
   // Generate/Materialize tuple data from row and attribute information
   std::vector<codegen::Value> values;
   auto &ais = insert_plan_.GetAttributeInfos();
+  auto *values_buf = codegen.AllocateBuffer(
+      ValueProxy::GetType(codegen), static_cast<uint32_t>(ais.size()),
+      "values");
+  values_buf =
+      codegen->CreatePointerCast(values_buf, codegen.CharPtrType());
+  llvm::Value *values_size = codegen.Const32((int32_t)ais.size());
+  size_t i = 0;
   for (const auto *ai : ais) {
     codegen::Value v = row.DeriveValue(codegen, ai);
     values.push_back(v);
+    peloton::codegen::BufferingConsumer::AddToTupleBuffer(v, codegen, values_buf, i);
+    i++;
   }
   table_storage_.StoreValues(codegen, tuple_ptr, values, pool);
 
+  std::vector<llvm::Value *> insert_args = {inserter, values_buf, values_size};
+
   // Call Inserter to insert the reserved tuple storage area
-  codegen.Call(InserterProxy::Insert, {inserter});
+  codegen.Call(InserterProxy::Insert, insert_args);
 }
 
 void InsertTranslator::TearDownState() {

src/codegen/operator/update_translator.cpp

@@ -10,12 +10,16 @@
 //
 //===----------------------------------------------------------------------===//
 
+#include "codegen/buffering_consumer.h"
 #include "codegen/lang/if.h"
 #include "codegen/proxy/storage_manager_proxy.h"
 #include "codegen/proxy/target_proxy.h"
 #include "codegen/proxy/updater_proxy.h"
+#include "codegen/proxy/value_proxy.h"
+#include "codegen/proxy/values_runtime_proxy.h"
 #include "codegen/operator/update_translator.h"
 #include "codegen/table_storage.h"
+#include "codegen/type/sql_type.h"
 #include "planner/update_plan.h"
 #include "storage/data_table.h"
 
@@ -92,17 +96,45 @@ void UpdateTranslator::Consume(ConsumerContext &, RowBatch::Row &row) const {
       static_cast<uint32_t>(target_list.size() + direct_map_list.size());
   auto &ais = update_plan_.GetAttributeInfos();
 
+  auto is_primary_key = update_plan_.GetUpdatePrimaryKey();
+  llvm::Value *diff = nullptr, *diff_size = nullptr;
+  if (is_primary_key) {
+    diff_size = codegen.Const32((int32_t)column_num);
+    diff = codegen.AllocateBuffer(
+        ValueProxy::GetType(codegen), column_num,
+        "diff");
+  } else {
+    diff_size = codegen.Const32((int32_t)target_list.size());
+    diff = codegen.AllocateBuffer(
+        ValueProxy::GetType(codegen), static_cast<uint32_t>(target_list.size()),
+        "diff");
+  }
+
+
+  diff =
+      codegen->CreatePointerCast(diff, codegen.CharPtrType());
   // Collect all the column values
   std::vector<codegen::Value> values;
-  for (uint32_t i = 0; i < column_num; i++) {
+  for (size_t i = 0, target_id = 0; i < column_num; i++) {
     codegen::Value val;
     uint32_t target_index = GetTargetIndex(target_list, i);
     if (target_index != INVALID_OID) {
       // Set the value for the update
       const auto &derived_attribute = target_list[target_index].second;
       val = row.DeriveValue(codegen, *derived_attribute.expr);
+      size_t offset = 0;
+      if (is_primary_key) {
+        offset = i;
+      } else {
+        offset = target_id;
+      }
+      peloton::codegen::BufferingConsumer::AddToTupleBuffer(val, codegen, diff, offset);
+      target_id++;
     } else {
       val = row.DeriveValue(codegen, ais[i]);
+      if (is_primary_key) {
+        peloton::codegen::BufferingConsumer::AddToTupleBuffer(val, codegen, diff, i);
+      }
     }
     values.push_back(val);
   }
@@ -112,7 +144,7 @@ void UpdateTranslator::Consume(ConsumerContext &, RowBatch::Row &row) const {
   llvm::Value *tuple_ptr;
   std::vector<llvm::Value *> prep_args = {updater, row.GetTileGroupID(),
                                           row.GetTID(codegen)};
-  if (update_plan_.GetUpdatePrimaryKey() == false) {
+  if (is_primary_key == false) {
     tuple_ptr = codegen.Call(UpdaterProxy::Prepare, prep_args);
   } else {
     tuple_ptr = codegen.Call(UpdaterProxy::PreparePK, prep_args);
@@ -130,8 +162,8 @@ void UpdateTranslator::Consume(ConsumerContext &, RowBatch::Row &row) const {
     table_storage_.StoreValues(codegen, tuple_ptr, values, pool_ptr);
 
     // Finally, update with help from the Updater
-    std::vector<llvm::Value *> update_args = {updater};
-    if (update_plan_.GetUpdatePrimaryKey() == false) {
+    std::vector<llvm::Value *> update_args = {updater, diff, diff_size};
+    if (is_primary_key == false) {
       codegen.Call(UpdaterProxy::Update, update_args);
     } else {
       codegen.Call(UpdaterProxy::UpdatePK, update_args);

src/codegen/updater.cpp

@@ -9,7 +9,7 @@
 // Copyright (c) 2015-17, Carnegie Mellon University Database Group
 //
 //===----------------------------------------------------------------------===//
-
+#include "codegen/buffering_consumer.h"
 #include "codegen/updater.h"
 #include "codegen/transaction_runtime.h"
 #include "common/container_tuple.h"
@@ -23,6 +23,13 @@
 #include "type/abstract_pool.h"
 #include "common/internal_types.h"
 #include "type/value.h"
+#include "threadpool/mono_queue_pool.h"
+#include "logging/log_record.h"
+#include "logging/log_buffer.h"
+#include "logging/wal_logger.h"
+#include "threadpool/logger_queue_pool.h"
+#include "../include/type/value.h"
+#include "../include/codegen/value.h"
 
 namespace peloton {
 namespace codegen {
@@ -126,7 +133,7 @@ peloton::type::AbstractPool *Updater::GetPool() {
   return tile_->GetPool();
 }
 
-void Updater::Update() {
+void Updater::Update(char *diff_array, uint32_t diff_size) {
   PELOTON_ASSERT(table_ != nullptr && executor_context_ != nullptr);
   LOG_TRACE("Updating tuple <%u, %u> from table '%s' (db ID: %u, table ID: %u)",
             old_location_.block, old_location_.offset,
@@ -138,7 +145,7 @@ void Updater::Update() {
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
   // Either update in-place
   if (is_owner_ == true) {
-    txn_manager.PerformUpdate(txn, old_location_);
+    txn_manager.PerformUpdate(txn, old_location_, diff_array, diff_size, target_list_);
     // we do not need to add any item pointer to statement-level write set
     // here, because we do not generate any new version
     executor_context_->num_processed++;
@@ -157,12 +164,13 @@ void Updater::Update() {
                                        old_location_.offset);
     return;
   }
-  txn_manager.PerformUpdate(txn, old_location_, new_location_);
-  AddToStatementWriteSet(new_location_);
+  txn_manager.PerformUpdate(txn, old_location_, new_location_, diff_array, diff_size, target_list_);
+
   executor_context_->num_processed++;
+  AddToStatementWriteSet(new_location_);
 }
 
-void Updater::UpdatePK() {
+void Updater::UpdatePK(char *diff_array, uint32_t diff_size) {
   PELOTON_ASSERT(table_ != nullptr && executor_context_ != nullptr);
   LOG_TRACE("Updating tuple <%u, %u> from table '%s' (db ID: %u, table ID: %u)",
             old_location_.block, old_location_.offset,
@@ -181,7 +189,7 @@ void Updater::UpdatePK() {
     txn_manager.SetTransactionResult(txn, ResultType::FAILURE);
     return;
   }
-  txn_manager.PerformInsert(txn, new_location_, index_entry_ptr);
+  txn_manager.PerformInsert(txn, new_location_, index_entry_ptr, diff_array, diff_size);
   AddToStatementWriteSet(new_location_);
   executor_context_->num_processed++;
 }

src/common/init.cpp

@@ -26,6 +26,7 @@
 #include "tuning/index_tuner.h"
 #include "tuning/layout_tuner.h"
 
+
 namespace peloton {
 
 ThreadPool thread_pool;
@@ -90,6 +91,21 @@ void PelotonInit::Initialize() {
 
   // Initialize the Statement Cache Manager
   StatementCacheManager::Init();
+
+  bool enable_logging = settings::SettingsManager::GetBool(settings::SettingId::enable_logging);
+  if(enable_logging){
+    if(!logging::LogManager::GetInstance().init()){
+      LOG_ERROR("LogManager Initialization failed");
+    }
+  }
+
+  bool enable_recovery = settings::SettingsManager::GetBool(settings::SettingId::enable_recovery);
+  if(enable_recovery){
+    logging::LogManager::GetInstance().DoRecovery();
+  }
+
+  threadpool::LoggerQueuePool::GetInstance().Startup();
+
 }
 
 void PelotonInit::Shutdown() {