diff --git a/script/testing/junit/AlterBenchmarkTest.java b/script/testing/junit/AlterBenchmarkTest.java
new file mode 100644
index 00000000000..38d8ce42203
--- /dev/null
+++ b/script/testing/junit/AlterBenchmarkTest.java
@@ -0,0 +1,211 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// AlterTableTest.java
+//
+// Identification: script/testing/junit/AlterBenchmarkTest.java
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+import java.sql.*;
+import org.junit.*;
+import org.junit.rules.ExpectedException;
+import org.postgresql.util.PSQLException;
+import static org.junit.Assert.assertEquals;
+
+/*
+ * Test case that compare performance under different workload
+ * Will need to contact with different local SQL
+ */
+public class AlterBenchmarkTest extends PLTestBase {
+    private Connection conn;
+    private Connection conn2;
+
+    private static final String SQL_DROP_TABLE =
+            "DROP TABLE IF EXISTS tbl;";
+
+    private static final String SQL_CREATE_TABLE =
+            "CREATE TABLE tbl (" +
+                    "id integer, " +
+                    "year integer," +
+                    "month integer);";
+
+
+    @Rule
+    public ExpectedException thrown = ExpectedException.none();
+
+    /**
+     * Initialize the database and table for testing
+     */
+    private void InitDatabase() throws SQLException {
+        Statement stmt = conn.createStatement();
+        stmt.execute(SQL_DROP_TABLE);
+        stmt.execute(SQL_CREATE_TABLE);
+    }
+
+    public static Connection makePostgresConnection(String host,
+                                            int port,
+                                            String username,
+                                            String pass) throws SQLException {
+        String url = String.format("jdbc:postgresql://%s:%d/postgres",
+                host, port);
+        Connection conn = DriverManager.getConnection(url, username, pass);
+        return conn;
+    }
+
+    /**
+     * Setup the connection to peloton or other DBMS
+     * @throws SQLException
+     */
+    @Before
+    public void Setup() throws SQLException {
+        //connection to Postgres
+        //conn = makePostgresConnection("localhost", 5432, "dingshilun", "");
+        conn = makeDefaultConnection();
+        conn.setAutoCommit(true);
+        InitDatabase();
+    }
+
+    @After
+    public void Teardown() throws SQLException {
+        Statement stmt = conn.createStatement();
+        stmt.execute(SQL_DROP_TABLE);
+    }
+
+    /**
+     * Insert workload{} tuples into the table
+     * In order to test performancce variance under different workload
+     * @throws SQLException
+     */
+    @Test
+    public void test_tuple_number_varies() throws SQLException {
+        int[] workload = {};
+        for (int i = 0; i< workload.length;i++) {
+            // firstly use select * to make sure all tuples are in memory
+            // for postgres and other disk based DBMS
+            InitDatabase();
+            NumVarInsertHelper(workload[i]);
+            String sql = "select * from tbl;";
+            conn.createStatement().execute(sql);
+
+            try {
+                Thread.sleep(1000);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+            
+            String alterSql1 = "alter table tbl add day integer;";
+            long startTime1 = System.currentTimeMillis();
+            conn.createStatement().execute(alterSql1);
+            long endTime1 = System.currentTimeMillis();
+
+            String alterSql2 = "alter table tbl drop month;";
+            long startTime2 = System.currentTimeMillis();
+            conn.createStatement().execute(alterSql2);
+            long endTime2 = System.currentTimeMillis();
+
+            String alterSql3 = "alter table tbl alter year type varchar";
+            long startTime3 = System.currentTimeMillis();
+            conn.createStatement().execute(alterSql3);
+            long endTime3 = System.currentTimeMillis();
+
+            String alterSql4 = "alter table tbl alter year type integer USING year::INTEGER";
+            long startTime4 = System.currentTimeMillis();
+            conn.createStatement().execute(alterSql4);
+            long endTime4 = System.currentTimeMillis();
+
+            System.out.println("Alter add column " + workload[i] + " tuples took: " + (endTime1 - startTime1) + " milliseconds");
+            System.out.println("Alter drop column " + workload[i] + " tuples took: " + (endTime2 - startTime2) + " milliseconds");
+            System.out.println("Alter change type from inline to not inline " + workload[i] + " tuples took: " +
+                    (endTime3 - startTime3) + " milliseconds");
+            System.out.println("Alter change type from not inline to inline " + workload[i] + " tuples took: " +
+                    (endTime4 - startTime4) + " milliseconds");
+
+        }
+    }
+
+    private void NumVarInsertHelper(int insertNum) throws SQLException {
+        String sql = "INSERT INTO tbl VALUES (?, ?, ?);";
+        PreparedStatement pstmt = conn.prepareStatement(sql);
+        for (int i = 0; i < insertNum; i++) {
+            setValues(pstmt, new int [] {i, i+1, i+2});
+            pstmt.addBatch();
+        }
+        pstmt.executeBatch();
+    }
+
+    /**
+     * Insert 10000 tuple, and test performance under different
+     * length of the tuple
+     * @throws SQLException
+     */
+    @Test
+    public void test_tuple_length_variance() throws SQLException {
+        int[] workload = {};
+        int tupleNum = 10000;
+        String dropSQL = "DROP TABLE IF EXISTS tbl";
+        String sql = "";
+        conn.createStatement().execute(dropSQL);
+        for (int i = 0; i < workload.length; i++) {
+            sql = "CREATE TABLE tbl(id INTEGER PRIMARY KEY, " +
+                    "payload1 VARCHAR(" + workload[i] + ")," +
+                    "payload2 VARCHAR(" + workload[i] + ")," +
+                    "payload3 INTEGER);";
+            conn.createStatement().execute(sql);
+            LengthVarInsertHelper(tupleNum, workload[i]);
+
+            try {
+                Thread.sleep(1000);
+            } catch (Exception e) {
+                e.printStackTrace();
+            }
+
+            long startTime1 = System.currentTimeMillis();
+            conn.createStatement().execute("ALTER TABLE tbl add payload4 integer;");
+            long endTime1 = System.currentTimeMillis();
+
+            long startTime2 = System.currentTimeMillis();
+            conn.createStatement().execute("ALTER TABLE tbl drop payload1;");
+            long endTime2 = System.currentTimeMillis();
+
+            long startTime3 = System.currentTimeMillis();
+            conn.createStatement().execute("ALTER TABLE tbl alter payload3 type varchar;");
+            long endTime3 = System.currentTimeMillis();
+
+            System.out.println("Alter add column " + workload[i] + " length took: " + (endTime1 - startTime1)
+                    + " milliseconds");
+            System.out.println("Alter drop column " + workload[i] + " length took: " + (endTime2 - startTime2)
+                    + " milliseconds");
+            System.out.println("Alter change type from not inline to inline " + workload[i] + " length took: " +
+                    (endTime3 - startTime3) + " milliseconds");
+
+            conn.createStatement().execute(dropSQL);
+        }
+    }
+
+    // will simply generate string with length and return
+    private String PayloadGenerate(int length) {
+        long seed = System.currentTimeMillis() % 26 + 'a';
+        StringBuilder builder = new StringBuilder();
+        for (int i = 0; i < length; i++) {
+            builder.append((char)seed);
+        }
+        return builder.toString();
+    }
+
+    private void LengthVarInsertHelper(int insertNum, int varLen) throws SQLException {
+        String payload1 = PayloadGenerate(varLen);
+        String payload2 = PayloadGenerate(varLen);
+        for (int i = 0; i<insertNum; i++){
+            if( i%1000 == 0) {
+                payload1 = PayloadGenerate(varLen);
+                payload2 = PayloadGenerate(varLen);
+            }
+            String sql = String.format("INSERT INTO tbl VALUES (%d, '%s', '%s', %d);", i, payload1, payload2, i+1);
+            conn.createStatement().execute(sql);
+        }
+    }
+}
diff --git a/script/testing/junit/AlterTableTest.java b/script/testing/junit/AlterTableTest.java
new file mode 100644
index 00000000000..c134a74bc70
--- /dev/null
+++ b/script/testing/junit/AlterTableTest.java
@@ -0,0 +1,402 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// AlterTableTest.java
+//
+// Identification: script/testing/junit/AlterTableTest.java
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+import java.sql.*;
+import org.junit.*;
+import org.junit.rules.ExpectedException;
+import org.postgresql.util.PSQLException;
+import static org.junit.Assert.assertEquals;
+
+public class AlterTableTest extends PLTestBase {
+    private Connection conn;
+    private Connection conn2;
+
+    private static final String SQL_DROP_TABLE =
+            "DROP TABLE IF EXISTS foo;";
+
+    private static final String SQL_CREATE_TABLE =
+            "CREATE TABLE foo (" +
+                    "id integer, " +
+                    "year integer);";
+
+    private static final String SQL_SELECT_STAR = "SELECT * FROM foo;";
+
+    private static final String SQL_RENAME_COLUMN =
+            "ALTER TABLE foo RENAME year to month;";
+
+    private static final double EPSILON = 1e-6;
+
+
+    @Rule
+    public ExpectedException thrown = ExpectedException.none();
+
+    /**
+     * Initialize the database and table for testing
+     */
+    private void InitDatabase() throws SQLException {
+        Statement stmt = conn.createStatement();
+        stmt.execute(SQL_DROP_TABLE);
+        stmt.execute(SQL_CREATE_TABLE);
+
+        String sql = "INSERT INTO foo VALUES (5, 400);";
+        conn.createStatement().execute(sql);
+    }
+
+    @Before
+    public void Setup() throws SQLException {
+        conn = makeDefaultConnection();
+        conn.setAutoCommit(true);
+        conn2 = makeDefaultConnection();
+        InitDatabase();
+    }
+
+    @After
+    public void Teardown() throws SQLException {
+        Statement stmt = conn.createStatement();
+        stmt.execute(SQL_DROP_TABLE);
+    }
+
+    /**
+     * Insert 1 tuple, rename the column, and see if the change is visible.
+     */
+    @Test
+    public void test_RenameCol_Base() throws SQLException {
+        conn.createStatement().execute(SQL_RENAME_COLUMN);
+        ResultSet rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        checkRow(rs,
+                new String [] {"id", "month"},
+                new int [] {5, 400});
+        assertNoMoreRows(rs);
+    }
+
+    /**
+     * Rename a column that does not exists, should throw exception
+     */
+    @Test
+    public void test_RenameCol_NotExist() throws SQLException {
+        String sql = "ALTER TABLE foo RENAME a to b;";
+
+        // Old column does not exist
+        thrown.expect(PSQLException.class);
+        conn.createStatement().execute(sql);
+    }
+
+    /**
+     * Rename a column to a name that already exists, should throw exception
+     */
+    @Test
+    public void test_RenameCol_Exist() throws SQLException {
+        String sql = "ALTER TABLE foo RENAME year to id;";
+
+        // New column already exists
+        thrown.expect(PSQLException.class);
+        conn.createStatement().execute(sql);
+    }
+
+    /**
+     * Two transactions try to rename at the same time, should throw exception
+     */
+    @Test
+    public void test_RenameCol_Concurrent() throws SQLException {
+        conn.setAutoCommit(false);
+        conn2.setAutoCommit(false);
+
+        conn.createStatement().execute(SQL_RENAME_COLUMN);
+
+        thrown.expect(PSQLException.class);
+        conn2.createStatement().execute(SQL_RENAME_COLUMN);
+
+        conn.commit();
+        conn2.commit();
+    }
+
+//    The following tests are currently broken.
+//
+//    /**
+//     * 2 transactions, t2 reads the table between t1 executes the rename
+//     * and commits the changes.
+//     */
+//    @Test
+//    public void test_RenameCol_ReadBeforeCommit() throws SQLException {
+//        conn.setAutoCommit(false);
+//        conn2.setAutoCommit(false);
+//
+//        String sql = "ALTER TABLE foo RENAME year to month;";
+//        conn.createStatement().execute(sql);
+//
+//        ResultSet rs = conn2.createStatement().executeQuery(SQL_SELECT_STAR);
+//        rs.next();
+//        checkRow(rs,
+//                new String [] {"id", "year"},
+//                new int [] {5, 400});
+//        assertNoMoreRows(rs);
+//
+//        conn.commit();
+//        conn2.commit();
+//    }
+//
+//    /**
+//     * 2 transactions, t2 reads the table before and after t1 changes the
+//     * column name and should not see the changes.
+//     */
+//    @Test
+//    public void test_RenameCol_ReadBeforeAndAfterCommit() throws SQLException {
+//        conn.setAutoCommit(false);
+//        conn2.setAutoCommit(false);
+//
+//        ResultSet rs_1 = conn2.createStatement().executeQuery(SQL_SELECT_STAR);
+//        rs_1.next();
+//        checkRow(rs_1,
+//                new String [] {"id", "year"},
+//                new int [] {5, 400});
+//        assertNoMoreRows(rs_1);
+//
+//        conn.createStatement().execute(SQL_RENAME_COLUMN);
+//        conn.commit();
+//
+//        ResultSet rs_2 = conn2.createStatement().executeQuery(SQL_SELECT_STAR);
+//        rs_2.next();
+//        checkRow(rs_2,
+//                new String [] {"id", "year"},
+//                new int [] {5, 400});
+//        assertNoMoreRows(rs_2);
+//        conn2.commit();
+//    }
+//
+//    /**
+//     * 2 transactions, t2 reads the table after t1 commits the changes.
+//     */
+//    @Test
+//    public void test_RenameCol_ReadAfterCommit() throws SQLException {
+//        conn.setAutoCommit(false);
+//        conn2.setAutoCommit(false);
+//
+//        Statement alter_statement = conn.createStatement();
+//        Statement select_statement = conn2.createStatement();
+//
+//        alter_statement.execute(SQL_RENAME_COLUMN);
+//        conn.commit();
+//
+//        ResultSet rs = select_statement.executeQuery(SQL_SELECT_STAR);
+//        rs.next();
+//        checkRow(rs,
+//                new String [] {"id", "year"},
+//                new int [] {5, 400});
+//        assertNoMoreRows(rs);
+//        conn2.commit();
+//    }
+
+    /**
+     * Add a column to the table, and do some insertion.
+     */
+    @Test
+    public void test_AddCol_Basic() throws SQLException {
+        String sql = "ALTER TABLE foo ADD month int;";
+        conn.createStatement().execute(sql);
+        ResultSet rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        checkRow(rs,
+                new String [] {"id", "year", "month"},
+                new int [] {5, 400, 0});
+        assertNoMoreRows(rs);
+
+        String sql2 = "INSERT INTO foo VALUES (6, 500, 1);";
+        conn.createStatement().execute(sql2);
+        rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        checkRow(rs,
+                new String [] {"id", "year", "month"},
+                new int [] {5, 400, 0});
+        rs.next();
+        checkRow(rs,
+                new String [] {"id", "year", "month"},
+                new int [] {6, 500, 1});
+        assertNoMoreRows(rs);
+    }
+
+    /**
+     * Add a column to a name that already exists, should throw exception
+     */
+    @Test
+    public void test_AddCol_Exist() throws SQLException {
+        String sql = "ALTER TABLE foo ADD year int;";
+
+        // New column already exists
+        thrown.expect(PSQLException.class);
+        conn.createStatement().execute(sql);
+    }
+
+    /**
+     * Drop a column from the table.
+     */
+    @Test
+    public void test_DropCol_Basic() throws SQLException {
+        String sql = "ALTER TABLE foo DROP year;";
+        conn.createStatement().execute(sql);
+        ResultSet rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        checkRow(rs,
+                new String [] {"id"},
+                new int [] {5});
+        assertNoMoreRows(rs);
+
+        String sql2 = "INSERT INTO foo VALUES (6);";
+        conn.createStatement().execute(sql2);
+        rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        checkRow(rs,
+                new String [] {"id"},
+                new int [] {5});
+        rs.next();
+        checkRow(rs,
+                new String [] {"id"},
+                new int [] {6});
+        assertNoMoreRows(rs);
+    }
+
+    /**
+     * Drop a column that does not exists, should throw exception
+     */
+    @Test
+    public void test_DropCol_NotExist() throws SQLException {
+        String sql = "ALTER TABLE foo DROP month;";
+
+        // Old column does not exist
+        thrown.expect(PSQLException.class);
+        conn.createStatement().execute(sql);
+    }
+
+    /**
+     * Alter column type from int to float and then alter to int again.
+     */
+    @Test
+    public void test_AlterType_Basic() throws SQLException {
+        String sql = "ALTER TABLE foo ALTER year TYPE float;";
+        conn.createStatement().execute(sql);
+        ResultSet rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        assertEquals(rs.getInt("id"), 5);
+        assertEquals(rs.getFloat("year"), 400, EPSILON);
+        assertNoMoreRows(rs);
+
+        String sql2 = "INSERT INTO foo VALUES (6, 3.5);";
+        conn.createStatement().execute(sql2);
+        rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        assertEquals(rs.getInt("id"), 5);
+        assertEquals(rs.getFloat("year"), 400, EPSILON);
+        rs.next();
+        assertEquals(rs.getInt("id"), 6);
+        assertEquals(rs.getFloat("year"), 3.5, EPSILON);
+        assertNoMoreRows(rs);
+
+        String sql3 = "ALTER TABLE foo ALTER year TYPE int;";
+        conn.createStatement().execute(sql3);
+        rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        assertEquals(rs.getInt("id"), 5);
+        assertEquals(rs.getInt("year"), 400);
+        rs.next();
+        assertEquals(rs.getInt("id"), 6);
+        assertEquals(rs.getInt("year"), 3);
+        assertNoMoreRows(rs);
+    }
+
+    /**
+     * Alter column type from int to varchar and backwards.
+     */
+    @Test
+    public void test_AlterType_Varchar() throws SQLException {
+        String sql = "ALTER TABLE foo ALTER year TYPE varchar;";
+        conn.createStatement().execute(sql);
+        ResultSet rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        assertEquals(rs.getInt("id"), 5);
+        assertEquals(rs.getString("year"), Integer.toString(400));
+        assertNoMoreRows(rs);
+
+        String sql2 = "ALTER TABLE foo ALTER year TYPE int;";
+        conn.createStatement().execute(sql2);
+        rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        assertEquals(rs.getInt("id"), 5);
+        assertEquals(rs.getInt("year"), 400);
+        assertNoMoreRows(rs);
+    }
+
+    /**
+     * Alter type to column that does not exist
+     */
+    @Test
+    public void test_AlterType_NonExist() throws SQLException {
+        String sql = "ALTER TABLE foo ALTER a TYPE int;";
+
+        thrown.expect(PSQLException.class);
+        conn.createStatement().execute(sql);
+    }
+
+    /**
+     * Alter to an unsupported column type
+     */
+    @Test
+    public void test_AlterType_UnSupported() throws SQLException {
+        String sql = "ALTER TABLE foo ALTER year TYPE non;";
+        thrown.expect(PSQLException.class);
+        conn.createStatement().execute(sql);
+    }
+
+    /**
+     * Add columns, drop columns, change column type in one sql statement
+     */
+    @Test
+    public void test_MultiOperation() throws SQLException {
+        String sql =
+            "ALTER TABLE foo ADD month INT, DROP year, ALTER id TYPE float;";
+        conn.createStatement().execute(sql);
+        ResultSet rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        assertEquals(rs.getFloat("id"), 5, EPSILON);
+        assertEquals(rs.getInt("month"), 0);
+        assertNoMoreRows(rs);
+
+        String sql2 = "INSERT INTO foo VALUES (4.5, 3);";
+        conn.createStatement().execute(sql2);
+        rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        assertEquals(rs.getFloat("id"), 5, EPSILON);
+        assertEquals(rs.getInt("month"), 0);
+        rs.next();
+        assertEquals(rs.getFloat("id"), 4.5, EPSILON);
+        assertEquals(rs.getInt("month"), 3);
+        assertNoMoreRows(rs);
+    }
+
+    /**
+     * If multiple operations in one statement failed, schema will not change
+     */
+    @Test
+    public void test_MultiOperationFailed() throws SQLException {
+        String sql =
+          "ALTER TABLE foo ADD month int, DROP month, ALTER year TYPE float;";
+
+        thrown.expect(PSQLException.class);
+        conn.createStatement().execute(sql);
+
+        ResultSet rs = conn.createStatement().executeQuery(SQL_SELECT_STAR);
+        rs.next();
+        checkRow(rs,
+                new String [] {"id", "year"},
+                new int [] {5, 400});
+        assertNoMoreRows(rs);
+    }
+}
diff --git a/src/binder/bind_node_visitor.cpp b/src/binder/bind_node_visitor.cpp
index a6ffe17b322..dd3416ed1bc 100644
--- a/src/binder/bind_node_visitor.cpp
+++ b/src/binder/bind_node_visitor.cpp
@@ -187,6 +187,9 @@ void BindNodeVisitor::Visit(parser::DropStatement *node) {
 void BindNodeVisitor::Visit(parser::PrepareStatement *) {}
 void BindNodeVisitor::Visit(parser::ExecuteStatement *) {}
 void BindNodeVisitor::Visit(parser::TransactionStatement *) {}
+void BindNodeVisitor::Visit(parser::AlterTableStatement *node) {
+  node->TryBindDatabaseName(default_database_name_);
+}
 void BindNodeVisitor::Visit(parser::AnalyzeStatement *node) {
   node->TryBindDatabaseName(default_database_name_);
 }
diff --git a/src/catalog/catalog.cpp b/src/catalog/catalog.cpp
index 90b9d13a62a..64c952febbb 100644
--- a/src/catalog/catalog.cpp
+++ b/src/catalog/catalog.cpp
@@ -27,11 +27,16 @@
 #include "catalog/table_metrics_catalog.h"
 #include "catalog/trigger_catalog.h"
 #include "concurrency/transaction_manager_factory.h"
+#include "executor/executor_context.h"
+#include "executor/insert_executor.h"
+#include "executor/seq_scan_executor.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 "planner/insert_plan.h"
+#include "planner/seq_scan_plan.h"
 #include "settings/settings_manager.h"
 #include "storage/storage_manager.h"
 #include "storage/table_factory.h"
@@ -930,6 +935,287 @@ std::shared_ptr<SystemCatalogs> Catalog::GetSystemCatalogs(
   return catalog_map_[database_oid];
 }
 
+//===--------------------------------------------------------------------===//
+// ALTER TABLE
+//===--------------------------------------------------------------------===//
+
+/**
+ * @brief Helper function for alter table, called internally
+ * @param database_oid database to which the table belongs to
+ * @param table_oid table to which the column belongs to
+ * @param new_schema the new table schema
+ * @param txn the transaction Context
+ * @return TransactionContext ResultType(SUCCESS or FAILURE)
+ */
+ResultType Catalog::AlterTable(oid_t database_oid, oid_t table_oid, const std::string &schema_name,
+                               std::unique_ptr<catalog::Schema> &new_schema,
+                               concurrency::TransactionContext *txn) {
+  LOG_TRACE("AlterTable in Catalog");
+
+  if (txn == nullptr)
+    throw CatalogException("Alter table requires transaction");
+  try {
+    auto storage_manager = storage::StorageManager::GetInstance();
+    auto database = storage_manager->GetDatabaseWithOid(database_oid);
+    try {
+      auto old_table = database->GetTableWithOid(table_oid);
+      auto old_schema = old_table->GetSchema();
+      auto pg_index = catalog_map_[database_oid]->GetIndexCatalog();
+
+      // Step 1: build empty table with new schema
+      bool own_schema = true;
+      bool adapt_table = false;
+      auto new_table = storage::TableFactory::GetDataTable(
+          database_oid, table_oid,
+          catalog::Schema::CopySchema(new_schema.get()), old_table->GetName(),
+          DEFAULT_TUPLES_PER_TILEGROUP, own_schema, adapt_table);
+      // Step 2: Copy indexes
+      auto old_index_oids = pg_index->GetIndexObjects(table_oid, txn);
+      for (auto index_oid_pair : old_index_oids) {
+        oid_t index_oid = index_oid_pair.first;
+        // delete record in pg_index
+        pg_index->DeleteIndex(index_oid, txn);
+        // Check if all indexed columns still exists
+        auto old_index = old_table->GetIndexWithOid(index_oid);
+        bool index_exist = true;
+        std::vector<oid_t> new_key_attrs;
+
+        for (oid_t column_id : old_index->GetMetadata()->GetKeyAttrs()) {
+          bool is_found = false;
+          std::string column_name = old_schema->GetColumn(column_id).GetName();
+          oid_t i = 0;
+          for (auto new_column : new_schema->GetColumns()) {
+            if (column_name == new_column.GetName()) {
+              is_found = true;
+              new_key_attrs.push_back(i);
+              break;
+            }
+            i++;
+          }
+          if (!is_found) {
+            index_exist = false;
+            break;
+          }
+        }
+        if (!index_exist) continue;
+
+        // construct index on new table
+        auto index_metadata = new index::IndexMetadata(
+            old_index->GetName(), index_oid, table_oid, database_oid,
+            old_index->GetMetadata()->GetIndexType(),
+            old_index->GetMetadata()->GetIndexConstraintType(),
+            new_schema.get(),
+            catalog::Schema::CopySchema(new_schema.get(), new_key_attrs),
+            new_key_attrs, old_index->GetMetadata()->HasUniqueKeys());
+
+        std::shared_ptr<index::Index> new_index(
+            index::IndexFactory::GetIndex(index_metadata));
+        new_table->AddIndex(new_index);
+
+        // reinsert record into pg_index
+        pg_index->InsertIndex(
+            index_oid, old_index->GetName(), table_oid, schema_name,
+            old_index->GetMetadata()->GetIndexType(),
+            old_index->GetMetadata()->GetIndexConstraintType(),
+            old_index->GetMetadata()->HasUniqueKeys(), new_key_attrs,
+            pool_.get(), txn);
+      }
+      std::unique_ptr<executor::ExecutorContext> context(
+          new executor::ExecutorContext(txn, {}));
+      // Step 3: build column mapping between old table and new table
+      // we're using column name as unique identifier
+      std::vector<oid_t> old_column_ids;
+      std::unordered_map<oid_t, oid_t> column_map;
+      for (oid_t old_column_id = 0;
+           old_column_id < old_schema->GetColumnCount(); old_column_id++) {
+        old_column_ids.push_back(old_column_id);
+        for (oid_t new_column_id = 0;
+             new_column_id < new_schema->GetColumnCount(); new_column_id++) {
+          if (old_schema->GetColumn(old_column_id).GetName() ==
+              new_schema->GetColumn(new_column_id).GetName()) {
+            column_map[new_column_id] = old_column_id;
+          }
+        }
+      }
+      // Step 4: Get tuples from old table with sequential scan
+      // TODO: Try to reuse Sequential scan function and insert function in
+      // abstract catalog
+      planner::SeqScanPlan seq_scan_node(old_table, nullptr, old_column_ids);
+      executor::SeqScanExecutor seq_scan_executor(&seq_scan_node,
+                                                  context.get());
+      seq_scan_executor.Init();
+      while (seq_scan_executor.Execute()) {
+        std::unique_ptr<executor::LogicalTile> result_tile(
+            seq_scan_executor.GetOutput());
+        for (size_t i = 0; i < result_tile->GetTupleCount(); i++) {
+          // Transform tuple into new schema
+          std::unique_ptr<storage::Tuple> tuple(
+              new storage::Tuple(new_schema.get(), true));
+
+          for (oid_t new_column_id = 0;
+               new_column_id < new_schema->GetColumnCount(); new_column_id++) {
+            auto it = column_map.find(new_column_id);
+            type::Value val;
+            if (it == column_map.end()) {
+              // new column, set value to null
+              val = type::ValueFactory::GetNullValueByType(
+                  new_schema->GetColumn(new_column_id).GetType());
+            } else {
+              // otherwise, copy value in old table
+              val = result_tile->GetValue(i, it->second);
+              if (new_schema->GetColumn(new_column_id).GetType() !=
+                  old_schema->GetColumn(it->second).GetType()) {
+                // change the value's type
+                LOG_TRACE(
+                    "CASTED: %s TO %s", val.GetInfo().c_str(),
+                    new_schema->GetColumn(new_column_id).GetInfo().c_str());
+                auto casted_val =
+                    val.CastAs(new_schema->GetColumn(new_column_id).GetType());
+                tuple->SetValue(new_column_id, casted_val, pool_.get());
+              } else {
+                tuple->SetValue(new_column_id, val, pool_.get());
+              }
+            }
+          }
+          // insert new tuple into new table
+          planner::InsertPlan node(new_table, std::move(tuple));
+          executor::InsertExecutor executor(&node, context.get());
+          executor.Init();
+          executor.Execute();
+        }
+      }
+      // Step 5: delete all the column(attribute) records in pg_attribute
+      // and reinsert them using new schema(column offset needs to change
+      // accordingly)
+      auto pg_attributes =
+          catalog_map_[database_oid]->GetColumnCatalog();
+      pg_attributes->DeleteColumns(table_oid, txn);
+      oid_t column_offset = 0;
+      for (auto new_column : new_schema->GetColumns()) {
+        pg_attributes->InsertColumn(
+            table_oid, new_column.GetName(), column_offset,
+            new_column.GetOffset(), new_column.GetType(),
+            new_column.IsInlined(), new_column.GetConstraints(), pool_.get(),
+            txn);
+        column_offset++;
+      }
+      // Garbage collection
+      txn->RecordDropTable(old_table);
+
+      // Final step of physical change should be moved to commit time
+      database->ReplaceTableWithOid(table_oid, new_table);
+
+      LOG_TRACE("Alter table with oid %d succeed.", table_oid);
+    } catch (CatalogException &e) {
+      LOG_TRACE("Alter table failed.");
+      return ResultType::FAILURE;
+    }
+  } catch (CatalogException &e) {
+    return ResultType::FAILURE;
+  }
+  return ResultType::SUCCESS;
+}
+
+/**
+ * @brief Add new columns to the table.
+ * @param database_name database to which the table belongs to
+ * @param table_name table to which the column belongs to
+ * @param columns the column to be added
+ * @param txn the transaction Context
+ * @return TransactionContext ResultType(SUCCESS or FAILURE)
+ *
+ */
+ResultType Catalog::AddColumn(
+    UNUSED_ATTRIBUTE const std::string &database_name,
+    UNUSED_ATTRIBUTE const std::string &table_name,
+    UNUSED_ATTRIBUTE const std::vector<std::string> &columns,
+    UNUSED_ATTRIBUTE concurrency::TransactionContext *txn) {
+  // TODO: perform ADD Operation
+  return ResultType::SUCCESS;
+}
+
+/**
+ * @brief Drop the column from the table.
+ * @param database_name database to which the table belongs to
+ * @param table_name table to which the columns belong to
+ * @param columns the columns to be dropped
+ * @param txn the transaction Context
+ * @return TransactionContext ResultType(SUCCESS or FAILURE)
+ */
+
+ResultType Catalog::DropColumn(UNUSED_ATTRIBUTE const std::string &database_name,
+                               UNUSED_ATTRIBUTE const std::string &table_name,
+                               UNUSED_ATTRIBUTE const std::vector<std::string> &columns,
+                               UNUSED_ATTRIBUTE concurrency::TransactionContext *txn) {
+  return ResultType::SUCCESS;
+}
+
+/**
+ * @brief Change the column name in the catalog.
+ * @param database_name database to which the table belongs to
+ * @param table_name table to which the column belongs to
+ * @param columns the column to be dropped
+ * @param txn the transaction Context
+ * @return TransactionContext ResultType(SUCCESS or FAILURE)
+ */
+ResultType Catalog::RenameColumn(const std::string &database_name,
+                                 const std::string &table_name,
+                                 const std::string &old_name,
+                                 const std::string &new_name,
+                                 const std::string &schema_name,
+                                 concurrency::TransactionContext *txn) {
+  if (txn == nullptr) {
+    throw CatalogException("Change Column requires transaction.");
+  }
+
+  if (new_name.size() == 0) {
+    throw CatalogException("Name can not be empty string.");
+  }
+
+  LOG_TRACE("Change Column Name %s to %s", old_name.c_str(), new_name.c_str());
+
+  try {
+    // Get table from the name
+    auto table = Catalog::GetInstance()->GetTableWithName(database_name, schema_name,
+                                                          table_name, txn);
+    auto schema = table->GetSchema();
+
+    // Currently we only support change the first column name!
+
+    // Check the validity of old name and the new name
+    oid_t columnId = schema->GetColumnID(new_name);
+    if (columnId != INVALID_OID) {
+      throw CatalogException("New column already exists in the table.");
+    }
+    columnId = schema->GetColumnID(old_name);
+    if (columnId == INVALID_OID) {
+      throw CatalogException("Old column does not exist in the table.");
+    }
+
+    // Change column name in the global schema
+    schema->ChangeColumnName(columnId, new_name);
+
+    // Modify the pg_table
+    oid_t table_oid = Catalog::GetInstance()
+                          ->GetTableObject(database_name, schema_name, table_name, txn)
+                          ->GetTableOid();
+    oid_t database_oid = Catalog::GetInstance()
+                          ->GetTableObject(database_name, schema_name, table_name, txn)
+                          ->GetDatabaseOid();
+    auto pg_attributes =
+        catalog_map_[database_oid]->GetColumnCatalog();
+    bool res = pg_attributes->RenameColumn(
+        database_oid, table_oid, old_name, new_name, txn);
+    if (!res) {
+      throw CatalogException("Change Column name failed.");
+    }
+
+  } catch (CatalogException &e) {
+    return ResultType::FAILURE;
+  }
+  return ResultType::SUCCESS;
+}
+
 //===--------------------------------------------------------------------===//
 // DEPRECATED
 //===--------------------------------------------------------------------===//
diff --git a/src/catalog/column_catalog.cpp b/src/catalog/column_catalog.cpp
index 8262ffe479c..18b67c2daf2 100644
--- a/src/catalog/column_catalog.cpp
+++ b/src/catalog/column_catalog.cpp
@@ -249,7 +249,38 @@ ColumnCatalog::GetColumnObjects(oid_t table_oid,
     }
   }
 
-  return table_object->GetColumnObjects();
+  return table_object->GetColumnObjects(true);
+}
+
+/*
+ * @brief Rename the column name to a new name.
+ * @ return whether the update succeed
+ */
+bool ColumnCatalog::RenameColumn(oid_t database_oid,
+                                 oid_t table_oid,
+                                 const std::string &column_name,
+                                 const std::string &new_name,
+                                 concurrency::TransactionContext *txn) {
+  std::vector<oid_t> update_columns({ColumnId::COLUMN_NAME});
+
+  std::vector<type::Value> update_values;
+  update_values.push_back(type::ValueFactory::GetVarcharValue(new_name).Copy());
+
+  // values to execute index scan
+  std::vector<type::Value> scan_values;
+  scan_values.push_back(type::ValueFactory::GetIntegerValue(table_oid).Copy());
+  scan_values.push_back(
+      type::ValueFactory::GetVarcharValue(column_name, nullptr).Copy());
+
+  // Index of table_oid & column_name
+  oid_t index_offset = IndexId::PRIMARY_KEY;
+
+  auto table_object =
+      Catalog::GetInstance()->GetTableObject(database_oid, table_oid, txn);
+  table_object->EvictColumnObject(column_name);
+
+  return UpdateWithIndexScan(update_columns, update_values, scan_values,
+                             index_offset, txn);
 }
 
 }  // namespace catalog
