[15721] Peloton Self-Driving Infrastructure

src/brain/catalog_sync_brain_job.cpp

@@ -0,0 +1,102 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// catalog_sync_brain_job.cpp
+//
+// Identification: src/brain/catalog_sync_brain_job.cpp
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+#include "brain/catalog_sync_brain_job.h"
+
+namespace peloton {
+namespace brain {
+void CatalogSyncBrainJob::OnJobInvocation(BrainEnvironment *env) {
+  auto &manager = concurrency::TransactionManagerFactory::GetInstance();
+  for (auto *catalog : catalog::Catalog::GetInstance()->AvailableCatalogs())
+    SyncCatalog(catalog, env, manager);
+}
+
+time_t CatalogSyncBrainJob::TimeFromString(const char *str) {
+  struct tm tm;
+  PELOTON_MEMSET(&tm, 0, sizeof(struct tm));
+  strptime(str, "%Y-%m-%d %H:%M:%S", &tm);
+  return mktime(&tm);
+}
+
+std::string CatalogSyncBrainJob::FetchCatalogQuery(catalog::AbstractCatalog *catalog) {
+  // We need to special cast these two tables because we cannot put a reasonable
+  // primary key on them without sequences
+  if (catalog->GetName() == QUERY_HISTORY_CATALOG_NAME)
+    return "SELECT * FROM pg_catalog." + std::string(QUERY_HISTORY_CATALOG_NAME)
+        + " WHERE timestamp > " + std::to_string(last_history_timestamp_);
+  else
+    return "SELECT * FROM pg_catalog." + catalog->GetName();
+}
+
+void CatalogSyncBrainJob::UpdateTimestamp(catalog::AbstractCatalog *catalog,
+                                          pqxx::field field) {
+  if (catalog->GetName() == QUERY_HISTORY_CATALOG_NAME
+      && field.name() == std::string("timestamp"))
+    last_history_timestamp_ =
+        std::max(last_history_timestamp_, field.as<int64_t>());
+}
+
+void CatalogSyncBrainJob::SyncCatalog(catalog::AbstractCatalog *catalog,
+                                      BrainEnvironment *env,
+                                      concurrency::TransactionManager &manager) {
+  pqxx::result r = env->ExecuteQuery(FetchCatalogQuery(catalog));
+  for (auto row : r) {
+    concurrency::TransactionContext *txn =
+        manager.BeginTransaction(IsolationLevelType::REPEATABLE_READS);
+    catalog::Schema *catalog_schema = catalog->GetDataTable()->GetSchema();
+    std::unique_ptr<storage::Tuple> tuple(
+        new storage::Tuple(catalog_schema, true));
+    for (auto field : row) {
+      oid_t column_id = catalog_schema->GetColumnID(field.name());
+      tuple->SetValue(column_id, PqxxFieldToPelotonValue(field));
+      UpdateTimestamp(catalog, field);
+    }
+    catalog->InsertTuple(std::move(tuple), txn);
+    // We know this will always succeed on the brain side
+    manager.CommitTransaction(txn);
+  }
+}
+
+type::Value CatalogSyncBrainJob::PqxxFieldToPelotonValue(pqxx::field &f) {
+  type::TypeId type = PostgresValueTypeToPelotonValueType(
+      static_cast<PostgresValueType>(f.type()));
+  if (f.is_null()) {
+    return type == peloton::type::TypeId::VARCHAR
+           ? type::ValueFactory::GetVarcharValue("")
+           : type::ValueFactory::GetNullValueByType(type);
+  }
+  switch (type) {
+    case type::TypeId::BOOLEAN:
+      return type::ValueFactory::GetBooleanValue(f.as<bool>());
+    case type::TypeId::TINYINT:
+      return type::ValueFactory::GetTinyIntValue(static_cast<int8_t>(f.as<
+          int32_t>()));
+    case type::TypeId::SMALLINT:
+      return type::ValueFactory::GetSmallIntValue(static_cast<int16_t>(f.as<
+          int32_t>()));
+    case type::TypeId::INTEGER:
+      return type::ValueFactory::GetIntegerValue(f.as<int32_t>());
+    case type::TypeId::BIGINT:
+      return type::ValueFactory::GetBigIntValue(f.as<int64_t>());
+    case type::TypeId::TIMESTAMP:
+      return type::ValueFactory::GetTimestampValue(TimeFromString(f.c_str()));
+    case type::TypeId::DECIMAL:
+      return type::ValueFactory::GetDecimalValue(f.as<double>());
+    case type::TypeId::VARCHAR:return type::ValueFactory::GetVarcharValue(f.c_str());
+    default:
+      throw ConversionException(StringUtil::Format(
+          "No corresponding c++ type for postgres type %d",
+          static_cast<int>(type)));
+  }
+}
+
+} // namespace brain
+} // nanespace peloton

