Merge pull request #16 from sxzh93/alter_executor

add alter executor

Author: Dingshilun

src/catalog/catalog.cpp

@@ -804,9 +804,10 @@ std::shared_ptr<TableCatalogObject> Catalog::GetTableObject(
  * @param txn the transaction Context
  * @return TransactionContext ResultType(SUCCESS or FAILURE)
  */
-ResultType Catalog::AlterTable(oid_t database_oid, oid_t table_oid,
-                               std::unique_ptr<catalog::Schema> new_schema,
-                               concurrency::TransactionContext *txn) {
+ResultType Catalog::AlterTable(
+    UNUSED_ATTRIBUTE oid_t database_oid, UNUSED_ATTRIBUTE oid_t table_oid,
+    UNUSED_ATTRIBUTE std::unique_ptr<catalog::Schema> &new_schema,
+    UNUSED_ATTRIBUTE concurrency::TransactionContext *txn) {
   LOG_TRACE("AlterTable in Catalog");
 
   if (txn == nullptr)

src/executor/alter_executor.cpp

@@ -13,18 +13,18 @@
 #include "executor/alter_executor.h"
 
 #include "catalog/catalog.h"
+#include "catalog/table_catalog.h"
 #include "common/logger.h"
 #include "executor/executor_context.h"
+#include "storage/data_table.h"
 
 namespace peloton {
 namespace executor {
 
 // Constructor for alter table executor
 AlterExecutor::AlterExecutor(const planner::AbstractPlan *node,
                              ExecutorContext *executor_context)
-    : AbstractExecutor(node, executor_context),
-      isAlter_(
-          !reinterpret_cast<const planner::AlterPlan *>(node)->IsRename()) {}
+    : AbstractExecutor(node, executor_context) {}
 
 // Initialize executor
 // Nothing to initialize for now
@@ -38,24 +38,20 @@ bool AlterExecutor::DInit() {
 bool AlterExecutor::DExecute() {
   LOG_TRACE("Executing Alter...");
   bool result = false;
-  if (!isAlter_) {
-    const planner::AlterPlan &node = GetPlanNode<planner::AlterPlan>();
-    auto current_txn = executor_context_->GetTransaction();
-    result = RenameColumn(node, current_txn);
-  } else {
-    const planner::AlterPlan &node = GetPlanNode<planner::AlterPlan>();
-    auto current_txn = executor_context_->GetTransaction();
-    AlterType type = node.GetAlterTableType();
-    switch (type) {
-      case AlterType::ALTER:
-        result = DropColumn(node, current_txn);
-        break;
-      default:
-        throw NotImplementedException(StringUtil::Format(
-            "Alter Type not supported, %s", AlterTypeToString(type).c_str()));
-    }
+  const planner::AlterPlan &node = GetPlanNode<planner::AlterPlan>();
+  auto current_txn = executor_context_->GetTransaction();
+  AlterType type = node.GetAlterTableType();
+  switch (type) {
+    case AlterType::RENAME:
+      result = RenameColumn(node, current_txn);
+      break;
+    case AlterType::ALTER:
+      result = AlterTable(node, current_txn);
+      break;
+    default:
+      throw NotImplementedException(StringUtil::Format(
+          "Alter Type not supported, %s", AlterTypeToString(type).c_str()));
   }
-
   return result;
 }
 
@@ -82,23 +78,109 @@ bool AlterExecutor::RenameColumn(
   return false;
 }
 
-bool AlterExecutor::DropColumn(const peloton::planner::AlterPlan &node,
+bool AlterExecutor::AlterTable(const peloton::planner::AlterPlan &node,
                                peloton::concurrency::TransactionContext *txn) {
   auto database_name = node.GetDatabaseName();
   auto table_name = node.GetTableName();
-  auto drop_columns = node.GetDroppedColumns();
 
-  ResultType result = catalog::Catalog::GetInstance()->DropColumn(
-      database_name, table_name, drop_columns, txn);
-  txn->SetResult(result);
+  auto table_catalog_obj = catalog::Catalog::GetInstance()->GetTableObject(
+      database_name, table_name, txn);
+  oid_t database_oid = table_catalog_obj->GetDatabaseOid();
+  oid_t table_oid = table_catalog_obj->GetTableOid();
+
+  auto old_table = catalog::Catalog::GetInstance()->GetTableWithName(
+      database_name, table_name, txn);
+  auto old_schema = old_table->GetSchema();
+  std::vector<oid_t> column_ids;
+
+  // Step 1: remove drop columns from old schema
+  for (oid_t i = 0; i < old_schema->GetColumnCount(); ++i) {
+    bool is_found = false;
+    for (auto drop_column : node.GetDroppedColumns()) {
+      if (old_schema->GetColumn(i).GetName() == drop_column) {
+        is_found = true;
+      }
+    }
+    if (!is_found) {
+      column_ids.push_back(i);
+    }
+  }
+  // Check if dropped column exists
+  if (column_ids.size() + node.GetDroppedColumns().size() !=
+      old_schema->GetColumnCount()) {
+    LOG_TRACE("Dropped column not exists");
+    txn->SetResult(ResultType::FAILURE);
+    return false;
+  }
+  std::unique_ptr<catalog::Schema> temp_schema(
+      catalog::Schema::CopySchema(old_schema, column_ids));
+
+  // Step 2: change column type if exists
+  for (auto change_pair : node.GetChangedTypeColumns()) {
+    bool is_found = false;
+    oid_t i = 0;
+    for (; i < temp_schema->GetColumnCount(); ++i) {
+      if (temp_schema->GetColumn(i).GetName() == change_pair.first) {
+        is_found = true;
+        break;
+      }
+    }
+    if (!is_found) {
+      LOG_TRACE("Change column type failed: Column %s does not exists",
+                change_pair.first.c_str());
+      txn->SetResult(ResultType::FAILURE);
+      return false;
+    } else {
+      temp_schema->ChangeColumnType(i, change_pair.second);
+    }
+  }
 
-  if (txn->GetResult() == ResultType::SUCCESS) {
-    LOG_TRACE("Drop column succeed!");
+  // Step 3: append add column to new schema
+  // construct add column schema
+  std::vector<catalog::Column> add_columns;
+  for (size_t i = 0; i < node.GetAddedColumns().size(); ++i) {
+    for (auto column : node.GetAddedColumns()[i]->GetColumns()) {
+      add_columns.push_back(column);
+    }
+  }
+  std::unique_ptr<catalog::Schema> add_column_schema(
+      new catalog::Schema(add_columns));
+  // Check if added column exists
+  for (auto new_column : add_column_schema->GetColumns()) {
+    for (auto old_column : old_schema->GetColumns()) {
+      if (new_column.GetName() == old_column.GetName()) {
+        LOG_TRACE("Add column failed: Column %s already exists",
+                  new_column.GetName().c_str());
+        txn->SetResult(ResultType::FAILURE);
+        return false;
+      }
+    }
+  }
 
-    // TODO: Add on succeed logic if necessary
-    executor_context_->num_processed = 1;
-  } else {
-    LOG_TRACE("Result is: %s", ResultTypeToString(txn->GetResult()).c_str());
+  // Construct new schema
+  std::unique_ptr<catalog::Schema> new_schema(catalog::Schema::AppendSchema(
+      temp_schema.get(), add_column_schema.get()));
+
+  // Copy and replace table content to new schema in catalog
+  ResultType result = catalog::Catalog::GetInstance()->AlterTable(
+      database_oid, table_oid, new_schema, txn);
+  txn->SetResult(result);
+
+  switch (txn->GetResult()) {
+    case ResultType::SUCCESS:
+      LOG_TRACE("Alter table succeed!");
+
+      // TODO: Add on succeed logic if necessary
+      executor_context_->num_processed = 1;
+      break;
+    case ResultType::FAILURE:
+      LOG_TRACE("Alter table failed!");
+
+      // TODO: Add on failed logic if necessary
+      break;
+    default:
+      LOG_TRACE("Result is: %s", ResultTypeToString(txn->GetResult()).c_str());
+      break;
   }
   return false;
 }

src/include/catalog/catalog.h

@@ -181,7 +181,7 @@ class Catalog {
   // ALTER TABLE
   //===--------------------------------------------------------------------===//
   ResultType AlterTable(oid_t database_oid, oid_t table_oid,
-                        std::unique_ptr<catalog::Schema> new_schema,
+                        std::unique_ptr<catalog::Schema> &new_schema,
                         concurrency::TransactionContext *txn);
 
   ResultType AddColumn(const std::string &database_name,

src/include/catalog/column.h

@@ -33,9 +33,8 @@ class Column : public Printable {
     // Nothing to see...
   }
 
-  Column(type::TypeId value_type, size_t column_length,
-         std::string column_name, bool is_inlined = false,
-         oid_t column_offset = INVALID_OID)
+  Column(type::TypeId value_type, size_t column_length, std::string column_name,
+         bool is_inlined = false, oid_t column_offset = INVALID_OID)
       : column_name(column_name),
         column_type(value_type),
         fixed_length(INVALID_OID),
@@ -78,6 +77,8 @@ class Column : public Printable {
 
   inline type::TypeId GetType() const { return column_type; }
 
+  inline void SetType(const type::TypeId &new_type) { column_type = new_type; }
+
   inline bool IsInlined() const { return is_inlined; }
 
   inline bool IsPrimary() const { return is_primary_; }

src/include/catalog/schema.h

@@ -149,6 +149,11 @@ class Schema : public Printable {
     columns[column_id].column_name = new_name;
   }
 
+  inline void ChangeColumnType(const oid_t column_id,
+                               const type::TypeId &new_type) {
+    columns[column_id].SetType(new_type);
+  }
+
   inline oid_t GetUninlinedColumn(const oid_t column_id) const {
     return uninlined_columns[column_id];
   }

src/include/executor/alter_executor.h

@@ -44,11 +44,8 @@ class AlterExecutor : public AbstractExecutor {
   bool RenameColumn(const planner::AlterPlan &node,
                     concurrency::TransactionContext *txn);
 
-  bool DropColumn(const planner::AlterPlan &node,
+  bool AlterTable(const planner::AlterPlan &node,
                   concurrency::TransactionContext *txn);
-
- private:
-  bool isAlter_;
 };
 
 }  // executor

src/include/planner/alter_plan.h

@@ -75,12 +75,19 @@ class AlterPlan : public AbstractPlan {
 
   std::string GetDatabaseName() const { return database_name; }
 
-  // catalog::Schema *GetAddedColumns() const { return added_columns; }
+  const std::vector<std::unique_ptr<catalog::Schema>> &GetAddedColumns() const {
+    return added_columns;
+  }
 
   const std::vector<std::string> &GetDroppedColumns() const {
     return dropped_columns;
   }
 
+  const std::vector<std::pair<std::string, type::TypeId>> &
+  GetChangedTypeColumns() const {
+    return changed_type_columns;
+  };
+
   AlterType GetAlterTableType() const { return type; }
 
   // function used for rename statement