diff --git a/src/common/internal_types.cpp b/src/common/internal_types.cpp
index 6d95bf1a35c..72dfd9f9306 100644
--- a/src/common/internal_types.cpp
+++ b/src/common/internal_types.cpp
@@ -427,6 +427,45 @@ std::ostream &operator<<(std::ostream &os, const DropType &type) {
   return os;
 }
 
+std::string AlterTypeToString(AlterType type) {
+  // TODO: add other AlterType
+  switch (type) {
+    case AlterType::INVALID: {
+      return "INVALID";
+    }
+    case AlterType::RENAME: {
+      return "RENAME";
+    }
+    case AlterType::ALTER: {
+      return "ALTER";
+    }
+    default: {
+      throw ConversionException(
+          StringUtil::Format("No string conversion for AlterType value '%d'",
+                             static_cast<int>(type)));
+    }
+  }
+}
+
+AlterType StringToAlterType(const std::string &str) {
+  std::string upper_str = StringUtil::Upper(str);
+  if (upper_str == "INVALID") {
+    return AlterType::INVALID;
+  } else if (upper_str == "RENAME") {
+    return AlterType::RENAME;
+  } else if (upper_str == "ALTER") {
+    return AlterType::ALTER;
+  } else {
+    throw ConversionException(StringUtil::Format(
+        "No AlterType conversion from string '%s'", upper_str.c_str()));
+  }
+  return AlterType::INVALID;
+}
+std::ostream &operator<<(std::ostream &os, const AlterType &type) {
+  os << AlterTypeToString(type);
+  return os;
+}
+
 //===--------------------------------------------------------------------===//
 // Statement - String Utilities
 //===--------------------------------------------------------------------===//