src/brain/query_logger.cpp

@@ -40,7 +40,7 @@ void QueryLogger::LogQuery(std::string query_string, uint64_t timestamp) {
   Fingerprint fingerprint{query_string};
 
   // Log query + fingerprint
-  auto &query_history_catalog = catalog::QueryHistoryCatalog::GetInstance();
+  auto &query_history_catalog = catalog::QueryHistoryCatalog::GetInstance(txn);
   query_history_catalog.InsertQueryHistory(
       query_string, fingerprint.GetFingerprint(), timestamp, nullptr, txn);
 

src/catalog/catalog.cpp

@@ -20,19 +20,20 @@
 #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/system_catalogs.h"
 #include "catalog/table_catalog.h"
 #include "catalog/table_metrics_catalog.h"
 #include "catalog/trigger_catalog.h"
+#include "catalog/tuple_access_metrics_catalog.h"
 #include "concurrency/transaction_manager_factory.h"
 #include "function/date_functions.h"
 #include "function/decimal_functions.h"
 #include "function/old_engine_string_functions.h"
 #include "function/timestamp_functions.h"
 #include "index/index_factory.h"
 #include "settings/settings_manager.h"
+#include "storage/database.h"
 #include "storage/storage_manager.h"
 #include "storage/table_factory.h"
 #include "type/ephemeral_pool.h"
@@ -191,11 +192,12 @@ void Catalog::Bootstrap() {
   DatabaseMetricsCatalog::GetInstance(txn);
   SettingsCatalog::GetInstance(txn);
   LanguageCatalog::GetInstance(txn);
+  TupleAccessMetricsCatalog::GetInstance(txn);
 
   // TODO: change pg_proc to per database
   ProcCatalog::GetInstance(txn);
 
-  if (settings::SettingsManager::GetBool(settings::SettingId::brain)) {
+  if (settings::SettingsManager::GetBool(settings::SettingId::brain_data_collection)) {
     QueryHistoryCatalog::GetInstance(txn);
   }
 

src/catalog/column_stats_catalog.cpp

@@ -41,13 +41,9 @@ ColumnStatsCatalog::ColumnStatsCatalog(concurrency::TransactionContext *txn)
                       "most_common_freqs VARCHAR, "
                       "histogram_bounds  VARCHAR, "
                       "column_name       VARCHAR, "
-                      "has_index         BOOLEAN);",
+                      "has_index         BOOLEAN,"
+                      "PRIMARY KEY(database_id, table_id, column_id));",
                       txn) {
-  // unique key: (database_id, table_id, column_id)
-  Catalog::GetInstance()->CreateIndex(
-      CATALOG_DATABASE_NAME, CATALOG_SCHEMA_NAME, COLUMN_STATS_CATALOG_NAME,
-      {0, 1, 2}, COLUMN_STATS_CATALOG_NAME "_skey0", true, IndexType::BWTREE,
-      txn);
   // non-unique key: (database_id, table_id)
   Catalog::GetInstance()->CreateIndex(
       CATALOG_DATABASE_NAME, CATALOG_SCHEMA_NAME, COLUMN_STATS_CATALOG_NAME,

src/catalog/database_metrics_catalog.cpp

@@ -4,7 +4,7 @@
 //
 // database_metrics_catalog.cpp
 //
-// Identification: src/catalog/query_metrics_catalog.cpp
+// Identification: src/catalog/database_metrics_catalog.cpp
 //
 // Copyright (c) 2015-17, Carnegie Mellon University Database Group
 //
@@ -19,6 +19,23 @@
 namespace peloton {
 namespace catalog {
 
+DatabaseMetricsCatalogObject::DatabaseMetricsCatalogObject(
+    executor::LogicalTile *tile, int tupleId)
+    : database_oid_(
+          tile->GetValue(tupleId,
+                         DatabaseMetricsCatalog::ColumnId::DATABASE_OID)
+              .GetAs<oid_t>()),
+      txn_committed_(
+          tile->GetValue(tupleId,
+                         DatabaseMetricsCatalog::ColumnId::TXN_COMMITTED)
+              .GetAs<int64_t>()),
+      txn_aborted_(
+          tile->GetValue(tupleId, DatabaseMetricsCatalog::ColumnId::TXN_ABORTED)
+              .GetAs<int64_t>()),
+      time_stamp_(
+          tile->GetValue(tupleId, DatabaseMetricsCatalog::ColumnId::TIME_STAMP)
+              .GetAs<int64_t>()) {}
+
 DatabaseMetricsCatalog *DatabaseMetricsCatalog::GetInstance(
     concurrency::TransactionContext *txn) {
   static DatabaseMetricsCatalog database_metrics_catalog{txn};
@@ -30,7 +47,7 @@ DatabaseMetricsCatalog::DatabaseMetricsCatalog(
     : AbstractCatalog("CREATE TABLE " CATALOG_DATABASE_NAME
                       "." CATALOG_SCHEMA_NAME "." DATABASE_METRICS_CATALOG_NAME
                       " ("
-                      "database_oid  INT NOT NULL, "
+                      "database_oid  INT NOT NULL PRIMARY KEY, "
                       "txn_committed INT NOT NULL, "
                       "txn_aborted   INT NOT NULL, "
                       "time_stamp    INT NOT NULL);",
@@ -71,5 +88,54 @@ bool DatabaseMetricsCatalog::DeleteDatabaseMetrics(
   return DeleteWithIndexScan(index_offset, values, txn);
 }
 
+bool DatabaseMetricsCatalog::UpdateDatabaseMetrics(
+    oid_t database_oid, oid_t txn_committed, oid_t txn_aborted,
+    oid_t time_stamp, concurrency::TransactionContext *txn) {
+  std::vector<oid_t> update_columns(all_column_ids_);
+  std::vector<type::Value> update_values;
+
+  update_values.push_back(
+      type::ValueFactory::GetIntegerValue(database_oid).Copy());
+  update_values.push_back(
+      type::ValueFactory::GetIntegerValue(txn_committed).Copy());
+  update_values.push_back(
+      type::ValueFactory::GetIntegerValue(txn_aborted).Copy());
+  update_values.push_back(
+      type::ValueFactory::GetIntegerValue(time_stamp).Copy());
+
+  std::vector<type::Value> scan_values;
+  scan_values.push_back(type::ValueFactory::GetIntegerValue(database_oid));
+  oid_t index_offset = IndexId::PRIMARY_KEY;
+
+  return UpdateWithIndexScan(update_columns, update_values, scan_values,
+                             index_offset, txn);
+}
+
+std::shared_ptr<DatabaseMetricsCatalogObject>
+DatabaseMetricsCatalog::GetDatabaseMetricsObject(
+    oid_t database_oid, concurrency::TransactionContext *txn) {
+  if (txn == nullptr) {
+    throw CatalogException("Transaction is invalid!");
+  }
+
+  // set up query
+  std::vector<oid_t> column_ids(all_column_ids_);
+  oid_t index_offset = IndexId::PRIMARY_KEY;
+  std::vector<type::Value> values;
+  values.push_back(type::ValueFactory::GetIntegerValue(database_oid).Copy());
+
+  auto result_tiles =
+      GetResultWithIndexScan(column_ids, index_offset, values, txn);
+
+  if (result_tiles->size() == 1 && (*result_tiles)[0]->GetTupleCount() == 1) {
+    auto database_metric_object =
+        std::make_shared<DatabaseMetricsCatalogObject>(
+            (*result_tiles)[0].get());
+    return database_metric_object;
+  }
+
+  return nullptr;
+}
+
 }  // namespace catalog
 }  // namespace peloton