Merge branch 'master' into ubuntu-cmake

Author: linmagit

cmake/External/capnproto.cmake

@@ -49,14 +49,6 @@ if (NOT __CAPNP_INCLUDED) # guard against multiple includes
     set(CAPNPC_CXX_EXECUTABLE ${capnp_INSTALL}/bin/capnpc-c++)
     set(CAPNP_INCLUDE_DIRS ${capnp_INSTALL}/include)
     set(CAPNP_INCLUDE_DIRECTORY ${capnp_INSTALL}/include)
-    set(CAPNP_LIBRARIES
-      ${capnp_INSTALL}/lib/libcapnp-rpc.a
-      ${capnp_INSTALL}/lib/libcapnp.a
-      ${capnp_INSTALL}/lib/libkj-async.a
-      ${capnp_INSTALL}/lib/libkj.a
-      ${CMAKE_THREAD_LIBS_INIT}
-      )
-    set(CAPNP_LIBRARY_DIRS ${capnp_INSTALL}/lib)
     set(CAPNP_EXTERNAL TRUE)
 
     list(APPEND external_project_dependencies capnp)

src/brain/query_logger.cpp

@@ -0,0 +1,34 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// query_logger.cpp
+//
+// Identification: src/brain/query_logger.cpp
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#include "brain/query_logger.h"
+#include "catalog/query_history_catalog.h"
+#include "concurrency/transaction_context.h"
+#include "concurrency/transaction_manager_factory.h"
+#include "parser/pg_query.h"
+
+namespace peloton {
+namespace brain {
+
+void QueryLogger::LogQuery(std::string query_string, uint64_t timestamp) {
+  auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
+  auto txn = txn_manager.BeginTransaction();
+  std::string fingerprint = pg_query_fingerprint(query_string.c_str()).hexdigest;
+
+  catalog::QueryHistoryCatalog::GetInstance()->InsertQueryHistory(
+      query_string, fingerprint, timestamp, nullptr, txn);
+
+  txn_manager.CommitTransaction(txn);
+}
+
+}  // namespace brain
+}  // namespace peloton

src/catalog/catalog.cpp

@@ -19,6 +19,7 @@
 #include "catalog/index_metrics_catalog.h"
 #include "catalog/language_catalog.h"
 #include "catalog/proc_catalog.h"
+#include "catalog/query_history_catalog.h"
 #include "catalog/query_metrics_catalog.h"
 #include "catalog/settings_catalog.h"
 #include "catalog/table_catalog.h"
@@ -30,6 +31,7 @@
 #include "function/old_engine_string_functions.h"
 #include "function/timestamp_functions.h"
 #include "index/index_factory.h"
+#include "settings/settings_manager.h"
 #include "storage/storage_manager.h"
 #include "storage/table_factory.h"
 #include "type/ephemeral_pool.h"
