Merge branch 'master' into unique

Author: apavlo

script/installation/packages.sh

@@ -48,22 +48,13 @@ if [ "$DISTRO" = "UBUNTU" ]; then
         fi
         sudo apt-key adv --keyserver keyserver.ubuntu.com --recv-keys 15CF4D18AF4F7421
         sudo apt-get update -qq
+        CMAKE_NAME="cmake3"
+        FORCE_Y="--force-yes"
+    else
+        CMAKE_NAME="cmake"
+        FORCE_Y=""
     fi
 
-    # Fix for cmake3 name change in Ubuntu 15.x and 16.x plus --force-yes deprecation
-    CMAKE_NAME="cmake3"
-    FORCE_Y="--force-yes"
-    MAJOR_VER=$(echo "$DISTRO_VER" | cut -d '.' -f 1)
-    for version in "15" "16"
-    do
-       if [ "$MAJOR_VER" = "$version" ]
-       then
-          FORCE_Y=""
-          CMAKE_NAME="cmake"
-          break
-       fi
-    done
-
     sudo apt-get -qq $FORCE_Y --ignore-missing -y install \
         git \
         g++ \

src/executor/create_executor.cpp

@@ -99,25 +99,26 @@ bool CreateExecutor::CreateDatabase(const planner::CreatePlan &node) {
 }
 
 bool CreateExecutor::CreateTable(const planner::CreatePlan &node) {
-  auto txn = context_->GetTransaction();
+  auto current_txn = context_->GetTransaction();
   std::string table_name = node.GetTableName();
   auto database_name = node.GetDatabaseName();
   std::unique_ptr<catalog::Schema> schema(node.GetSchema());
 
   ResultType result = catalog::Catalog::GetInstance()->CreateTable(
-      database_name, table_name, std::move(schema), txn);
-  txn->SetResult(result);
+      database_name, table_name, std::move(schema), current_txn);
+  current_txn->SetResult(result);
 