@@ -1391,6 +1430,12 @@ std::string PlanNodeTypeToString(PlanNodeType type) {
     case PlanNodeType::ANALYZE: {
       return ("ANALYZE");
     }
+    case PlanNodeType::RENAME: {
+      return ("RENAME");
+    }
+    case PlanNodeType::ALTER: {
+      return ("ALTER");
+    }
     default: {
       throw ConversionException(
           StringUtil::Format("No string conversion for PlanNodeType value '%d'",
@@ -1462,6 +1507,10 @@ PlanNodeType StringToPlanNodeType(const std::string &str) {
     return PlanNodeType::MOCK;
   } else if (upper_str == "ANALYZE") {
     return PlanNodeType::ANALYZE;
+  } else if (upper_str == "RENAME") {
+    return PlanNodeType::RENAME;
+  } else if (upper_str == "ALTER") {
+    return PlanNodeType::ALTER;
   } else {
     throw ConversionException(StringUtil::Format(
         "No PlanNodeType conversion from string '%s'", upper_str.c_str()));
diff --git a/src/executor/alter_executor.cpp b/src/executor/alter_executor.cpp
new file mode 100644
index 00000000000..4f0b0fc09b9
--- /dev/null
+++ b/src/executor/alter_executor.cpp
@@ -0,0 +1,187 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// alter_executor.cpp
+//
+// Identification: src/executor/alter_executor.cpp
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#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) {}
+
+// Initialize executor
+// Nothing to initialize for now
+bool AlterExecutor::DInit() {
+  LOG_TRACE("Initializing Alter Executer...");
+
+  LOG_TRACE("Alter Executor initialized!");
+  return true;
+}
+
+bool AlterExecutor::DExecute() {
+  LOG_TRACE("Executing Alter...");
+  bool result = false;
+  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;
+}
+
+bool AlterExecutor::RenameColumn(
+    const peloton::planner::AlterPlan &node,
+    peloton::concurrency::TransactionContext *txn) {
+  auto database_name = node.GetDatabaseName();
+  auto table_name = node.GetTableName();
+  auto schema_name = node.GetSchemaName();
+  auto new_column_name = node.GetNewName();
+  auto old_column_name = node.GetOldName();
+
+  ResultType result = catalog::Catalog::GetInstance()->RenameColumn(
+      database_name, table_name, old_column_name, new_column_name, schema_name, txn);
+  txn->SetResult(result);
+
+  if (txn->GetResult() == ResultType::SUCCESS) {
+    LOG_TRACE("Rename column succeeded!");
+
+    // TODO: Add on succeed logic if necessary
+    executor_context_->num_processed = 1;
+  } else {
+    LOG_TRACE("Result is: %s", ResultTypeToString(txn->GetResult()).c_str());
+  }
+  return false;
+}
+
+bool AlterExecutor::AlterTable(const peloton::planner::AlterPlan &node,
+                               peloton::concurrency::TransactionContext *txn) {
+  auto database_name = node.GetDatabaseName();
+  auto table_name = node.GetTableName();
+  auto schema_name = node.GetSchemaName();
+  auto table_catalog_obj = catalog::Catalog::GetInstance()->GetTableObject(
+      database_name, schema_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, schema_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));
+  auto columns = temp_schema->GetColumns();
+  // Step 2: change column type if exists
+  for (auto &change_col : node.GetChangedTypeColumns().get()->GetColumns()) {
+    bool is_found = false;
+    oid_t i = 0;
+    for (; i < columns.size(); ++i) {
+      if (columns[i].GetName() == change_col.GetName()) {
+        is_found = true;
+        break;
+      }
+    }
+    if (!is_found) {
+      LOG_TRACE("Change column type failed: Column %s does not exists",
+                change_col.GetName().c_str());
+      txn->SetResult(ResultType::FAILURE);
+      return false;
+    } else {
+      columns[i] = std::move(change_col);
+      // TODO: decide VARCHAR's size when change type
+      // if (change_pair.second == type::TypeId::VARCHAR) {}
+    }
+  }
+
+  // Step 3: append add column to new schema
+  // construct add column schema
+  std::vector<catalog::Column> add_columns;
+  for (auto column : node.GetAddedColumns()->GetColumns()) {
+    add_columns.push_back(column);
+  }
+  // Check if added column exists
+  for (auto new_column : add_columns) {
+    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;
+      }
+    }
+  }
+  columns.insert(columns.end(), add_columns.begin(), add_columns.end());
+  // Construct new schema
+  std::unique_ptr<catalog::Schema> new_schema(new catalog::Schema(columns));
+
+  // Copy and replace table content to new schema in catalog
+  ResultType result = catalog::Catalog::GetInstance()->AlterTable(
+      database_oid, table_oid, schema_name, 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;
+}
+
+}  // executor
+}  // peloton
diff --git a/src/executor/plan_executor.cpp b/src/executor/plan_executor.cpp
index 104aff1351c..74105756902 100644
--- a/src/executor/plan_executor.cpp
+++ b/src/executor/plan_executor.cpp
@@ -64,7 +64,7 @@ static void CompileAndExecutePlan(
   }
 
   auto on_query_result =
-    [&on_complete, &consumer, plan](executor::ExecutionResult result) {
+      [&on_complete, &consumer, plan](executor::ExecutionResult result) {
         std::vector<ResultValue> values;
         for (const auto &tuple : consumer.GetOutputTuples()) {
           for (uint32_t i = 0; i < tuple.tuple_.size(); i++) {
@@ -336,7 +336,10 @@ executor::AbstractExecutor *BuildExecutorTree(
       child_executor =
           new executor::PopulateIndexExecutor(plan, executor_context);
       break;
-
+    case PlanNodeType::ALTER:
+      child_executor =
+          new executor::AlterExecutor(plan, executor_context);
+      break;
     default:
       LOG_ERROR("Unsupported plan node type : %s",
                 PlanNodeTypeToString(plan_node_type).c_str());
diff --git a/src/gc/transaction_level_gc_manager.cpp b/src/gc/transaction_level_gc_manager.cpp
index c2274a330e0..077c66dc258 100644
--- a/src/gc/transaction_level_gc_manager.cpp
+++ b/src/gc/transaction_level_gc_manager.cpp
@@ -278,6 +278,11 @@ void TransactionLevelGCManager::AddToRecycleMap(
     LOG_DEBUG("GCing index %u", index_oid);
   }
 
+  for (auto table : txn_ctx->GetDroppedTables()) {
+    LOG_TRACE("deleting table when txn commits");
+    delete table;
+  }
+
   delete txn_ctx;
 }
 
diff --git a/src/include/binder/bind_node_visitor.h b/src/include/binder/bind_node_visitor.h
index 9ca09c68693..3812e1f8e0d 100644
--- a/src/include/binder/bind_node_visitor.h
+++ b/src/include/binder/bind_node_visitor.h
@@ -67,6 +67,7 @@ class BindNodeVisitor : public SqlNodeVisitor {
   void Visit(parser::UpdateStatement *) override;
   void Visit(parser::CopyStatement *) override;
   void Visit(parser::AnalyzeStatement *) override;
+  void Visit(parser::AlterTableStatement *) override;
 
   void Visit(expression::CaseExpression *expr) override;
   void Visit(expression::SubqueryExpression *expr) override;
diff --git a/src/include/catalog/catalog.h b/src/include/catalog/catalog.h
index cd19d7fe72b..2d89e35db01 100644
--- a/src/include/catalog/catalog.h
+++ b/src/include/catalog/catalog.h
@@ -185,6 +185,30 @@ class Catalog {
       oid_t database_oid, oid_t table_oid,
       concurrency::TransactionContext *txn);
 
+  //===--------------------------------------------------------------------===//
+  // ALTER TABLE
+  //===--------------------------------------------------------------------===//
+  ResultType AlterTable(oid_t database_oid, oid_t table_oid, const std::string &schema_name,
+                        std::unique_ptr<catalog::Schema> &new_schema,
+                        concurrency::TransactionContext *txn);
+
+  ResultType AddColumn(const std::string &database_name,
+                       const std::string &table_name,
+                       const std::vector<std::string> &columns,
+                       concurrency::TransactionContext *txn);
+
+  ResultType DropColumn(const std::string &database_name,
+                        const std::string &table_name,
+                        const std::vector<std::string> &columns,
+                        concurrency::TransactionContext *txn);
+
+  ResultType RenameColumn(const std::string &database_name,
+                          const std::string &table_name,
+                          const std::string &old_name,
+                          const std::string &new_name,
+                          const std::string &schema_name,
+                          concurrency::TransactionContext *txn);
+
   /*
    * Using database oid to get system catalog object
    */
diff --git a/src/include/catalog/column.h b/src/include/catalog/column.h
index f351f72c5b9..9917c5e2829 100644
--- a/src/include/catalog/column.h
+++ b/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_; }
diff --git a/src/include/catalog/column_catalog.h b/src/include/catalog/column_catalog.h
index 7dcdf96e4a5..ffa5037c364 100644
--- a/src/include/catalog/column_catalog.h
+++ b/src/include/catalog/column_catalog.h
@@ -89,6 +89,9 @@ class ColumnCatalog : public AbstractCatalog {
   bool DeleteColumn(oid_t table_oid, const std::string &column_name,
                     concurrency::TransactionContext *txn);
   bool DeleteColumns(oid_t table_oid, concurrency::TransactionContext *txn);
+  bool RenameColumn(oid_t database_oid, oid_t table_oid, const std::string &column_name,
+                    const std::string &new_name,
+                    concurrency::TransactionContext *txn);
 
  private:
   //===--------------------------------------------------------------------===//
diff --git a/src/include/catalog/schema.h b/src/include/catalog/schema.h
index 8a2feec5dcd..6f401f5e0a2 100644
--- a/src/include/catalog/schema.h
+++ b/src/include/catalog/schema.h
@@ -144,6 +144,16 @@ class Schema : public Printable {
     return index;
   }
 
+  inline void ChangeColumnName(const oid_t column_id,
+                               const std::string &new_name) {
+    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];
   }
@@ -191,7 +201,7 @@ class Schema : public Printable {
   }
 
   // Get the default value for the column
-  inline type::Value* GetDefaultValue(const oid_t column_id) const {
+  inline type::Value *GetDefaultValue(const oid_t column_id) const {
     for (auto constraint : columns[column_id].GetConstraints()) {
       if (constraint.GetType() == ConstraintType::DEFAULT) {
         return constraint.getDefaultValue();
diff --git a/src/include/common/internal_types.h b/src/include/common/internal_types.h
index 17020512944..9ab5680ebe7 100644
--- a/src/include/common/internal_types.h
+++ b/src/include/common/internal_types.h
@@ -573,6 +573,8 @@ enum class PlanNodeType {
   CREATE = 34,
   POPULATE_INDEX = 35,
   ANALYZE = 36,
+  RENAME = 37,
+  ALTER = 38,
 
   // Communication Nodes
   SEND = 40,
@@ -638,6 +640,19 @@ std::string DropTypeToString(DropType type);
 DropType StringToDropType(const std::string &str);
 std::ostream &operator<<(std::ostream &os, const DropType &type);
 
+//===--------------------------------------------------------------------===//
+// Alter Types
+//===--------------------------------------------------------------------===//
+
+enum class AlterType {
+  INVALID = INVALID_TYPE_ID,  // invalid alter type
+  RENAME = 1,                 // rename table, column, database...
+  ALTER = 2
+};
+std::string AlterTypeToString(AlterType type);
+AlterType StringToAlterType(const std::string &str);
+std::ostream &operator<<(std::ostream &os, const AlterType &type);
+
 template <class E>
 class EnumHash {
  public:
diff --git a/src/include/common/logger.h b/src/include/common/logger.h
index 0e912ca21c2..72a4286251a 100644
--- a/src/include/common/logger.h
+++ b/src/include/common/logger.h
@@ -26,8 +26,8 @@
 
 // Fix for PRId64 (See https://stackoverflow.com/a/18719205)
 #if defined(__cplusplus) && !defined(__STDC_FORMAT_MACROS)
-#define __STDC_FORMAT_MACROS 1 // Not sure where to put this
-#endif 
+#define __STDC_FORMAT_MACROS 1  // Not sure where to put this
+#endif
 #include <inttypes.h>
 
 namespace peloton {
@@ -178,10 +178,8 @@ inline void outputLogHeader_(const char *file, int line, const char *func,
       type = "UNKWN";
   }
   // PAVLO: DO NOT CHANGE THIS
-  ::fprintf(LOG_OUTPUT_STREAM, "%s [%s:%d:%s] %s - ",
-            time_str,
-            file, line, func,
-            type);
+  ::fprintf(LOG_OUTPUT_STREAM, "%s [%s:%d:%s] %s - ", time_str, file, line,
+            func, type);
 }
 
 }  // namespace peloton
diff --git a/src/include/common/sql_node_visitor.h b/src/include/common/sql_node_visitor.h
index 96071410adb..2adcbb19696 100644
--- a/src/include/common/sql_node_visitor.h
+++ b/src/include/common/sql_node_visitor.h
@@ -21,6 +21,7 @@ class CreateFunctionStatement;
 class InsertStatement;
 class DeleteStatement;
 class DropStatement;
+class AlterTableStatement;
 class ExplainStatement;
 class PrepareStatement;
 class ExecuteStatement;
@@ -80,7 +81,8 @@ class SqlNodeVisitor {
   virtual void Visit(parser::TransactionStatement *) {}
   virtual void Visit(parser::UpdateStatement *) {}
   virtual void Visit(parser::CopyStatement *) {}
-  virtual void Visit(parser::AnalyzeStatement *){};
+  virtual void Visit(parser::AnalyzeStatement *) {}
+  virtual void Visit(parser::AlterTableStatement *) {}
   virtual void Visit(parser::ExplainStatement *){};
 
   virtual void Visit(expression::ComparisonExpression *expr);
diff --git a/src/include/concurrency/transaction_context.h b/src/include/concurrency/transaction_context.h
index 892149b510e..55a1841f78f 100644
--- a/src/include/concurrency/transaction_context.h
+++ b/src/include/concurrency/transaction_context.h
@@ -33,6 +33,10 @@ class TriggerSet;
 class TriggerData;
 }  // namespace trigger
 
+namespace storage {
+class DataTable;
+}
+
 namespace concurrency {
 
 //===--------------------------------------------------------------------===//
@@ -173,6 +177,14 @@ class TransactionContext : public Printable {
 
   void RecordInsert(const ItemPointer &);
 
+  void RecordDropTable(storage::DataTable *table) {
+    dropped_tables.push_back(table);
+  }
+
+  std::vector<storage::DataTable *> &GetDroppedTables() {
+    return dropped_tables;
+  }
+
   /**
    * @brief      Delete the record.
    *
@@ -341,6 +353,9 @@ class TransactionContext : public Printable {
   IsolationLevelType isolation_level_;
 
   std::unique_ptr<trigger::TriggerSet> on_commit_triggers_;
+
+  /** vector of dropped data tables **/
+  std::vector<storage::DataTable *> dropped_tables;
 };
 
 }  // namespace concurrency
diff --git a/src/include/executor/alter_executor.h b/src/include/executor/alter_executor.h
new file mode 100644
index 00000000000..b644d48456b
--- /dev/null
+++ b/src/include/executor/alter_executor.h
@@ -0,0 +1,53 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// alter_executor.h
+//
+// Identification: src/include/executor/alter_executor.h
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "concurrency/transaction_context.h"
+#include "executor/abstract_executor.h"
+#include "planner/alter_plan.h"
+
+namespace peloton {
+
+namespace planner {
+class AbstractPlan;
+}
+
+namespace executor {
+
+class AlterExecutor : public AbstractExecutor {
+ public:
+  AlterExecutor(const AlterExecutor &) = delete;
+  AlterExecutor &operator=(const AlterExecutor &) = delete;
+  AlterExecutor(AlterExecutor &&) = delete;
+  AlterExecutor &operator=(AlterExecutor &&) = delete;
+
+  AlterExecutor(const planner::AbstractPlan *node,
+                ExecutorContext *executor_context);
+
+  ~AlterExecutor() {}
+
+ protected:
+  bool DInit();
+
+  bool DExecute();
+
+  bool RenameColumn(const planner::AlterPlan &node,
+                    concurrency::TransactionContext *txn);
+
+  bool AlterTable(const planner::AlterPlan &node,
+                  concurrency::TransactionContext *txn);
+};
+
+}  // executor
+
+}  // namespace peloton
diff --git a/src/include/executor/executors.h b/src/include/executor/executors.h
index 16614120444..38a0fb606fa 100644
--- a/src/include/executor/executors.h
+++ b/src/include/executor/executors.h
@@ -16,6 +16,7 @@
 
 #include "executor/aggregate_executor.h"
 #include "executor/aggregator.h"
+#include "executor/alter_executor.h"
 #include "executor/analyze_executor.h"
 #include "executor/append_executor.h"
 #include "executor/copy_executor.h"
diff --git a/src/include/optimizer/query_node_visitor.h b/src/include/optimizer/query_node_visitor.h
index 49dc04ff54f..305be8e972a 100644
--- a/src/include/optimizer/query_node_visitor.h
+++ b/src/include/optimizer/query_node_visitor.h
@@ -29,6 +29,7 @@ class CopyStatement;
 class AnalyzeStatement;
 class VariableSetStatement;
 class JoinDefinition;
+class AlterTableStatement;
 struct TableRef;
 
 class GroupByDescription;
diff --git a/src/include/optimizer/query_to_operator_transformer.h b/src/include/optimizer/query_to_operator_transformer.h
index 2ba6bc8e117..103aade4922 100644
--- a/src/include/optimizer/query_to_operator_transformer.h
+++ b/src/include/optimizer/query_to_operator_transformer.h
@@ -63,6 +63,7 @@ class QueryToOperatorTransformer : public SqlNodeVisitor {
   void Visit(parser::UpdateStatement *op) override;
   void Visit(parser::CopyStatement *op) override;
   void Visit(parser::AnalyzeStatement *op) override;
+  void Visit(parser::AlterTableStatement *op) override;
   void Visit(expression::ComparisonExpression *expr) override;
   void Visit(expression::OperatorExpression *expr) override;
 
diff --git a/src/include/parser/alter_statement.h b/src/include/parser/alter_statement.h
new file mode 100644
index 00000000000..7262e6abb04
--- /dev/null
+++ b/src/include/parser/alter_statement.h
@@ -0,0 +1,77 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// alter_function_statement.h
+//
+// Identification: src/include/parser/alter_function_statement.h
+//
+// Copyright (c) 2015-17, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "parser/sql_statement.h"
+#include "common/sql_node_visitor.h"
+#include "parser/create_statement.h"
+
+namespace peloton {
+namespace parser {
+/**
+ * @class AlterTableStatement
+ * @brief Represents "ALTER TABLE add column COLUMN_NAME COLUMN_TYPE"
+ * TODO: add implementation of AlterTableStatement
+ */
+class AlterTableStatement : public TableRefStatement {
+ public:
+  enum class AlterTableType { INVALID = 0, ALTER = 1, RENAME = 2 };
+  AlterTableStatement(AlterTableType type)
+      : TableRefStatement(StatementType::ALTER),
+        type(type),
+        oldName(nullptr),
+        newName(nullptr) {
+    dropped_names =
+        type == AlterTableType::RENAME ? nullptr : (new std::vector<char *>);
+    added_columns = type == AlterTableType::RENAME
+                        ? nullptr
+                        : (new std::vector<std::unique_ptr<ColumnDefinition>>);
+    changed_type_columns =
+        type == AlterTableType::RENAME
+            ? nullptr
+            : (new std::vector<std::unique_ptr<ColumnDefinition>>);
+  }
+
+  virtual ~AlterTableStatement() {
+    if (added_columns != nullptr) {
+      delete added_columns;
+    }
+    if (dropped_names != nullptr) {
+      for (auto name : *dropped_names) delete name;
+      delete dropped_names;
+    }
+    if (changed_type_columns != nullptr) {
+      delete changed_type_columns;
+    }
+    if (oldName) delete oldName;
+    if (newName) delete newName;
+  }
+
+  virtual void Accept(SqlNodeVisitor *v) override { v->Visit(this); }
+
+  AlterTableType type;
+
+  // Dropped columns
+  std::vector<char *> *dropped_names;
+
+  // Added columns
+  std::vector<std::unique_ptr<ColumnDefinition>> *added_columns;
+
+  std::vector<std::unique_ptr<ColumnDefinition>> *changed_type_columns;
+  // the name that needs to be changed
+  char *oldName;
+  char *newName;
+};
+
+}  // End parser namespace
+}  // End peloton namespace
diff --git a/src/include/parser/parsenodes.h b/src/include/parser/parsenodes.h
index bf818ff6b86..7f8523a053a 100644
--- a/src/include/parser/parsenodes.h
+++ b/src/include/parser/parsenodes.h
@@ -73,7 +73,8 @@ typedef struct Expr { NodeTag type; } Expr;
 * of the lefthand expression (if any), and operName is the String name of
 * the combining operator.  Also, subselect is a raw parsetree.  During parse
 * analysis, the parser transforms testexpr into a complete boolean expression
-    * that compares the lefthand value(s) to PARAM_SUBLINK nodes representing the
+    * that compares the lefthand value(s) to PARAM_SUBLINK nodes representing
+*the
     * output columns of the subselect.  And subselect is transformed to a Query.
 * This is the representation seen in saved rules and in the rewriter.
 *
@@ -89,8 +90,7 @@ typedef struct Expr { NodeTag type; } Expr;
 * The CTE_SUBLINK case never occurs in actual SubLink nodes, but it is used
     * in SubPlans generated for WITH subqueries.
 */
-typedef enum SubLinkType
-{
+typedef enum SubLinkType {
   EXISTS_SUBLINK,
   ALL_SUBLINK,
   ANY_SUBLINK,
@@ -98,19 +98,17 @@ typedef enum SubLinkType
   EXPR_SUBLINK,
   MULTIEXPR_SUBLINK,
   ARRAY_SUBLINK,
-  CTE_SUBLINK					/* for SubPlans only */
+  CTE_SUBLINK /* for SubPlans only */
 } SubLinkType;
 
-
-typedef struct SubLink
-{
-  Expr		xpr;
-  SubLinkType subLinkType;	/* see above */
-  int			subLinkId;		/* ID (1..n); 0 if not MULTIEXPR */
-  Node	   *testexpr;		/* outer-query test for ALL/ANY/ROWCOMPARE */
-  List	   *operName;		/* originally specified operator name */
-  Node	   *subselect;		/* subselect as Query* or raw parsetree */
-  int			location;		/* token location, or -1 if unknown */
+typedef struct SubLink {
+  Expr xpr;
+  SubLinkType subLinkType; /* see above */
+  int subLinkId;           /* ID (1..n); 0 if not MULTIEXPR */
+  Node *testexpr;          /* outer-query test for ALL/ANY/ROWCOMPARE */
+  List *operName;          /* originally specified operator name */
+  Node *subselect;         /* subselect as Query* or raw parsetree */
+  int location;            /* token location, or -1 if unknown */
 } SubLink;
 
 typedef struct BoolExpr {
@@ -663,8 +661,8 @@ typedef struct DropStmt {
 
 typedef struct DropDatabaseStmt {
   NodeTag type;
-  char *dbname;          /* name of database to drop */
-  bool missing_ok;       /* skip error if object is missing? */
+  char *dbname;    /* name of database to drop */
+  bool missing_ok; /* skip error if object is missing? */
 } DropDatabaseStmt;
 
 typedef struct TruncateStmt {
@@ -674,6 +672,141 @@ typedef struct TruncateStmt {
   DropBehavior behavior; /* RESTRICT or CASCADE behavior */
 } TruncateStmt;
 
+/* ----------------------
+ *	Alter Table
+ * ----------------------
+ */
+typedef struct AlterTableStmt {
+  NodeTag type;
+  RangeVar *relation; /* table to work on */
+  List *cmds;         /* list of subcommands */
+  ObjectType relkind; /* type of object */
+  bool missing_ok;    /* skip error if table missing */
+} AlterTableStmt;
+
+typedef enum AlterTableType {
+  AT_AddColumn,                 /* add column */
+  AT_AddColumnRecurse,          /* internal to commands/tablecmds.c */
+  AT_AddColumnToView,           /* implicitly via CREATE OR REPLACE VIEW */
+  AT_ColumnDefault,             /* alter column default */
+  AT_DropNotNull,               /* alter column drop not null */
+  AT_SetNotNull,                /* alter column set not null */
+  AT_SetStatistics,             /* alter column set statistics */
+  AT_SetOptions,                /* alter column set ( options ) */
+  AT_ResetOptions,              /* alter column reset ( options ) */
+  AT_SetStorage,                /* alter column set storage */
+  AT_DropColumn,                /* drop column */
+  AT_DropColumnRecurse,         /* internal to commands/tablecmds.c */
+  AT_AddIndex,                  /* add index */
+  AT_ReAddIndex,                /* internal to commands/tablecmds.c */
+  AT_AddConstraint,             /* add constraint */
+  AT_AddConstraintRecurse,      /* internal to commands/tablecmds.c */
+  AT_ReAddConstraint,           /* internal to commands/tablecmds.c */
+  AT_AlterConstraint,           /* alter constraint */
+  AT_ValidateConstraint,        /* validate constraint */
+  AT_ValidateConstraintRecurse, /* internal to commands/tablecmds.c */
+  AT_ProcessedConstraint,       /* pre-processed add constraint (local in
+                     * parser/parse_utilcmd.c) */
+  AT_AddIndexConstraint,        /* add constraint using existing index */
+  AT_DropConstraint,            /* drop constraint */
+  AT_DropConstraintRecurse,     /* internal to commands/tablecmds.c */
+  AT_ReAddComment,              /* internal to commands/tablecmds.c */
+  AT_AlterColumnType,           /* alter column type */
+  AT_AlterColumnGenericOptions, /* alter column OPTIONS (...) */
+  AT_ChangeOwner,               /* change owner */
+  AT_ClusterOn,                 /* CLUSTER ON */
+  AT_DropCluster,               /* SET WITHOUT CLUSTER */
+  AT_SetLogged,                 /* SET LOGGED */
+  AT_SetUnLogged,               /* SET UNLOGGED */
+  AT_AddOids,                   /* SET WITH OIDS */
+  AT_AddOidsRecurse,            /* internal to commands/tablecmds.c */
+  AT_DropOids,                  /* SET WITHOUT OIDS */
+  AT_SetTableSpace,             /* SET TABLESPACE */
+  AT_SetRelOptions,             /* SET (...) -- AM specific parameters */
+  AT_ResetRelOptions,           /* RESET (...) -- AM specific parameters */
+  AT_ReplaceRelOptions,         /* replace reloption list in its entirety */
+  AT_EnableTrig,                /* ENABLE TRIGGER name */
+  AT_EnableAlwaysTrig,          /* ENABLE ALWAYS TRIGGER name */
+  AT_EnableReplicaTrig,         /* ENABLE REPLICA TRIGGER name */
+  AT_DisableTrig,               /* DISABLE TRIGGER name */
+  AT_EnableTrigAll,             /* ENABLE TRIGGER ALL */
+  AT_DisableTrigAll,            /* DISABLE TRIGGER ALL */
+  AT_EnableTrigUser,            /* ENABLE TRIGGER USER */
+  AT_DisableTrigUser,           /* DISABLE TRIGGER USER */
+  AT_EnableRule,                /* ENABLE RULE name */
+  AT_EnableAlwaysRule,          /* ENABLE ALWAYS RULE name */
+  AT_EnableReplicaRule,         /* ENABLE REPLICA RULE name */
+  AT_DisableRule,               /* DISABLE RULE name */
+  AT_AddInherit,                /* INHERIT parent */
+  AT_DropInherit,               /* NO INHERIT parent */
+  AT_AddOf,                     /* OF <type_name> */
+  AT_DropOf,                    /* NOT OF */
+  AT_ReplicaIdentity,           /* REPLICA IDENTITY */
+  AT_EnableRowSecurity,         /* ENABLE ROW SECURITY */
+  AT_DisableRowSecurity,        /* DISABLE ROW SECURITY */
+  AT_ForceRowSecurity,          /* FORCE ROW SECURITY */
+  AT_NoForceRowSecurity,        /* NO FORCE ROW SECURITY */
+  AT_GenericOptions             /* OPTIONS (...) */
+} AlterTableType;
+
+typedef struct AlterTableCmd /* one subcommand of an ALTER TABLE */
+    {
+  NodeTag type;
+  AlterTableType subtype; /* Type of table alteration to apply */
+  char *name;             /* column, constraint, or trigger to act on,
+                           * or tablespace */
+  Node *newowner;         /* RoleSpec */
+  Node *def;              /* definition of new column, index,
+                           * constraint, or parent table */
+  DropBehavior behavior;  /* RESTRICT or CASCADE for DROP cases */
+  bool missing_ok;        /* skip error if missing? */
+} AlterTableCmd;
+
+/* ----------------------
+ *		Alter Object Rename Statement
+ * ----------------------
+ */
+typedef struct RenameStmt {
+  NodeTag type;
+  ObjectType renameType;   /* OBJECT_TABLE, OBJECT_COLUMN, etc */
+  ObjectType relationType; /* if column name, associated relation type */
+  RangeVar *relation;      /* in case it's a table */
+  List *object;            /* in case it's some other object */
+  List *objarg;            /* argument types, if applicable */
+  char *subname;           /* name of contained object (column, rule,
+                    * trigger, etc) */
+  char *newname;           /* the new name */
+  DropBehavior behavior;   /* RESTRICT or CASCADE behavior */
+  bool missing_ok;         /* skip error if missing? */
+} RenameStmt;
+
+/* ----------------------
+ *		ALTER object SET SCHEMA Statement
+ * ----------------------
+ */
+typedef struct AlterObjectSchemaStmt {
+  NodeTag type;
+  ObjectType objectType; /* OBJECT_TABLE, OBJECT_TYPE, etc */
+  RangeVar *relation;    /* in case it's a table */
+  List *object;          /* in case it's some other object */
+  List *objarg;          /* argument types, if applicable */
+  char *newschema;       /* the new schema */
+  bool missing_ok;       /* skip error if missing? */
+} AlterObjectSchemaStmt;
+
+/* ----------------------
+ *		Alter Object Owner Statement
+ * ----------------------
+ */
+typedef struct AlterOwnerStmt {
+  NodeTag type;
+  ObjectType objectType; /* OBJECT_TABLE, OBJECT_TYPE, etc */
+  RangeVar *relation;    /* in case it's a table */
+  List *object;          /* in case it's some other object */
+  List *objarg;          /* argument types, if applicable */
+  Node *newowner;        /* the new owner */
+} AlterOwnerStmt;
+
 typedef struct ExecuteStmt {
   NodeTag type;
   char *name;   /* The name of the plan to execute */
@@ -721,47 +854,42 @@ typedef struct CreateDatabaseStmt {
   List *options; /* List of DefElem nodes */
 } CreateDatabaseStmt;
 
-typedef struct CreateSchemaStmt
-{
-  NodeTag   type;
-  char     *schemaname;   /* the name of the schema to create */
-  Node     *authrole;   /* the owner of the created schema */
-  List     *schemaElts;   /* schema components (list of parsenodes) */
-  bool    if_not_exists;  /* just do nothing if schema already exists? */
+typedef struct CreateSchemaStmt {
+  NodeTag type;
+  char *schemaname;   /* the name of the schema to create */
+  Node *authrole;     /* the owner of the created schema */
+  List *schemaElts;   /* schema components (list of parsenodes) */
+  bool if_not_exists; /* just do nothing if schema already exists? */
 } CreateSchemaStmt;
 
-typedef enum RoleSpecType
-{
-  ROLESPEC_CSTRING,     /* role name is stored as a C string */
-  ROLESPEC_CURRENT_USER,    /* role spec is CURRENT_USER */
-  ROLESPEC_SESSION_USER,    /* role spec is SESSION_USER */
-  ROLESPEC_PUBLIC       /* role name is "public" */
+typedef enum RoleSpecType {
+  ROLESPEC_CSTRING,      /* role name is stored as a C string */
+  ROLESPEC_CURRENT_USER, /* role spec is CURRENT_USER */
+  ROLESPEC_SESSION_USER, /* role spec is SESSION_USER */
+  ROLESPEC_PUBLIC        /* role name is "public" */
 } RoleSpecType;
 
-typedef struct RoleSpec
-{
-  NodeTag   type;
-  RoleSpecType roletype;    /* Type of this rolespec */
-  char     *rolename;   /* filled only for ROLESPEC_CSTRING */
-  int     location;   /* token location, or -1 if unknown */
+typedef struct RoleSpec {
+  NodeTag type;
+  RoleSpecType roletype; /* Type of this rolespec */
+  char *rolename;        /* filled only for ROLESPEC_CSTRING */
+  int location;          /* token location, or -1 if unknown */
 } RoleSpec;
 
-typedef enum ViewCheckOption
-{
+typedef enum ViewCheckOption {
   NO_CHECK_OPTION,
   LOCAL_CHECK_OPTION,
   CASCADED_CHECK_OPTION
 } ViewCheckOption;
 
-typedef struct ViewStmt
-{
-  NodeTag   type;
-  RangeVar   *view;     /* the view to be created */
-  List     *aliases;    /* target column names */
-  Node     *query;      /* the SELECT query */
-  bool    replace;    /* replace an existing view? */
-  List     *options;    /* options from WITH clause */
-  ViewCheckOption withCheckOption;  /* WITH CHECK OPTION */
+typedef struct ViewStmt {
+  NodeTag type;
+  RangeVar *view;                  /* the view to be created */
+  List *aliases;                   /* target column names */
+  Node *query;                     /* the SELECT query */
+  bool replace;                    /* replace an existing view? */
+  List *options;                   /* options from WITH clause */
+  ViewCheckOption withCheckOption; /* WITH CHECK OPTION */
 } ViewStmt;
 
 typedef struct ParamRef {
@@ -787,29 +915,26 @@ typedef struct VacuumStmt {
   List *va_cols;      /* list of column names, or NIL for all */
 } VacuumStmt;
 
-typedef enum
-{
-  VAR_SET_VALUE,              /* SET var = value */
-  VAR_SET_DEFAULT,            /* SET var TO DEFAULT */
-  VAR_SET_CURRENT,            /* SET var FROM CURRENT */
-  VAR_SET_MULTI,              /* special case for SET TRANSACTION ... */
-  VAR_RESET,                  /* RESET var */
-  VAR_RESET_ALL               /* RESET ALL */
+typedef enum {
+  VAR_SET_VALUE,   /* SET var = value */
+  VAR_SET_DEFAULT, /* SET var TO DEFAULT */
+  VAR_SET_CURRENT, /* SET var FROM CURRENT */
+  VAR_SET_MULTI,   /* special case for SET TRANSACTION ... */
+  VAR_RESET,       /* RESET var */
+  VAR_RESET_ALL    /* RESET ALL */
 } VariableSetKind;
 
-typedef struct VariableSetStmt
-{
-  NodeTag     type;
+typedef struct VariableSetStmt {
+  NodeTag type;
   VariableSetKind kind;
-  char       *name;           /* variable to be set */
-  List       *args;           /* List of A_Const nodes */
-  bool        is_local;       /* SET LOCAL? */
+  char *name;    /* variable to be set */
+  List *args;    /* List of A_Const nodes */
+  bool is_local; /* SET LOCAL? */
 } VariableSetStmt;
 
-typedef struct VariableShowStmt
-{
-  NodeTag     type;
-  char       *name;
+typedef struct VariableShowStmt {
+  NodeTag type;
+  char *name;
 } VariableShowStmt;
 
 /// **********  For UDFs *********** ///
diff --git a/src/include/parser/postgresparser.h b/src/include/parser/postgresparser.h
index decd43d9ee7..87345629cbd 100644
--- a/src/include/parser/postgresparser.h
+++ b/src/include/parser/postgresparser.h
@@ -118,8 +118,8 @@ class PostgresParser {
   static parser::TableRef *FromTransform(SelectStmt *root);
 
   // transform helper for select targets
-  static std::vector<std::unique_ptr<expression::AbstractExpression>>
-      *TargetTransform(List *root);
+  static std::vector<std::unique_ptr<expression::AbstractExpression>> *
+  TargetTransform(List *root);
 
   // transform helper for all expr nodes
   static expression::AbstractExpression *ExprTransform(Node *root);
@@ -167,7 +167,8 @@ class PostgresParser {
   static parser::OrderDescription *OrderByTransform(List *order);
 
   // transform helper for table column definitions
-  static void ColumnDefTransform(ColumnDef* root, parser::CreateStatement* stmt);
+  static void ColumnDefTransform(ColumnDef *root,
+                                 parser::CreateStatement *stmt);
 
   // transform helper for create statements
   static parser::SQLStatement *CreateTransform(CreateStmt *root);
@@ -195,7 +196,8 @@ class PostgresParser {
    * @param Postgres CreateDatabaseStmt parsenode
    * @return a peloton CreateStatement node
    */
-  static parser::SQLStatement *CreateDatabaseTransform(CreateDatabaseStmt *root);
+  static parser::SQLStatement *CreateDatabaseTransform(
+      CreateDatabaseStmt *root);
 
   // transform helper for create schema statements
   static parser::SQLStatement *CreateSchemaTransform(CreateSchemaStmt *root);
@@ -208,8 +210,8 @@ class PostgresParser {
 
   // transform helper for ListsTransform (insert multiple rows)
   static std::vector<
-      std::vector<std::unique_ptr<expression::AbstractExpression>>>
-      *ValueListsTransform(List *root);
+      std::vector<std::unique_ptr<expression::AbstractExpression>>> *
+  ValueListsTransform(List *root);
 
   // transform helper for insert statements
   static parser::SQLStatement *InsertTransform(InsertStmt *root);
@@ -233,8 +235,8 @@ class PostgresParser {
   static parser::UpdateStatement *UpdateTransform(UpdateStmt *update_stmt);
 
   // transform helper for update statement
-  static std::vector<std::unique_ptr<parser::UpdateClause>>
-      *UpdateTargetTransform(List *root);
+  static std::vector<std::unique_ptr<parser::UpdateClause>> *
+  UpdateTargetTransform(List *root);
 
   // transform helper for drop statement
   static parser::DropStatement *DropTransform(DropStmt *root);
@@ -282,13 +284,17 @@ class PostgresParser {
   static parser::CopyStatement *CopyTransform(CopyStmt *root);
 
   // transform helper for analyze statement
-  static parser::AnalyzeStatement *VacuumTransform(VacuumStmt* root);
+  static parser::AnalyzeStatement *VacuumTransform(VacuumStmt *root);
 
-  static parser::VariableSetStatement *VariableSetTransform(VariableSetStmt* root);
+  static parser::VariableSetStatement *VariableSetTransform(
+      VariableSetStmt *root);
 
   // transform helper for subquery expressions
   static expression::AbstractExpression *SubqueryExprTransform(SubLink *node);
 
+  // transform helper for alter table statement
+  static parser::AlterTableStatement *AlterTransform(Node *root);
+
 };
 
 }  // namespace parser
diff --git a/src/include/parser/statements.h b/src/include/parser/statements.h
index 628b708a4c0..3b062c02748 100644
--- a/src/include/parser/statements.h
+++ b/src/include/parser/statements.h
@@ -15,6 +15,7 @@
 // This is just for convenience
 
 #include "analyze_statement.h"
+#include "alter_statement.h"
 #include "copy_statement.h"
 #include "create_function_statement.h"
 #include "create_statement.h"
diff --git a/src/include/planner/alter_plan.h b/src/include/planner/alter_plan.h
new file mode 100644
index 00000000000..0e0c5cfe6d4
--- /dev/null
+++ b/src/include/planner/alter_plan.h
@@ -0,0 +1,131 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// alter_plan.h
+//
+// Identification: src/include/parser/alter_plan.h
+//
+// Copyright (c) 2015-17, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+#include "planner/abstract_plan.h"
+#include "parser/alter_statement.h"
+namespace peloton {
+
+namespace parser {
+class AlterTableStatement;
+}
+namespace catalog {
+class Schema;
+}
+namespace storage {
+class DataTable;
+}
+
+namespace planner {
+/** @brief The plan used for altering
+ *  TODO: adding support for add/drop column
+ */
+class AlterPlan : public AbstractPlan {
+ public:
+  AlterPlan() = delete;
+
+  explicit AlterPlan(
+      const std::string &database_name, const std::string &table_name,
+      const std::vector<std::string> &dropped_columns,
+      const std::unique_ptr<catalog::Schema> &added_columns,
+      const std::unique_ptr<catalog::Schema> &changed_type_columns,
+      AlterType a_type);
+  explicit AlterPlan(const std::string &database_name,
+                     const std::string &table_name,
+                     const std::vector<std::string> &old_names,
+                     const std::vector<std::string> &new_names,
+                     AlterType a_type);
+  explicit AlterPlan(parser::AlterTableStatement *parse_tree);
+
+  virtual ~AlterPlan() {}
+
+  virtual PlanNodeType GetPlanNodeType() const { return PlanNodeType::ALTER; }
+
+  virtual std::unique_ptr<AbstractPlan> Copy() { return nullptr; }
+
+  const std::string GetInfo() const {
+    return StringUtil::Format("AlterPlan table:%s, database:%s",
+                              this->table_name.c_str(),
+                              this->database_name.c_str());
+  }
+
+  std::unique_ptr<AbstractPlan> Copy() const {
+    switch (this->type) {
+      case AlterType::ALTER:
+        return std::unique_ptr<AbstractPlan>(new AlterPlan(
+            database_name, table_name, dropped_columns, added_columns, changed_type_columns, type));
+      case AlterType::RENAME:
+        return std::unique_ptr<AbstractPlan>(new AlterPlan(
+            database_name, table_name, old_names_, new_names_, type));
+      default:
+        LOG_ERROR("Not supported Copy of Alter type yet");
+        return nullptr;
+    }
+  }
+
+  std::string GetTableName() const { return table_name; }
+
+  std::string GetDatabaseName() const { return database_name; }
+
+  const std::unique_ptr<catalog::Schema> &GetAddedColumns() const {
+    return added_columns;
+  }
+
+  const std::vector<std::string> &GetDroppedColumns() const {
+    return dropped_columns;
+  }
+
+  const std::unique_ptr<catalog::Schema> &
+  GetChangedTypeColumns() const {
+    return changed_type_columns;
+  };
+
+  AlterType GetAlterTableType() const { return type; }
+
+  // function used for rename statement
+  std::string GetOldName() const { return this->old_names_[0]; }
+
+  // function used for rename statement
+  std::string GetNewName() const { return this->new_names_[0]; }
+
+  // return true if the alter plan is rename statement
+  bool IsRename() const { return this->type == AlterType::RENAME; }
+
+  // return schema name
+  std::string GetSchemaName() const { return this->schema_name; }
+ private:
+  // Target Table
+  storage::DataTable *target_table_ = nullptr;
+
+  // Table Name
+  std::string table_name;
+
+  // Database Name
+  std::string database_name;
+
+  // Schema Name
+  std::string schema_name;
+  // Schema delta, define the column txn want to add
+  std::unique_ptr<catalog::Schema> added_columns;
+  // dropped_column, define the column you want to drop
+  std::vector<std::string> dropped_columns;
+  // changed-type columns, define the column you want to change type
+  std::unique_ptr<catalog::Schema> changed_type_columns;
+  // used for store rename function data
+  std::vector<std::string> old_names_;
+  std::vector<std::string> new_names_;
+
+  // Check to either AlterTable Table, INDEX or Rename
+  AlterType type;
+};
+}
+}
diff --git a/src/include/storage/database.h b/src/include/storage/database.h
index 630b3729bcf..73c9a43face 100644
--- a/src/include/storage/database.h
+++ b/src/include/storage/database.h
@@ -58,6 +58,9 @@ class Database : public Printable {
 
   void DropTableWithOid(const oid_t table_oid);
 
+  storage::DataTable *ReplaceTableWithOid(const oid_t table_oid,
+                                          storage::DataTable *new_table);
+
   //===--------------------------------------------------------------------===//
   // UTILITIES
   //===--------------------------------------------------------------------===//
diff --git a/src/include/storage/tile.h b/src/include/storage/tile.h
index edb210f868c..adff52f7739 100644
--- a/src/include/storage/tile.h
+++ b/src/include/storage/tile.h
@@ -139,6 +139,11 @@ class Tile : public Printable {
     return schema.GetColumn(column_index).GetName();
   }
 
+  inline void ChangeColumnName(const oid_t column_index,
+                               const std::string &name) {
+    schema.ChangeColumnName(column_index, name);
+  }
+
   inline oid_t GetColumnCount() const { return column_count; };
 
   inline TileGroupHeader *GetHeader() const { return tile_group_header; }
diff --git a/src/optimizer/optimizer.cpp b/src/optimizer/optimizer.cpp
index 62f813ec876..d0a7ec024f7 100644
--- a/src/optimizer/optimizer.cpp
+++ b/src/optimizer/optimizer.cpp
@@ -41,6 +41,7 @@
 #include "planner/populate_index_plan.h"
 #include "planner/projection_plan.h"
 #include "planner/seq_scan_plan.h"
+#include "planner/alter_plan.h"
 
 #include "storage/data_table.h"
 
@@ -145,6 +146,14 @@ unique_ptr<planner::AbstractPlan> Optimizer::HandleDDLStatement(
   is_ddl_stmt = true;
   auto stmt_type = tree->GetType();
   switch (stmt_type) {
+    case StatementType::ALTER: {
+      // TODO (shilun) adding support of Alter
+      LOG_TRACE("Adding Alter Plan");
+      unique_ptr<planner::AbstractPlan> alter_plan(
+          new planner::AlterPlan((parser::AlterTableStatement *)tree));
+      ddl_plan = move(alter_plan);
+      break;
+    }
     case StatementType::DROP: {
       LOG_TRACE("Adding Drop plan...");
       unique_ptr<planner::AbstractPlan> drop_plan(
diff --git a/src/optimizer/query_to_operator_transformer.cpp b/src/optimizer/query_to_operator_transformer.cpp
index ff75140d5f5..de9aa73a632 100644
--- a/src/optimizer/query_to_operator_transformer.cpp
+++ b/src/optimizer/query_to_operator_transformer.cpp
@@ -335,6 +335,8 @@ void QueryToOperatorTransformer::Visit(
     UNUSED_ATTRIBUTE parser::ExecuteStatement *op) {}
 void QueryToOperatorTransformer::Visit(
     UNUSED_ATTRIBUTE parser::TransactionStatement *op) {}
+void QueryToOperatorTransformer::Visit(
+    UNUSED_ATTRIBUTE parser::AlterTableStatement *op) {}
 void QueryToOperatorTransformer::Visit(parser::UpdateStatement *op) {
   auto target_table = catalog::Catalog::GetInstance()->GetTableObject(
       op->table->GetDatabaseName(), op->table->GetSchemaName(),
diff --git a/src/parser/postgresparser.cpp b/src/parser/postgresparser.cpp
index 797b77406b5..668b96b5701 100644
--- a/src/parser/postgresparser.cpp
+++ b/src/parser/postgresparser.cpp
@@ -1776,6 +1776,109 @@ parser::TransactionStatement *PostgresParser::TransactionTransform(
   }
 }
 
+parser::AlterTableStatement *PostgresParser::AlterTransform(Node *root) {
+  switch (root->type) {
+    case T_RenameStmt: {
+      RenameStmt *newRoot = reinterpret_cast<RenameStmt *>(root);
+      if (newRoot->renameType != ObjectType::OBJECT_COLUMN) {
+        throw NotImplementedException(StringUtil::Format(
+            "Rename type %d not supported yes...\n", newRoot->relationType));
+      }
+      parser::AlterTableStatement *result = new parser::AlterTableStatement(
+          parser::AlterTableStatement::AlterTableType::RENAME);
+      RangeVar *relation = newRoot->relation;
+      result->table_info_ =
+          std::unique_ptr<parser::TableInfo>(new parser::TableInfo());
+      if (relation->relname) {
+        result->table_info_.get()->table_name = strdup(relation->relname);
+      }
+      if (relation->catalogname) {
+        result->table_info_.get()->database_name =
+            strdup(relation->catalogname);
+      }
+      if (relation->schemaname) {
+        result->table_info_.get()->schema_name = strdup(relation->schemaname);
+      }
+      if (newRoot->subname) {
+        result->oldName = strdup(newRoot->subname);
+      }
+      if (newRoot->newname) {
+        result->newName = strdup(newRoot->newname);
+      }
+      LOG_TRACE("finished transform");
+      return result;
+    }
+    case T_AlterTableStmt: {
+      // TODO (shilun) adding alter type check
+      // Currently we only support add/drop column type
+      AlterTableStmt *newRoot = reinterpret_cast<AlterTableStmt *>(root);
+      parser::AlterTableStatement *result =
+          new AlterTableStatement(AlterTableStatement::AlterTableType::ALTER);
+
+      // Get database and table name
+      RangeVar *relation = newRoot->relation;
+      result->table_info_ =
+          std::unique_ptr<parser::TableInfo>(new parser::TableInfo());
+      if (relation->relname) {
+        result->table_info_.get()->table_name = strdup(relation->relname);
+      }
+      if (relation->catalogname) {
+        result->table_info_.get()->database_name =
+            strdup(relation->catalogname);
+      }
+      if (relation->schemaname) {
+        result->table_info_.get()->schema_name = strdup(relation->schemaname);
+      }
+
+      for (auto cell = newRoot->cmds->head; cell != NULL; cell = cell->next) {
+        auto cmd = reinterpret_cast<AlterTableCmd *>(cell->data.ptr_value);
+        switch (cmd->subtype) {
+          case AT_AddColumn: {
+            // as the ColumnDefTransform only accepts CreateStatement
+            // we have to create one here
+            parser::CreateStatement tmpStatement(
+                parser::CreateStatement::CreateType::kTable);
+            ColumnDefTransform(reinterpret_cast<ColumnDef *>(cmd->def),
+                               &tmpStatement);
+            for (size_t i = 0; i < tmpStatement.columns.size(); i++) {
+              result->added_columns->emplace_back(
+                  std::move(tmpStatement.columns[i]));
+            }
+            break;
+          }
+          case AT_DropColumn: {
+            result->dropped_names->push_back(strdup(cmd->name));
+            break;
+          }
+          case AT_AlterColumnType: {
+            ColumnDef *def = (ColumnDef *)cmd->def;
+            def->colname = cmd->name;
+            parser::CreateStatement tmp_statement(
+                parser::CreateStatement::CreateType::kTable);
+            LOG_TRACE("Adding change type column");
+            ColumnDefTransform(reinterpret_cast<ColumnDef *>(def),
+                               &tmp_statement);
+            for (size_t i = 0; i < tmp_statement.columns.size();i++){
+              result->changed_type_columns->emplace_back(std::move(tmp_statement.columns[i]));
+            }
+            LOG_TRACE("adding end");
+            break;
+          }
+          default: {
+            throw NotImplementedException(StringUtil::Format(
+                "Alter Table type %d not supported yet...\n", cmd->subtype));
+          }
+        }
+      }
+      return result;
+    }
+    default:
+      LOG_ERROR("Not supported Alter Node type yet");
+      throw NotImplementedException(StringUtil::Format(
+          "Alter Table type %d not supported yet...", root->type));
+  }
+}
+
 // This function transfers a single Postgres statement into
 // a Peloton SQLStatement object. It checks the type of
 // Postgres parsenode of the input and call the corresponding
@@ -1783,6 +1886,11 @@ parser::TransactionStatement *PostgresParser::TransactionTransform(
 parser::SQLStatement *PostgresParser::NodeTransform(Node *stmt) {
   parser::SQLStatement *result = nullptr;
   switch (stmt->type) {
+    case T_RenameStmt:  // also use alter transform to transform the node
+    case T_AlterTableStmt:
+      // TODO (Shilun): adding T_ALTER_TABLE_STMT
+      result = AlterTransform(stmt);
+      break;
     case T_SelectStmt:
       result = SelectTransform(reinterpret_cast<SelectStmt *>(stmt));
       break;
diff --git a/src/planner/alter_plan.cpp b/src/planner/alter_plan.cpp
new file mode 100644
index 00000000000..62d7dd4e5cc
--- /dev/null
+++ b/src/planner/alter_plan.cpp
@@ -0,0 +1,119 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// alter_table_plan.cpp
+//
+// Identification: src/planner/alter_table_plan.cpp
+//
+// Copyright (c) 2015-16, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#include "planner/alter_plan.h"
+
+#include "catalog/column.h"
+#include "catalog/schema.h"
+#include "parser/alter_statement.h"
+#include "storage/data_table.h"
+
+namespace peloton {
+namespace planner {
+
+AlterPlan::AlterPlan(
+    const std::string &database_name, const std::string &table_name,
+    const std::vector<std::string> &dropped_columns,
+    const std::unique_ptr<catalog::Schema> &added_columns,
+    const std::unique_ptr<catalog::Schema> &changed_type_columns,
+    AlterType a_type)
+    : table_name(table_name),
+      database_name(database_name),
+      dropped_columns(dropped_columns),
+      type(a_type) {
+  this->added_columns =
+      std::unique_ptr<catalog::Schema>(new catalog::Schema(*added_columns));
+  this->changed_type_columns = std::unique_ptr<catalog::Schema>(
+      new catalog::Schema(*changed_type_columns));
+}
+
+AlterPlan::AlterPlan(const std::string &database_name,
+                     const std::string &table_name,
+                     const std::vector<std::string> &old_names,
+                     const std::vector<std::string> &new_names,
+                     const AlterType a_type)
+    : table_name(table_name),
+      database_name(database_name),
+      old_names_(old_names),
+      new_names_(new_names),
+      type(a_type) {}
+
+AlterPlan::AlterPlan(parser::AlterTableStatement *parse_tree) {
+  table_name = std::string(parse_tree->GetTableName());
+  database_name = std::string(parse_tree->GetDatabaseName());
+  schema_name = std::string(parse_tree->GetSchemaName());
+  switch (parse_tree->type) {
+    case parser::AlterTableStatement::AlterTableType::RENAME: {
+      old_names_.emplace_back(std::string{parse_tree->oldName});
+      new_names_.emplace_back(std::string{parse_tree->newName});
+      type = AlterType::RENAME;
+      break;
+    }
+    case parser::AlterTableStatement::AlterTableType::ALTER: {
+      // deal with dropped columns
+      type = AlterType::ALTER;
+      for (auto col : *parse_tree->dropped_names) {
+        LOG_TRACE("Drooped column name: %s", col);
+        dropped_columns.push_back(std::string(col));
+      }
+
+      // deal with added columns
+      std::vector<catalog::Column> columns;
+      for (size_t i = 0; i < (*parse_tree->added_columns).size(); i++) {
+        type::TypeId val = parser::ColumnDefinition::GetValueType(
+            (*parse_tree->added_columns)[i].get()->type);
+        // LOG_TRACE("Column Name: %s", (char
+        // *)((*parse_tree->columns)[i].get()->name.c_str()));
+
+        bool is_inline = (val == type::TypeId::VARCHAR) ? false : true;
+        auto column = catalog::Column(
+            val, type::Type::GetTypeSize(val),
+            std::string((*parse_tree->added_columns)[i].get()->name),
+            is_inline);
+
+        if ((*parse_tree->added_columns)[i].get()->not_null) {
+          catalog::Constraint constraint(ConstraintType::NOTNULL,
+                                         "con_not_null");
+          column.AddConstraint(constraint);
+        }
+        columns.emplace_back(column);
+      }
+      added_columns =
+          std::unique_ptr<catalog::Schema>(new catalog::Schema(columns));
+      columns.clear();
+      // deal with change column types
+      for (size_t i = 0; i < (*parse_tree->changed_type_columns).size(); i++) {
+        auto &tmp = (*parse_tree->changed_type_columns)[i];
+        type::TypeId val =
+            parser::ColumnDefinition::GetValueType(tmp.get()->type);
+        auto column = catalog::Column(
+            val, type::Type::GetTypeSize(val),
+            std::string((*parse_tree->changed_type_columns)[i].get()->name));
+        if ((*parse_tree->changed_type_columns)[i].get()->not_null) {
+          catalog::Constraint constraint(ConstraintType::NOTNULL,
+                                         "con_not_null");
+          column.AddConstraint(constraint);
+        }
+        columns.emplace_back(column);
+      }
+      changed_type_columns =
+          std::unique_ptr<catalog::Schema>(new catalog::Schema(columns));
+      break;
+    }
+    default:
+      LOG_ERROR("Not Implemented the plan yet!");
+      type = AlterType::INVALID;
+  }
+}
+
+}  // namespace planner
+}  // namespace peloton
diff --git a/src/storage/database.cpp b/src/storage/database.cpp
index 8a7506805c8..66b6dab99a8 100644
--- a/src/storage/database.cpp
+++ b/src/storage/database.cpp
@@ -94,6 +94,33 @@ void Database::DropTableWithOid(const oid_t table_oid) {
   }
 }
 
+/**
+ * Replace the the data table with oid.
+ * @param table_oid the oid of that table
+ * @param new_table the new table storage
+ * @return pointer to that table.
+ */
+storage::DataTable *Database::ReplaceTableWithOid(
+    const oid_t table_oid, storage::DataTable *new_table) {
+  {
+    std::lock_guard<std::mutex> lock(database_mutex);
+    codegen::QueryCache::Instance().Remove(table_oid);
+
+    oid_t table_offset = 0;
+    for (auto table : tables) {
+      if (table->GetOid() == table_oid) {
+        break;
+      }
+      table_offset++;
+    }
+    PELOTON_ASSERT(table_offset < tables.size());
+
+    auto old_table = tables.at(table_offset);
+    tables[table_offset] = new_table;
+    return old_table;
+  }
+}
+
 storage::DataTable *Database::GetTable(const oid_t table_offset) const {
   PELOTON_ASSERT(table_offset < tables.size());
   auto table = tables.at(table_offset);