@@ -146,11 +148,15 @@ void Catalog::Bootstrap() {
   DatabaseMetricsCatalog::GetInstance(txn);
   TableMetricsCatalog::GetInstance(txn);
   IndexMetricsCatalog::GetInstance(txn);
-  QueryMetricsCatalog::GetInstance(txn);
+  QueryMetricsCatalog::GetInstance(txn);  
   SettingsCatalog::GetInstance(txn);
   TriggerCatalog::GetInstance(txn);
   LanguageCatalog::GetInstance(txn);
   ProcCatalog::GetInstance(txn);
+  
+  if (settings::SettingsManager::GetBool(settings::SettingId::brain)) {
+    QueryHistoryCatalog::GetInstance(txn);
+  }
 
   txn_manager.CommitTransaction(txn);
 

src/catalog/query_history_catalog.cpp

@@ -0,0 +1,61 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// query_history_catalog.cpp
+//
+// Identification: src/catalog/query_history_catalog.cpp
+//
+// Copyright (c) 2015-18, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#include "catalog/query_history_catalog.h"
+
+#include "catalog/catalog.h"
+#include "executor/logical_tile.h"
+#include "parser/pg_query.h"
+#include "storage/data_table.h"
+#include "type/value_factory.h"
+
+namespace peloton {
+namespace catalog {
+
+QueryHistoryCatalog *QueryHistoryCatalog::GetInstance(
+    concurrency::TransactionContext *txn) {
+  static QueryHistoryCatalog query_history_catalog{txn};
+  return &query_history_catalog;
+}
+
+QueryHistoryCatalog::QueryHistoryCatalog(concurrency::TransactionContext *txn)
+    : AbstractCatalog("CREATE TABLE " CATALOG_DATABASE_NAME
+                      "." QUERY_HISTORY_CATALOG_NAME
+                      " ("
+                      "query_string   VARCHAR NOT NULL, "
+                      "fingerprint    VARCHAR NOT NULL, "
+                      "timestamp      TIMESTAMP NOT NULL);",
+                      txn) {}
+
+QueryHistoryCatalog::~QueryHistoryCatalog() {}
+
+bool QueryHistoryCatalog::InsertQueryHistory(const std::string &query_string, 
+                                  std::string &fingerprint, uint64_t timestamp,
+                                  type::AbstractPool *pool,
+                                  concurrency::TransactionContext *txn) {
+  std::unique_ptr<storage::Tuple> tuple(
+      new storage::Tuple(catalog_table_->GetSchema(), true));
+
+  auto val0 = type::ValueFactory::GetVarcharValue(query_string, pool);
+  auto val1 = type::ValueFactory::GetVarcharValue(fingerprint, pool);
+  auto val2 = type::ValueFactory::GetTimestampValue(timestamp);
+
+  tuple->SetValue(ColumnId::QUERY_STRING, val0, pool);
+  tuple->SetValue(ColumnId::FINGERPRINT, val1, pool);
+  tuple->SetValue(ColumnId::TIMESTAMP, val2, pool);
+
+  // Insert the tuple
+  return InsertTuple(std::move(tuple), txn);
+}
+
+}  // namespace catalog
+}  // namespace peloton

src/common/init.cpp

@@ -42,6 +42,11 @@ void PelotonInit::Initialize() {
   // start worker pool
   threadpool::MonoQueuePool::GetInstance().Startup();
 
+  // start brain thread pool
+  if (settings::SettingsManager::GetBool(settings::SettingId::brain)) {
+    threadpool::MonoQueuePool::GetBrainInstance().Startup();
+  }
+
   int parallelism = (std::thread::hardware_concurrency() + 3) / 4;
   storage::DataTable::SetActiveTileGroupCount(parallelism);
   storage::DataTable::SetActiveIndirectionArrayCount(parallelism);
@@ -107,6 +112,11 @@ void PelotonInit::Shutdown() {
   // stop worker pool
   threadpool::MonoQueuePool::GetInstance().Shutdown();
 
+  // stop brain thread pool
+  if (settings::SettingsManager::GetBool(settings::SettingId::brain)) {
+    threadpool::MonoQueuePool::GetBrainInstance().Shutdown();
+  }
+
   thread_pool.Shutdown();
 
   // shutdown protocol buf library

src/concurrency/transaction_manager.cpp

@@ -14,6 +14,7 @@
 
 #include "catalog/manager.h"
 #include "concurrency/transaction_context.h"
+#include "function/date_functions.h"
 #include "gc/gc_manager_factory.h"
 #include "logging/log_manager.h"
 #include "settings/settings_manager.h"
@@ -72,6 +73,8 @@ TransactionContext *TransactionManager::BeginTransaction(
         .StartTimer();
   }
 
+  txn->SetTimestamp(function::DateFunctions::Now());
+
   return txn;
 }
 

src/gc/transaction_level_gc_manager.cpp

@@ -12,14 +12,18 @@
 
 #include "gc/transaction_level_gc_manager.h"
 
+#include "brain/query_logger.h"
 #include "catalog/manager.h"
 #include "common/container_tuple.h"
 #include "concurrency/epoch_manager_factory.h"
 #include "concurrency/transaction_manager_factory.h"
+#include "settings/settings_manager.h"
 #include "storage/database.h"
 #include "storage/tile_group.h"
 #include "storage/tuple.h"
 #include "storage/storage_manager.h"
+#include "threadpool/mono_queue_pool.h"
+
 
 namespace peloton {
 namespace gc {
@@ -127,6 +131,20 @@ int TransactionLevelGCManager::Unlink(const int &thread_id,
       break;
     }
 
+    // Log the query into query_history_catalog
+    if (settings::SettingsManager::GetBool(settings::SettingId::brain)) {
+      std::vector<std::string> query_strings = txn_ctx->GetQueryStrings();
+      if(query_strings.size() != 0) {
+        uint64_t timestamp = txn_ctx->GetTimestamp();
+        auto &pool = threadpool::MonoQueuePool::GetBrainInstance();
+        for(auto query_string: query_strings) {
+          pool.SubmitTask([this, query_string, timestamp] {
+            brain::QueryLogger::LogQuery(query_string, timestamp);
+          });        
+        }
+      }
+    }
+
     // Deallocate the Transaction Context of transactions that don't involve
     // any garbage collection
     if (txn_ctx->GetIsolationLevel() == IsolationLevelType::READ_ONLY || \

src/include/brain/query_logger.h

@@ -0,0 +1,37 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// query_logger.h
+//
+// Identification: src/include/brain/query_logger.h
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include <string>
+
+namespace peloton {
+namespace brain {
+
+//===--------------------------------------------------------------------===//
+// QueryLogger
+//===--------------------------------------------------------------------===//
+
+class QueryLogger {
+ public:
+  QueryLogger() = default;
+
+  /*
+   * @brief This function logs the query into query_history_catalog
+   * @param the sql string corresponding to the query
+   * @param timestamp of the transaction that executed the query
+   */
+  static void LogQuery(std::string query_string, uint64_t timestamp);
+};
+
+}  // namespace brain
+}  // namespace peloton

src/include/catalog/query_history_catalog.h

@@ -0,0 +1,60 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// query_history_catalog.h
+//
+// Identification: src/include/catalog/query_history_catalog.h
+//
+// Copyright (c) 2015-18, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+//===----------------------------------------------------------------------===//
+// pg_query
+//
+// Schema: (column offset: column_name)
+// 0: query_string
+// 1: fingerprint
+// 2: timestamp
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "catalog/abstract_catalog.h"
+
+#define QUERY_HISTORY_CATALOG_NAME "pg_query_history"
+
+namespace peloton {
+namespace catalog {
+
+class QueryHistoryCatalog : public AbstractCatalog {
+ public:
+  ~QueryHistoryCatalog();
+
+  // Global Singleton
+  static QueryHistoryCatalog *GetInstance(
+      concurrency::TransactionContext *txn = nullptr);
+
+  //===--------------------------------------------------------------------===//
+  // write Related API
+  //===--------------------------------------------------------------------===//
+  bool InsertQueryHistory(const std::string &query_string,
+                          std::string &fingerprint, uint64_t timestamp,
+                          type::AbstractPool *pool,
+                          concurrency::TransactionContext *txn);
+
+  enum ColumnId {
+    QUERY_STRING = 0,
+    FINGERPRINT = 1,
+    TIMESTAMP = 2,
+  };
+
+ private:
+  QueryHistoryCatalog(concurrency::TransactionContext *txn);
+
+};
+
+}  // namespace catalog
+}  // namespace peloton

src/include/concurrency/transaction_context.h

@@ -73,9 +73,20 @@ class TransactionContext : public Printable {
 
   inline eid_t GetEpochId() const { return epoch_id_; }
 
+  inline uint64_t GetTimestamp() const { return timestamp_; }
+
+  inline const std::vector<std::string>& GetQueryStrings() const {
+                                                      return query_strings_; }
+
   inline void SetCommitId(const cid_t commit_id) { commit_id_ = commit_id; }
 
   inline void SetEpochId(const eid_t epoch_id) { epoch_id_ = epoch_id; }
+  
+  inline void SetTimestamp(const uint64_t timestamp) { timestamp_ = timestamp; }
+
+  inline void AddQueryString(const char* query_string) {
+    query_strings_.push_back(std::string(query_string));
+  }
 
   void RecordCreate(oid_t database_oid, oid_t table_oid, oid_t index_oid) {
     rw_object_set_.emplace_back(database_oid, table_oid, index_oid,
@@ -174,6 +185,13 @@ class TransactionContext : public Printable {
   // GC manager uses this id to check whether a version is still visible.
   eid_t epoch_id_;
 
+  // vector of strings to log at the end of the transaction
+  // populated only if the brain is running
+  std::vector<std::string> query_strings_;
+
+  // timestamp when the transaction began
+  uint64_t timestamp_;
+
   ReadWriteSet rw_set_;
   CreateDropSet rw_object_set_;