-  if (txn->GetResult() == ResultType::SUCCESS) {
+  if (current_txn->GetResult() == ResultType::SUCCESS) {
     LOG_TRACE("Creating table succeeded!");
 
     // Add the foreign key constraint (or other multi-column constraints)
     if (node.GetForeignKeys().empty() == false) {
       auto catalog = catalog::Catalog::GetInstance();
-      auto db = catalog->GetDatabaseWithName(database_name, txn);
+      auto db = catalog->GetDatabaseWithName(database_name, current_txn);
 
       auto source_table = db->GetTableWithName(table_name);
       int count = 1;
+
       for (auto fk : node.GetForeignKeys()) {
         auto sink_table = db->GetTableWithName(fk.sink_table_name);
 
@@ -139,7 +140,7 @@ bool CreateExecutor::CreateTable(const planner::CreatePlan &node) {
         // Sink Column Offsets
         std::vector<oid_t> sink_col_ids;
         for (auto col_name : fk.foreign_key_sinks) {
-          oid_t col_id = source_table->GetSchema()->GetColumnID(col_name);
+          oid_t col_id = sink_table->GetSchema()->GetColumnID(col_name);
           if (col_id == INVALID_OID) {
             std::string error = StringUtil::Format(
                 "Invalid sink column name '%s.%s' for foreign key '%s'",
@@ -152,43 +153,51 @@ bool CreateExecutor::CreateTable(const planner::CreatePlan &node) {
         PL_ASSERT(sink_col_ids.size() == fk.foreign_key_sinks.size());
 
         // Create the catalog object and shove it into the table
-        auto catalog_fk = new catalog::ForeignKey(
+        auto catalog_fk = new catalog::ForeignKey(INVALID_OID,
             sink_table->GetOid(), sink_col_ids, source_col_ids,
             fk.upd_action, fk.del_action, fk.constraint_name);
         source_table->AddForeignKey(catalog_fk);
 
         // Register FK with the sink table for delete/update actions
-        sink_table->RegisterForeignKeySource(table_name);
+        catalog_fk = new catalog::ForeignKey(source_table->GetOid(),
+                                              INVALID_OID,
+                                              sink_col_ids,
+                                              source_col_ids,
+                                              fk.upd_action,
+                                              fk.del_action,
+                                              fk.constraint_name);
+        sink_table->RegisterForeignKeySource(catalog_fk);
 
         // Add a non-unique index on the source table if needed
-        if (catalog_fk->GetUpdateAction() != FKConstrActionType::NOACTION ||
-            catalog_fk->GetDeleteAction() != FKConstrActionType::NOACTION) {
-          std::vector<std::string> source_col_names =
-              fk.foreign_key_sources;
-          std::string index_name =
-              source_table->GetName() + "_FK_" + std::to_string(count);
-          catalog->CreateIndex(database_name, source_table->GetName(),
-                               source_col_ids, index_name, false,
-                               IndexType::BWTREE, txn);
-          count++;
+        std::vector<std::string> source_col_names =
+            fk.foreign_key_sources;
+        std::string index_name =
+            source_table->GetName() + "_FK_" + sink_table->GetName() + "_"
+            + std::to_string(count);
+        catalog->CreateIndex(database_name, source_table->GetName(),
+                              source_col_ids, index_name, false,
+                              IndexType::BWTREE, current_txn);
+        count++;
 
 #ifdef LOG_DEBUG_ENABLED
-          LOG_DEBUG("Added a FOREIGN index on in %s.", table_name.c_str());
-          LOG_DEBUG("Foreign key column names: ");
-          for (auto c : source_col_names) {
-            LOG_DEBUG("FK col name: %s", c.c_str());
-          }
-#endif
+        LOG_DEBUG("Added a FOREIGN index on in %s.\n", table_name.c_str());
+        LOG_DEBUG("Foreign key column names: \n");
+        for (auto c : source_col_names) {
+          LOG_DEBUG("FK col name: %s\n", c.c_str());
+        }
+        for (auto c : fk.foreign_key_sinks) {
+          LOG_DEBUG("FK sink col name: %s\n", c.c_str());
         }
+#endif
       }
     }
-  } else if (txn->GetResult() == ResultType::FAILURE) {
+  } else if (current_txn->GetResult() == ResultType::FAILURE) {
     LOG_TRACE("Creating table failed!");
-    return (false);
   } else {
     LOG_TRACE("Result is: %s",
-              ResultTypeToString(txn->GetResult()).c_str());
+              ResultTypeToString(current_txn->GetResult()).c_str());
   }
+
   return (true);
 }
 

src/executor/delete_executor.cpp

@@ -133,6 +133,27 @@ bool DeleteExecutor::DExecute() {
       physical_tuple_id = old_location.offset;
     }
 
+    ContainerTuple<storage::TileGroup> old_tuple(tile_group, physical_tuple_id);
+    storage::Tuple prev_tuple(target_table_->GetSchema(), true);
+
+    // Get a copy of the old tuple
+    for (oid_t column_itr = 0; column_itr < target_table_schema->GetColumnCount(); column_itr++) {
+      type::Value val = (old_tuple.GetValue(column_itr));
+      prev_tuple.SetValue(column_itr, val, executor_context_->GetPool());
+    }
+
+    // Check the foreign key source table
+    if (target_table_->CheckForeignKeySrcAndCascade(&prev_tuple,
+                                                    nullptr,
+                                                    current_txn,
+                                                    executor_context_,
+                                                    false) == false)
+    {
+      transaction_manager.SetTransactionResult(current_txn,
+                                              peloton::ResultType::FAILURE);
+      return false;
+    }
+
     bool is_owner = transaction_manager.IsOwner(current_txn, tile_group_header,
                                                 physical_tuple_id);
 

src/executor/update_executor.cpp

@@ -22,6 +22,8 @@
 #include "storage/data_table.h"
 #include "storage/tile_group_header.h"
 #include "storage/tile.h"
+#include "storage/storage_manager.h"
+#include "catalog/foreign_key.h"
 
 namespace peloton {
 namespace executor {
@@ -112,6 +114,27 @@ bool UpdateExecutor::PerformUpdatePrimaryKey(
     return false;
   }
 
+  // Check the source table of any foreign key constraint
+  if (target_table_->GetForeignKeySrcCount() > 0) {
+    storage::Tuple prev_tuple(target_table_schema, true);
+    // Get a copy of the old tuple
+    for (oid_t column_itr = 0; column_itr < target_table_schema->GetColumnCount(); column_itr++) {
+      type::Value val = (old_tuple.GetValue(column_itr));
+      prev_tuple.SetValue(column_itr, val, executor_context_->GetPool());
+    }
+
+    if (target_table_->CheckForeignKeySrcAndCascade(&prev_tuple,
+                                                    &new_tuple,
+                                                    current_txn,
+                                                    executor_context_,
+                                                    true) == false)
+    {
+      transaction_manager.SetTransactionResult(current_txn,
+                                              peloton::ResultType::FAILURE);
+      return false;
+    }
+  }
+
   transaction_manager.PerformInsert(current_txn, location, index_entry_ptr);
 
   return true;

src/include/catalog/foreign_key.h

@@ -28,20 +28,23 @@ namespace catalog {
 // Stores info about foreign key constraints, like the sink table id etc.
 class ForeignKey {
  public:
-  ForeignKey(oid_t sink_table_id,
+  ForeignKey(oid_t source_table_id,
+             oid_t sink_table_id,
              std::vector<oid_t> sink_col_ids,
              std::vector<oid_t> source_col_ids,
              FKConstrActionType update_action,
              FKConstrActionType delete_action,
              std::string constraint_name)
 
-      : sink_table_id(sink_table_id),
+      : source_table_id(source_table_id),
+        sink_table_id(sink_table_id),
         sink_col_ids(sink_col_ids),
         source_col_ids(source_col_ids),
         update_action(update_action),
         delete_action(delete_action),
         fk_name(constraint_name) {}
 
+  oid_t GetSourceTableOid() const { return source_table_id; }
   oid_t GetSinkTableOid() const { return sink_table_id; }
 
   std::vector<oid_t> GetSinkColumnIds() const { return sink_col_ids; }
@@ -52,6 +55,7 @@ class ForeignKey {
   std::string &GetConstraintName() { return fk_name; }
 
  private:
+  oid_t source_table_id = INVALID_OID;
   oid_t sink_table_id = INVALID_OID;
 
   // Columns in the reference table (sink)

src/include/common/exception.h

@@ -56,7 +56,8 @@ enum class ExceptionType {
   CONNECTION = 21,        // connection related
   SYNTAX = 22,            // syntax related
   SETTINGS = 23,          // settings related
-  BINDER = 24             // settings related
+  BINDER = 24,             // settings related
+  NETWORK = 25
 };
 
 class Exception : public std::runtime_error {
@@ -427,6 +428,13 @@ class ConnectionException : public Exception {
       : Exception(ExceptionType::CONNECTION, msg) {}
 };
 
+class NetworkProcessException : public Exception {
+  NetworkProcessException() = delete;
+
+  public:
+  NetworkProcessException(std::string msg) : Exception(ExceptionType::NETWORK, msg) {}
+};
+
 class SettingsException : public Exception {
   SettingsException() = delete;
 

src/include/common/thread_pool.h

@@ -10,30 +10,29 @@
 //
 //===----------------------------------------------------------------------===//
 
-
 #pragma once
 
-#include <vector>
 #include <thread>
+#include <vector>
 
 #include <boost/asio/io_service.hpp>
-#include <boost/thread/thread.hpp>
 #include <boost/bind.hpp>
 #include <boost/function.hpp>
+#include <boost/thread/thread.hpp>
 
 #include "common/macros.h"
 
 namespace peloton {
 // a wrapper for boost worker thread pool.
 class ThreadPool {
  public:
-  ThreadPool() : pool_size_(0), 
-                 dedicated_thread_count_(0),
-                 work_(io_service_) { }
+  ThreadPool()
+      : pool_size_(0), dedicated_thread_count_(0), work_(io_service_) {}
 
-  ~ThreadPool() { }
+  ~ThreadPool() {}
 
-  void Initialize(const size_t &pool_size, const size_t &dedicated_thread_count) {
+  void Initialize(const size_t &pool_size,
+                  const size_t &dedicated_thread_count) {
     current_thread_count_ = ATOMIC_VAR_INIT(0);
     pool_size_ = pool_size;
     // PL_ASSERT(pool_size_ != 0);
@@ -73,7 +72,8 @@ class ThreadPool {
     size_t thread_id =
         current_thread_count_.fetch_add(1, std::memory_order_relaxed);
     // assign task to dedicated thread.
-    dedicated_threads_[thread_id].reset(new std::thread(std::thread(func, params...)));
+    dedicated_threads_[thread_id].reset(
+        new std::thread(std::thread(func, params...)));
   }
 
  private:

src/include/network/connection_dispatcher_task.h

@@ -0,0 +1,69 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// connection_dispatcher_task.h
+//
+// Identification: src/include/network/connection_dispatcher_task.h
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "notifiable_task.h"
+#include "network_state.h"
+#include "concurrency/epoch_manager_factory.h"
+#include "connection_handler_task.h"
+
+namespace peloton {
+namespace network {
+
+/**
+ * @brief A ConnectionDispatcherTask on the main server thread and dispatches
+ * incoming connections to handler threads.
+ *
+ * On construction, the dispatcher also spawns a number of handlers running on
+ * their own threads. The dispatcher is
+ * then responsible for maintain, and when shutting down, shutting down the
+ * spawned handlers also.
+ */
+class ConnectionDispatcherTask : public NotifiableTask {
+ public:
+  /**
+   * Creates a new ConnectionDispatcherTask, spawning the specified number of
+   * handlers, each running on their own threads.
+   *
+   * @param num_handlers The number of handler tasks to spawn.
+   * @param listen_fd The server socket fd to listen on.
+   */
+  ConnectionDispatcherTask(int num_handlers, int listen_fd);
+
+  /**
+   * @brief Dispatches the client connection at fd to a handler.
+   * Currently, the dispatch uses round-robin, and thread communication is
+   * achieved
+   * through channels. The dispatch writes a symbol to the fd that the handler
+   * is configured
+   * to receive updates on.
+   *
+   * @param fd the socket fd of the client connection being dispatched
+   * @param flags Unused. This is here to conform to libevent callback function
+   * signature.
+   */
+  void DispatchConnection(int fd, short flags);
+
+  /**
+   * Breaks the dispatcher and managed handlers from their event loops.
+   */
+  void ExitLoop() override;
+
+ private:
+  std::vector<std::shared_ptr<ConnectionHandlerTask>> handlers_;
+  // TODO: have a smarter dispatch scheduler, we currently use round-robin
+  std::atomic<int> next_handler_;
+};
+
+}  // namespace network
+}  // namespace peloton