diff --git a/src/observer/common/type/attr_type.cpp b/src/observer/common/type/attr_type.cpp
index e8af0084a..8410a3c79 100644
--- a/src/observer/common/type/attr_type.cpp
+++ b/src/observer/common/type/attr_type.cpp
@@ -12,7 +12,7 @@ See the Mulan PSL v2 for more details. */
 #include "common/lang/string.h"
 #include "common/type/attr_type.h"
 
-const char *ATTR_TYPE_NAME[] = {"undefined", "chars", "ints", "floats", "vectors", "booleans"};
+const char *ATTR_TYPE_NAME[] = {"undefined", "chars", "ints", "floats", "vectors", "booleans", "dates"};
 
 const char *attr_type_to_string(AttrType type)
 {
@@ -40,4 +40,4 @@ bool is_numerical_type(AttrType type)
 bool is_string_type(AttrType type)
 {
   return (type == AttrType::CHARS);
-}
\ No newline at end of file
+}
diff --git a/src/observer/common/type/attr_type.h b/src/observer/common/type/attr_type.h
index d1c07e3a2..d31b7e9f5 100644
--- a/src/observer/common/type/attr_type.h
+++ b/src/observer/common/type/attr_type.h
@@ -22,6 +22,7 @@ enum class AttrType
   FLOATS,    ///< 浮点数类型(4字节)
   VECTORS,   ///< 向量类型
   BOOLEANS,  ///< boolean类型，当前不是由parser解析出来的，是程序内部使用的
+  DATES,     ///< 日期类型(4字节，按YYYYMMDD编码)
   MAXTYPE,   ///< 请在 UNDEFINED 与 MAXTYPE 之间增加新类型
 };
 
diff --git a/src/observer/common/type/char_type.cpp b/src/observer/common/type/char_type.cpp
index 6c93114a9..ece9a7ba5 100644
--- a/src/observer/common/type/char_type.cpp
+++ b/src/observer/common/type/char_type.cpp
@@ -29,6 +29,8 @@ RC CharType::set_value_from_str(Value &val, const string &data) const
 RC CharType::cast_to(const Value &val, AttrType type, Value &result) const
 {
   switch (type) {
+    case AttrType::DATES:
+      return DataType::type_instance(AttrType::DATES)->set_value_from_str(result, val.get_string());
     default: return RC::UNIMPLEMENTED;
   }
   return RC::SUCCESS;
@@ -38,6 +40,8 @@ int CharType::cast_cost(AttrType type)
 {
   if (type == AttrType::CHARS) {
     return 0;
+  } else if (type == AttrType::DATES) {
+    return 1;
   }
   return INT32_MAX;
 }
@@ -48,4 +52,4 @@ RC CharType::to_string(const Value &val, string &result) const
   ss << val.value_.pointer_value_;
   result = ss.str();
   return RC::SUCCESS;
-}
\ No newline at end of file
+}
diff --git a/src/observer/common/type/data_type.cpp b/src/observer/common/type/data_type.cpp
index 6df3fa6ac..248c52bd6 100644
--- a/src/observer/common/type/data_type.cpp
+++ b/src/observer/common/type/data_type.cpp
@@ -9,6 +9,7 @@ MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 See the Mulan PSL v2 for more details. */
 
 #include "common/type/char_type.h"
+#include "common/type/date_type.h"
 #include "common/type/float_type.h"
 #include "common/type/integer_type.h"
 #include "common/type/data_type.h"
@@ -24,4 +25,5 @@ array<unique_ptr<DataType>, static_cast<int>(AttrType::MAXTYPE)> DataType::type_
     make_unique<FloatType>(),
     make_unique<VectorType>(),
     make_unique<DataType>(AttrType::BOOLEANS),
-};
\ No newline at end of file
+    make_unique<DateType>(),
+};
diff --git a/src/observer/common/type/date_type.cpp b/src/observer/common/type/date_type.cpp
new file mode 100644
index 000000000..facfb2c22
--- /dev/null
+++ b/src/observer/common/type/date_type.cpp
@@ -0,0 +1,97 @@
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+#include "common/type/date_type.h"
+
+#include <cstdio>
+#include <algorithm>
+#include <cctype>
+
+#include "common/lang/comparator.h"
+#include "common/value.h"
+#include "storage/common/column.h"
+
+RC DateType::parse_date(const string &text, int &encoded_date)
+{
+  const size_t first_dash = text.find('-');
+  const size_t second_dash = first_dash == string::npos ? string::npos : text.find('-', first_dash + 1);
+  if (first_dash != 4 || second_dash == string::npos || text.find('-', second_dash + 1) != string::npos ||
+      second_dash - first_dash < 2 || second_dash - first_dash > 3 ||
+      text.size() - second_dash < 2 || text.size() - second_dash > 3) {
+    return RC::SCHEMA_FIELD_TYPE_MISMATCH;
+  }
+
+  const auto all_digits = [](const string &part) {
+    return !part.empty() && std::all_of(part.begin(), part.end(), [](unsigned char ch) { return std::isdigit(ch); });
+  };
+  const string year_text = text.substr(0, first_dash);
+  const string month_text = text.substr(first_dash + 1, second_dash - first_dash - 1);
+  const string day_text = text.substr(second_dash + 1);
+  if (!all_digits(year_text) || !all_digits(month_text) || !all_digits(day_text)) {
+    return RC::SCHEMA_FIELD_TYPE_MISMATCH;
+  }
+
+  const int year = stoi(year_text);
+  const int month = stoi(month_text);
+  const int day = stoi(day_text);
+  if (year < 1 || month < 1 || month > 12) {
+    return RC::SCHEMA_FIELD_TYPE_MISMATCH;
+  }
+
+  static constexpr int DAYS_PER_MONTH[] = {31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31};
+  int max_day = DAYS_PER_MONTH[month - 1];
+  const bool leap_year = (year % 4 == 0 && year % 100 != 0) || year % 400 == 0;
+  if (month == 2 && leap_year) {
+    max_day = 29;
+  }
+  if (day < 1 || day > max_day) {
+    return RC::SCHEMA_FIELD_TYPE_MISMATCH;
+  }
+
+  encoded_date = year * 10000 + month * 100 + day;
+  return RC::SUCCESS;
+}
+
+int DateType::compare(const Value &left, const Value &right) const
+{
+  ASSERT(left.attr_type() == AttrType::DATES && right.attr_type() == AttrType::DATES, "invalid date type");
+  int left_date = left.get_int();
+  int right_date = right.get_int();
+  return common::compare_int(&left_date, &right_date);
+}
+
+int DateType::compare(const Column &left, const Column &right, int left_idx, int right_idx) const
+{
+  ASSERT(left.attr_type() == AttrType::DATES && right.attr_type() == AttrType::DATES, "invalid date column type");
+  return common::compare_int(
+      (void *)&((int *)left.data())[left_idx], (void *)&((int *)right.data())[right_idx]);
+}
+
+RC DateType::to_string(const Value &val, string &result) const
+{
+  const int encoded_date = val.get_int();
+  const int year = encoded_date / 10000;
+  const int month = encoded_date / 100 % 100;
+  const int day = encoded_date % 100;
+  char buffer[16];
+  snprintf(buffer, sizeof(buffer), "%04d-%02d-%02d", year, month, day);
+  result = buffer;
+  return RC::SUCCESS;
+}
+
+RC DateType::set_value_from_str(Value &val, const string &data) const
+{
+  int encoded_date = 0;
+  RC rc = parse_date(data, encoded_date);
+  if (OB_SUCC(rc)) {
+    val.set_date(encoded_date);
+  }
+  return rc;
+}
diff --git a/src/observer/common/type/date_type.h b/src/observer/common/type/date_type.h
new file mode 100644
index 000000000..e2b4a929e
--- /dev/null
+++ b/src/observer/common/type/date_type.h
@@ -0,0 +1,27 @@
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+#pragma once
+
+#include "common/type/data_type.h"
+
+class DateType : public DataType
+{
+public:
+  DateType() : DataType(AttrType::DATES) {}
+  ~DateType() override = default;
+
+  int compare(const Value &left, const Value &right) const override;
+  int compare(const Column &left, const Column &right, int left_idx, int right_idx) const override;
+  RC  to_string(const Value &val, string &result) const override;
+  RC  set_value_from_str(Value &val, const string &data) const override;
+
+  static RC parse_date(const string &text, int &encoded_date);
+};
diff --git a/src/observer/common/value.cpp b/src/observer/common/value.cpp
index 424c67350..c383894a3 100644
--- a/src/observer/common/value.cpp
+++ b/src/observer/common/value.cpp
@@ -113,6 +113,10 @@ void Value::reset()
 void Value::set_data(char *data, int length)
 {
   switch (attr_type_) {
+    case AttrType::DATES: {
+      value_.int_value_ = *(int *)data;
+      length_           = length;
+    } break;
     case AttrType::CHARS: {
       set_string(data, length);
     } break;
@@ -149,6 +153,13 @@ void Value::set_float(float val)
   value_.float_value_ = val;
   length_             = sizeof(val);
 }
+void Value::set_date(int val)
+{
+  reset();
+  attr_type_        = AttrType::DATES;
+  value_.int_value_ = val;
+  length_           = sizeof(val);
+}
 void Value::set_boolean(bool val)
 {
   reset();
@@ -194,6 +205,9 @@ void Value::set_empty_string(int len)
 void Value::set_value(const Value &value)
 {
   switch (value.attr_type_) {
+    case AttrType::DATES: {
+      set_date(value.get_int());
+    } break;
     case AttrType::INTS: {
       set_int(value.get_int());
     } break;
@@ -250,6 +264,9 @@ int Value::compare(const Value &other) const { return DataType::type_instance(th
 int Value::get_int() const
 {
   switch (attr_type_) {
+    case AttrType::DATES: {
+      return value_.int_value_;
+    }
     case AttrType::CHARS: {
       try {
         return (int)(stol(value_.pointer_value_));
diff --git a/src/observer/common/value.h b/src/observer/common/value.h
index ca31eedc4..6264403be 100644
--- a/src/observer/common/value.h
+++ b/src/observer/common/value.h
@@ -116,6 +116,7 @@ class Value final
 public:
   void set_int(int val);
   void set_float(float val);
+  void set_date(int val);
   void set_string(const char *s, int len = 0);
   void set_empty_string(int len);
   void set_string_from_other(const Value &other);
diff --git a/src/observer/sql/executor/command_executor.cpp b/src/observer/sql/executor/command_executor.cpp
index de9527afe..0dde3c113 100644
--- a/src/observer/sql/executor/command_executor.cpp
+++ b/src/observer/sql/executor/command_executor.cpp
@@ -19,6 +19,7 @@ See the Mulan PSL v2 for more details. */
 #include "sql/executor/create_index_executor.h"
 #include "sql/executor/create_table_executor.h"
 #include "sql/executor/desc_table_executor.h"
+#include "sql/executor/drop_table_executor.h"
 #include "sql/executor/help_executor.h"
 #include "sql/executor/load_data_executor.h"
 #include "sql/executor/set_variable_executor.h"
@@ -43,6 +44,11 @@ RC CommandExecutor::execute(SQLStageEvent *sql_event)
       rc = executor.execute(sql_event);
     } break;
 
+    case StmtType::DROP_TABLE: {
+      DropTableExecutor executor;
+      rc = executor.execute(sql_event);
+    } break;
+
     case StmtType::DESC_TABLE: {
       DescTableExecutor executor;
       rc = executor.execute(sql_event);
diff --git a/src/observer/sql/executor/drop_table_executor.cpp b/src/observer/sql/executor/drop_table_executor.cpp
new file mode 100644
index 000000000..c7e3311a7
--- /dev/null
+++ b/src/observer/sql/executor/drop_table_executor.cpp
@@ -0,0 +1,30 @@
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+#include "sql/executor/drop_table_executor.h"
+
+#include "common/log/log.h"
+#include "event/session_event.h"
+#include "event/sql_event.h"
+#include "session/session.h"
+#include "sql/stmt/drop_table_stmt.h"
+#include "storage/db/db.h"
+
+RC DropTableExecutor::execute(SQLStageEvent *sql_event)
+{
+  Stmt    *stmt    = sql_event->stmt();
+  Session *session = sql_event->session_event()->session();
+  ASSERT(stmt->type() == StmtType::DROP_TABLE,
+      "drop table executor can not run this command: %d",
+      static_cast<int>(stmt->type()));
+
+  DropTableStmt *drop_table_stmt = static_cast<DropTableStmt *>(stmt);
+  return session->get_current_db()->drop_table(drop_table_stmt->table_name().c_str());
+}
diff --git a/src/observer/sql/executor/drop_table_executor.h b/src/observer/sql/executor/drop_table_executor.h
new file mode 100644
index 000000000..b78f9f6e6
--- /dev/null
+++ b/src/observer/sql/executor/drop_table_executor.h
@@ -0,0 +1,24 @@
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+#pragma once
+
+#include "common/sys/rc.h"
+
+class SQLStageEvent;
+
+class DropTableExecutor
+{
+public:
+  DropTableExecutor()          = default;
+  virtual ~DropTableExecutor() = default;
+
+  RC execute(SQLStageEvent *sql_event);
+};
diff --git a/src/observer/sql/expr/aggregator.cpp b/src/observer/sql/expr/aggregator.cpp
index 29ba468a7..ce0735de2 100644
--- a/src/observer/sql/expr/aggregator.cpp
+++ b/src/observer/sql/expr/aggregator.cpp
@@ -25,12 +25,80 @@ RC SumAggregator::accumulate(const Value &value)
   ASSERT(value.attr_type() == value_.attr_type(), "type mismatch. value type: %s, value_.type: %s", 
         attr_type_to_string(value.attr_type()), attr_type_to_string(value_.attr_type()));
   
-  Value::add(value, value_, value_);
+  return Value::add(value, value_, value_);
+}
+
+RC SumAggregator::evaluate(Value &result)
+{
+  result = value_;
+  return RC::SUCCESS;
+}
+
+RC CountAggregator::accumulate(const Value &value)
+{
+  count_++;
+  return RC::SUCCESS;
+}
+
+RC CountAggregator::evaluate(Value &result)
+{
+  result.set_int(count_);
+  return RC::SUCCESS;
+}
+
+RC AvgAggregator::accumulate(const Value &value)
+{
+  switch (value.attr_type()) {
+    case AttrType::INTS: sum_ += value.get_int(); break;
+    case AttrType::FLOATS: sum_ += value.get_float(); break;
+    default:
+      LOG_WARN("unsupported value type for avg: %s", attr_type_to_string(value.attr_type()));
+      return RC::SCHEMA_FIELD_TYPE_MISMATCH;
+  }
+
+  count_++;
+  return RC::SUCCESS;
+}
+
+RC AvgAggregator::evaluate(Value &result)
+{
+  if (count_ == 0) {
+    return RC::EMPTY;
+  }
+  result.set_float(static_cast<float>(sum_ / count_));
+  return RC::SUCCESS;
+}
+
+RC MinAggregator::accumulate(const Value &value)
+{
+  if (value_.attr_type() == AttrType::UNDEFINED || value.compare(value_) < 0) {
+    value_ = value;
+  }
+  return RC::SUCCESS;
+}
+
+RC MinAggregator::evaluate(Value &result)
+{
+  if (value_.attr_type() == AttrType::UNDEFINED) {
+    return RC::EMPTY;
+  }
+  result = value_;
   return RC::SUCCESS;
 }
 
-RC SumAggregator::evaluate(Value& result)
+RC MaxAggregator::accumulate(const Value &value)
 {
+  if (value_.attr_type() == AttrType::UNDEFINED || value.compare(value_) > 0) {
+    value_ = value;
+  }
+  return RC::SUCCESS;
+}
+
+RC MaxAggregator::evaluate(Value &result)
+{
+  if (value_.attr_type() == AttrType::UNDEFINED) {
+    return RC::EMPTY;
+  }
   result = value_;
   return RC::SUCCESS;
 }
diff --git a/src/observer/sql/expr/aggregator.h b/src/observer/sql/expr/aggregator.h
index 08a9b8ae3..e8d16a5a4 100644
--- a/src/observer/sql/expr/aggregator.h
+++ b/src/observer/sql/expr/aggregator.h
@@ -35,3 +35,38 @@ class SumAggregator : public Aggregator
   RC accumulate(const Value &value) override;
   RC evaluate(Value &result) override;
 };
+
+class CountAggregator : public Aggregator
+{
+public:
+  RC accumulate(const Value &value) override;
+  RC evaluate(Value &result) override;
+
+private:
+  int count_ = 0;
+};
+
+class AvgAggregator : public Aggregator
+{
+public:
+  RC accumulate(const Value &value) override;
+  RC evaluate(Value &result) override;
+
+private:
+  double sum_   = 0;
+  int    count_ = 0;
+};
+
+class MinAggregator : public Aggregator
+{
+public:
+  RC accumulate(const Value &value) override;
+  RC evaluate(Value &result) override;
+};
+
+class MaxAggregator : public Aggregator
+{
+public:
+  RC accumulate(const Value &value) override;
+  RC evaluate(Value &result) override;
+};
diff --git a/src/observer/sql/expr/expression.cpp b/src/observer/sql/expr/expression.cpp
index df5d1c807..2b62cc419 100644
--- a/src/observer/sql/expr/expression.cpp
+++ b/src/observer/sql/expr/expression.cpp
@@ -593,12 +593,24 @@ unique_ptr<Aggregator> AggregateExpr::create_aggregator() const
 {
   unique_ptr<Aggregator> aggregator;
   switch (aggregate_type_) {
+    case Type::COUNT: {
+      aggregator = make_unique<CountAggregator>();
+      break;
+    }
     case Type::SUM: {
       aggregator = make_unique<SumAggregator>();
       break;
     }
-    default: {
-      ASSERT(false, "unsupported aggregate type");
+    case Type::AVG: {
+      aggregator = make_unique<AvgAggregator>();
+      break;
+    }
+    case Type::MAX: {
+      aggregator = make_unique<MaxAggregator>();
+      break;
+    }
+    case Type::MIN: {
+      aggregator = make_unique<MinAggregator>();
       break;
     }
   }
diff --git a/src/observer/sql/operator/logical_operator.cpp b/src/observer/sql/operator/logical_operator.cpp
index a2fb251db..f5ea5e5b6 100644
--- a/src/observer/sql/operator/logical_operator.cpp
+++ b/src/observer/sql/operator/logical_operator.cpp
@@ -26,6 +26,7 @@ bool LogicalOperator::can_generate_vectorized_operator(const LogicalOperatorType
   switch (type)
   {
   case LogicalOperatorType::CALC:
+  case LogicalOperatorType::UPDATE:
   case LogicalOperatorType::DELETE:
   case LogicalOperatorType::INSERT:
     bool_ret = false;
@@ -45,4 +46,3 @@ void LogicalOperator::generate_general_child()
     child->generate_general_child();
   }
 }
-
diff --git a/src/observer/sql/operator/logical_operator.h b/src/observer/sql/operator/logical_operator.h
index d212dbcc2..a9a93208f 100644
--- a/src/observer/sql/operator/logical_operator.h
+++ b/src/observer/sql/operator/logical_operator.h
@@ -32,6 +32,7 @@ See the Mulan PSL v2 for more details. */
 enum class LogicalOperatorType
 {
   CALC,
+  UPDATE,
   TABLE_GET,   ///< 从表中获取数据
   PREDICATE,   ///< 过滤，就是谓词
   PROJECTION,  ///< 投影，就是select
diff --git a/src/observer/sql/operator/physical_operator.cpp b/src/observer/sql/operator/physical_operator.cpp
index 7ff99d67c..2d2dfb0bb 100644
--- a/src/observer/sql/operator/physical_operator.cpp
+++ b/src/observer/sql/operator/physical_operator.cpp
@@ -17,6 +17,7 @@ See the Mulan PSL v2 for more details. */
 string physical_operator_type_name(PhysicalOperatorType type)
 {
   switch (type) {
+    case PhysicalOperatorType::UPDATE: return "UPDATE";
     case PhysicalOperatorType::TABLE_SCAN: return "TABLE_SCAN";
     case PhysicalOperatorType::INDEX_SCAN: return "INDEX_SCAN";
     case PhysicalOperatorType::NESTED_LOOP_JOIN: return "NESTED_LOOP_JOIN";
diff --git a/src/observer/sql/operator/physical_operator.h b/src/observer/sql/operator/physical_operator.h
index 8f2b67ba1..24f0489ca 100644
--- a/src/observer/sql/operator/physical_operator.h
+++ b/src/observer/sql/operator/physical_operator.h
@@ -34,6 +34,7 @@ class Trx;
  */
 enum class PhysicalOperatorType
 {
+  UPDATE,
   TABLE_SCAN,
   TABLE_SCAN_VEC,
   INDEX_SCAN,
diff --git a/src/observer/sql/operator/scalar_group_by_physical_operator.cpp b/src/observer/sql/operator/scalar_group_by_physical_operator.cpp
index d07546dbe..35baad6e6 100644
--- a/src/observer/sql/operator/scalar_group_by_physical_operator.cpp
+++ b/src/observer/sql/operator/scalar_group_by_physical_operator.cpp
@@ -17,6 +17,8 @@ See the Mulan PSL v2 for more details. */
 #include "sql/expr/expression_tuple.h"
 #include "sql/expr/composite_tuple.h"
 
+#include <algorithm>
+
 using namespace std;
 using namespace common;
 
@@ -82,6 +84,20 @@ RC ScalarGroupByPhysicalOperator::open(Trx *trx)
     return rc;
   }
 
+  // COUNT over an empty input still produces one row whose value is zero.
+  if (group_value_ == nullptr) {
+    const bool only_count =
+        std::all_of(aggregate_expressions_.begin(), aggregate_expressions_.end(), [](Expression *expr) {
+          auto *aggregate_expr = static_cast<AggregateExpr *>(expr);
+          return aggregate_expr->aggregate_type() == AggregateExpr::Type::COUNT;
+        });
+    if (only_count) {
+      AggregatorList aggregator_list;
+      create_aggregator_list(aggregator_list);
+      group_value_ = make_unique<GroupValueType>(std::move(aggregator_list), CompositeTuple());
+    }
+  }
+
   // 得到最终聚合后的值
   if (group_value_) {
     rc = evaluate(*group_value_);
@@ -117,4 +133,4 @@ Tuple *ScalarGroupByPhysicalOperator::current_tuple()
   }
 
   return &get<1>(*group_value_);
-}
\ No newline at end of file
+}
diff --git a/src/observer/sql/operator/update_logical_operator.h b/src/observer/sql/operator/update_logical_operator.h
new file mode 100644
index 000000000..73bee3d49
--- /dev/null
+++ b/src/observer/sql/operator/update_logical_operator.h
@@ -0,0 +1,37 @@
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+#pragma once
+
+#include "sql/operator/logical_operator.h"
+
+class Table;
+class FieldMeta;
+
+class UpdateLogicalOperator : public LogicalOperator
+{
+public:
+  UpdateLogicalOperator(Table *table, const FieldMeta *field_meta, const Value &value)
+      : table_(table), field_meta_(field_meta), value_(value)
+  {}
+  ~UpdateLogicalOperator() override = default;
+
+  LogicalOperatorType type() const override { return LogicalOperatorType::UPDATE; }
+  OpType              get_op_type() const override { return OpType::LOGICALUPDATE; }
+
+  Table           *table() const { return table_; }
+  const FieldMeta *field_meta() const { return field_meta_; }
+  const Value     &value() const { return value_; }
+
+private:
+  Table           *table_      = nullptr;
+  const FieldMeta *field_meta_ = nullptr;
+  Value            value_;
+};
diff --git a/src/observer/sql/operator/update_physical_operator.cpp b/src/observer/sql/operator/update_physical_operator.cpp
new file mode 100644
index 000000000..1f5894a32
--- /dev/null
+++ b/src/observer/sql/operator/update_physical_operator.cpp
@@ -0,0 +1,67 @@
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+#include "sql/operator/update_physical_operator.h"
+
+#include "common/log/log.h"
+#include "storage/table/table.h"
+#include "storage/trx/trx.h"
+
+RC UpdatePhysicalOperator::open(Trx *trx)
+{
+  if (children_.empty()) {
+    return RC::SUCCESS;
+  }
+
+  unique_ptr<PhysicalOperator> &child = children_[0];
+  RC rc = child->open(trx);
+  if (OB_FAIL(rc)) {
+    LOG_WARN("failed to open child operator for update. rc=%s", strrc(rc));
+    return rc;
+  }
+
+  while (OB_SUCC(rc = child->next())) {
+    Tuple *tuple = child->current_tuple();
+    if (tuple == nullptr) {
+      child->close();
+      return RC::INTERNAL;
+    }
+
+    RowTuple *row_tuple = static_cast<RowTuple *>(tuple);
+    records_.emplace_back(row_tuple->record());
+  }
+
+  RC close_rc = child->close();
+  if (rc != RC::RECORD_EOF) {
+    return rc;
+  }
+  if (OB_FAIL(close_rc)) {
+    return close_rc;
+  }
+
+  for (Record &old_record : records_) {
+    Record new_record;
+    rc = table_->make_updated_record(old_record, *field_meta_, value_, new_record);
+    if (OB_FAIL(rc)) {
+      LOG_WARN("failed to make updated record. table=%s, field=%s, rc=%s",
+          table_->name(), field_meta_->name(), strrc(rc));
+      return rc;
+    }
+
+    rc = trx->update_record(table_, old_record, new_record);
+    if (OB_FAIL(rc)) {
+      LOG_WARN("failed to update record. table=%s, rid=%s, rc=%s",
+          table_->name(), old_record.rid().to_string().c_str(), strrc(rc));
+      return rc;
+    }
+  }
+
+  return RC::SUCCESS;
+}
diff --git a/src/observer/sql/operator/update_physical_operator.h b/src/observer/sql/operator/update_physical_operator.h
new file mode 100644
index 000000000..248589fcf
--- /dev/null
+++ b/src/observer/sql/operator/update_physical_operator.h
@@ -0,0 +1,40 @@
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+#pragma once
+
+#include "sql/operator/physical_operator.h"
+#include "storage/record/record.h"
+
+class Table;
+class FieldMeta;
+class Trx;
+
+class UpdatePhysicalOperator : public PhysicalOperator
+{
+public:
+  UpdatePhysicalOperator(Table *table, const FieldMeta *field_meta, const Value &value)
+      : table_(table), field_meta_(field_meta), value_(value)
+  {}
+  ~UpdatePhysicalOperator() override = default;
+
+  PhysicalOperatorType type() const override { return PhysicalOperatorType::UPDATE; }
+  OpType                get_op_type() const override { return OpType::UPDATE; }
+
+  RC open(Trx *trx) override;
+  RC next() override { return RC::RECORD_EOF; }
+  RC close() override { return RC::SUCCESS; }
+
+private:
+  Table           *table_      = nullptr;
+  const FieldMeta *field_meta_ = nullptr;
+  Value            value_;
+  vector<Record>   records_;
+};
diff --git a/src/observer/sql/optimizer/cascade/implementation_rules.cpp b/src/observer/sql/optimizer/cascade/implementation_rules.cpp
index 740f44381..6f6a334b0 100644
--- a/src/observer/sql/optimizer/cascade/implementation_rules.cpp
+++ b/src/observer/sql/optimizer/cascade/implementation_rules.cpp
@@ -22,6 +22,8 @@ See the Mulan PSL v2 for more details. */
 #include "sql/operator/calc_physical_operator.h"
 #include "sql/operator/delete_logical_operator.h"
 #include "sql/operator/delete_physical_operator.h"
+#include "sql/operator/update_logical_operator.h"
+#include "sql/operator/update_physical_operator.h"
 #include "sql/operator/predicate_logical_operator.h"
 #include "sql/operator/predicate_physical_operator.h"
 #include "sql/operator/group_by_logical_operator.h"
@@ -171,6 +173,25 @@ void LogicalDeleteToDelete::transform(OperatorNode* input,
   transformed->emplace_back(std::move(delete_phys_oper));
 }
 
+LogicalUpdateToUpdate::LogicalUpdateToUpdate()
+{
+  type_ = RuleType::UPDATE_TO_PHYSICAL;
+  match_pattern_ = unique_ptr<Pattern>(new Pattern(OpType::LOGICALUPDATE));
+  match_pattern_->add_child(new Pattern(OpType::LEAF));
+}
+
+void LogicalUpdateToUpdate::transform(OperatorNode *input,
+    std::vector<std::unique_ptr<OperatorNode>> *transformed, OptimizerContext *context) const
+{
+  auto update_oper = dynamic_cast<UpdateLogicalOperator *>(input);
+  auto update_phys_oper = make_unique<UpdatePhysicalOperator>(
+      update_oper->table(), update_oper->field_meta(), update_oper->value());
+  for (auto &child : update_oper->children()) {
+    update_phys_oper->add_general_child(child.get());
+  }
+  transformed->emplace_back(std::move(update_phys_oper));
+}
+
 // -------------------------------------------------------------------------------------------------
 // Physical Predicate
 // -------------------------------------------------------------------------------------------------
@@ -261,4 +282,4 @@ void LogicalPredicateToPredicate::transform(OperatorNode* input,
 //   }
 
 //   transformed->emplace_back(std::move(groupby_phys_oper));
-// }
\ No newline at end of file
+// }
diff --git a/src/observer/sql/optimizer/cascade/implementation_rules.h b/src/observer/sql/optimizer/cascade/implementation_rules.h
index d82182ce6..86e9d9539 100644
--- a/src/observer/sql/optimizer/cascade/implementation_rules.h
+++ b/src/observer/sql/optimizer/cascade/implementation_rules.h
@@ -88,6 +88,15 @@ class LogicalDeleteToDelete : public Rule
       OptimizerContext *context) const override;
 };
 
+class LogicalUpdateToUpdate : public Rule
+{
+public:
+  LogicalUpdateToUpdate();
+
+  void transform(OperatorNode *input, std::vector<std::unique_ptr<OperatorNode>> *transformed,
+      OptimizerContext *context) const override;
+};
+
 /**
  * Rule transforms Logical predicate -> Physical predicate
  * TODO: In practice, this rule may not be used and can be removed
@@ -125,4 +134,4 @@ class LogicalPredicateToPredicate : public Rule
 //   void transform(OperatorNode* input,
 //                          std::vector<std::unique_ptr<OperatorNode>> *transformed,
 //                          OptimizerContext *context) const override;
-// };
\ No newline at end of file
+// };
diff --git a/src/observer/sql/optimizer/cascade/rules.cpp b/src/observer/sql/optimizer/cascade/rules.cpp
index c78fbf596..10d8b3c54 100644
--- a/src/observer/sql/optimizer/cascade/rules.cpp
+++ b/src/observer/sql/optimizer/cascade/rules.cpp
@@ -20,5 +20,6 @@ RuleSet::RuleSet()
   add_rule(RuleSetName::PHYSICAL_IMPLEMENTATION, new LogicalExplainToExplain());
   add_rule(RuleSetName::PHYSICAL_IMPLEMENTATION, new LogicalCalcToCalc());
   add_rule(RuleSetName::PHYSICAL_IMPLEMENTATION, new LogicalDeleteToDelete());
+  add_rule(RuleSetName::PHYSICAL_IMPLEMENTATION, new LogicalUpdateToUpdate());
   add_rule(RuleSetName::PHYSICAL_IMPLEMENTATION, new LogicalPredicateToPredicate());
-}
\ No newline at end of file
+}
diff --git a/src/observer/sql/optimizer/logical_plan_generator.cpp b/src/observer/sql/optimizer/logical_plan_generator.cpp
index f9da53f3a..f6490c803 100644
--- a/src/observer/sql/optimizer/logical_plan_generator.cpp
+++ b/src/observer/sql/optimizer/logical_plan_generator.cpp
@@ -25,6 +25,7 @@ See the Mulan PSL v2 for more details. */
 #include "sql/operator/predicate_logical_operator.h"
 #include "sql/operator/project_logical_operator.h"
 #include "sql/operator/table_get_logical_operator.h"
+#include "sql/operator/update_logical_operator.h"
 #include "sql/operator/group_by_logical_operator.h"
 
 #include "sql/stmt/calc_stmt.h"
@@ -34,6 +35,7 @@ See the Mulan PSL v2 for more details. */
 #include "sql/stmt/insert_stmt.h"
 #include "sql/stmt/select_stmt.h"
 #include "sql/stmt/stmt.h"
+#include "sql/stmt/update_stmt.h"
 
 #include "sql/expr/expression_iterator.h"
 
@@ -68,6 +70,11 @@ RC LogicalPlanGenerator::create(Stmt *stmt, unique_ptr<LogicalOperator> &logical
       rc = create_plan(delete_stmt, logical_operator);
     } break;
 
+    case StmtType::UPDATE: {
+      UpdateStmt *update_stmt = static_cast<UpdateStmt *>(stmt);
+      rc = create_plan(update_stmt, logical_operator);
+    } break;
+
     case StmtType::EXPLAIN: {
       ExplainStmt *explain_stmt = static_cast<ExplainStmt *>(stmt);
 
@@ -263,6 +270,31 @@ RC LogicalPlanGenerator::create_plan(DeleteStmt *delete_stmt, unique_ptr<Logical
   return rc;
 }
 
+RC LogicalPlanGenerator::create_plan(UpdateStmt *update_stmt, unique_ptr<LogicalOperator> &logical_operator)
+{
+  Table                      *table       = update_stmt->table();
+  FilterStmt                 *filter_stmt = update_stmt->filter_stmt();
+  unique_ptr<LogicalOperator> table_get_oper(new TableGetLogicalOperator(table, ReadWriteMode::READ_WRITE));
+
+  unique_ptr<LogicalOperator> predicate_oper;
+  RC rc = create_plan(filter_stmt, predicate_oper);
+  if (OB_FAIL(rc)) {
+    return rc;
+  }
+
+  unique_ptr<LogicalOperator> update_oper(
+      new UpdateLogicalOperator(table, update_stmt->field_meta(), update_stmt->value()));
+  if (predicate_oper) {
+    predicate_oper->add_child(std::move(table_get_oper));
+    update_oper->add_child(std::move(predicate_oper));
+  } else {
+    update_oper->add_child(std::move(table_get_oper));
+  }
+
+  logical_operator = std::move(update_oper);
+  return RC::SUCCESS;
+}
+
 RC LogicalPlanGenerator::create_plan(ExplainStmt *explain_stmt, unique_ptr<LogicalOperator> &logical_operator)
 {
   unique_ptr<LogicalOperator> child_oper;
diff --git a/src/observer/sql/optimizer/logical_plan_generator.h b/src/observer/sql/optimizer/logical_plan_generator.h
index e0707deff..852d4cee7 100644
--- a/src/observer/sql/optimizer/logical_plan_generator.h
+++ b/src/observer/sql/optimizer/logical_plan_generator.h
@@ -24,6 +24,7 @@ class SelectStmt;
 class FilterStmt;
 class InsertStmt;
 class DeleteStmt;
+class UpdateStmt;
 class ExplainStmt;
 class LogicalOperator;
 
@@ -41,6 +42,7 @@ class LogicalPlanGenerator
   RC create_plan(FilterStmt *filter_stmt, unique_ptr<LogicalOperator> &logical_operator);
   RC create_plan(InsertStmt *insert_stmt, unique_ptr<LogicalOperator> &logical_operator);
   RC create_plan(DeleteStmt *delete_stmt, unique_ptr<LogicalOperator> &logical_operator);
+  RC create_plan(UpdateStmt *update_stmt, unique_ptr<LogicalOperator> &logical_operator);
   RC create_plan(ExplainStmt *explain_stmt, unique_ptr<LogicalOperator> &logical_operator);
 
   RC create_group_by_plan(SelectStmt *select_stmt, unique_ptr<LogicalOperator> &logical_operator);
diff --git a/src/observer/sql/optimizer/physical_plan_generator.cpp b/src/observer/sql/optimizer/physical_plan_generator.cpp
index 4ce64a24b..278ec8040 100644
--- a/src/observer/sql/optimizer/physical_plan_generator.cpp
+++ b/src/observer/sql/optimizer/physical_plan_generator.cpp
@@ -42,6 +42,8 @@ See the Mulan PSL v2 for more details. */
 #include "sql/operator/hash_group_by_physical_operator.h"
 #include "sql/operator/scalar_group_by_physical_operator.h"
 #include "sql/operator/table_scan_vec_physical_operator.h"
+#include "sql/operator/update_logical_operator.h"
+#include "sql/operator/update_physical_operator.h"
 #include "sql/optimizer/physical_plan_generator.h"
 
 using namespace std;
@@ -75,6 +77,10 @@ RC PhysicalPlanGenerator::create(LogicalOperator &logical_operator, unique_ptr<P
       return create_plan(static_cast<DeleteLogicalOperator &>(logical_operator), oper, session);
     } break;
 
+    case LogicalOperatorType::UPDATE: {
+      return create_plan(static_cast<UpdateLogicalOperator &>(logical_operator), oper, session);
+    } break;
+
     case LogicalOperatorType::EXPLAIN: {
       return create_plan(static_cast<ExplainLogicalOperator &>(logical_operator), oper, session);
     } break;
@@ -276,6 +282,27 @@ RC PhysicalPlanGenerator::create_plan(DeleteLogicalOperator &delete_oper, unique
   return rc;
 }
 
+RC PhysicalPlanGenerator::create_plan(UpdateLogicalOperator &update_oper, unique_ptr<PhysicalOperator> &oper, Session *session)
+{
+  vector<unique_ptr<LogicalOperator>> &child_opers = update_oper.children();
+  unique_ptr<PhysicalOperator> child_physical_oper;
+
+  RC rc = RC::SUCCESS;
+  if (!child_opers.empty()) {
+    rc = create(*child_opers.front(), child_physical_oper, session);
+    if (OB_FAIL(rc)) {
+      LOG_WARN("failed to create child physical operator for update. rc=%s", strrc(rc));
+      return rc;
+    }
+  }
+
+  oper = make_unique<UpdatePhysicalOperator>(update_oper.table(), update_oper.field_meta(), update_oper.value());
+  if (child_physical_oper) {
+    oper->add_child(std::move(child_physical_oper));
+  }
+  return RC::SUCCESS;
+}
+
 RC PhysicalPlanGenerator::create_plan(ExplainLogicalOperator &explain_oper, unique_ptr<PhysicalOperator> &oper, Session* session)
 {
   vector<unique_ptr<LogicalOperator>> &child_opers = explain_oper.children();
diff --git a/src/observer/sql/optimizer/physical_plan_generator.h b/src/observer/sql/optimizer/physical_plan_generator.h
index 85f2e27d2..f08a1210d 100644
--- a/src/observer/sql/optimizer/physical_plan_generator.h
+++ b/src/observer/sql/optimizer/physical_plan_generator.h
@@ -24,6 +24,7 @@ class PredicateLogicalOperator;
 class ProjectLogicalOperator;
 class InsertLogicalOperator;
 class DeleteLogicalOperator;
+class UpdateLogicalOperator;
 class ExplainLogicalOperator;
 class JoinLogicalOperator;
 class CalcLogicalOperator;
@@ -50,6 +51,7 @@ class PhysicalPlanGenerator
   RC create_plan(ProjectLogicalOperator &logical_oper, unique_ptr<PhysicalOperator> &oper, Session *session);
   RC create_plan(InsertLogicalOperator &logical_oper, unique_ptr<PhysicalOperator> &oper, Session *session);
   RC create_plan(DeleteLogicalOperator &logical_oper, unique_ptr<PhysicalOperator> &oper, Session *session);
+  RC create_plan(UpdateLogicalOperator &logical_oper, unique_ptr<PhysicalOperator> &oper, Session *session);
   RC create_plan(ExplainLogicalOperator &logical_oper, unique_ptr<PhysicalOperator> &oper, Session *session);
   RC create_plan(JoinLogicalOperator &logical_oper, unique_ptr<PhysicalOperator> &oper, Session *session);
   RC create_plan(CalcLogicalOperator &logical_oper, unique_ptr<PhysicalOperator> &oper, Session *session);
diff --git a/src/observer/sql/parser/lex_sql.l b/src/observer/sql/parser/lex_sql.l
index 42f7a89f3..90935cbbe 100644
--- a/src/observer/sql/parser/lex_sql.l
+++ b/src/observer/sql/parser/lex_sql.l
@@ -111,6 +111,7 @@ ROLLBACK                                RETURN_TOKEN(TRX_ROLLBACK);
 INT                                     RETURN_TOKEN(INT_T);
 CHAR                                    RETURN_TOKEN(STRING_T);
 FLOAT                                   RETURN_TOKEN(FLOAT_T);
+DATE                                    RETURN_TOKEN(DATE_T);
 VECTOR                                  RETURN_TOKEN(VECTOR_T);
 LOAD                                    RETURN_TOKEN(LOAD);
 DATA                                    RETURN_TOKEN(DATA);
diff --git a/src/observer/sql/parser/yacc_sql.y b/src/observer/sql/parser/yacc_sql.y
index 73ec35b9f..01f5e0d87 100644
--- a/src/observer/sql/parser/yacc_sql.y
+++ b/src/observer/sql/parser/yacc_sql.y
@@ -88,6 +88,7 @@ UnboundAggregateExpr *create_aggregate_expression(const char *aggregate_name,
         INT_T
         STRING_T
         FLOAT_T
+        DATE_T
         VECTOR_T
         HELP
         EXIT
@@ -380,6 +381,7 @@ type:
     INT_T      { $$ = static_cast<int>(AttrType::INTS); }
     | STRING_T { $$ = static_cast<int>(AttrType::CHARS); }
     | FLOAT_T  { $$ = static_cast<int>(AttrType::FLOATS); }
+    | DATE_T   { $$ = static_cast<int>(AttrType::DATES); }
     | VECTOR_T { $$ = static_cast<int>(AttrType::VECTORS); }
     ;
 primary_key:
@@ -476,6 +478,7 @@ update_stmt:      /*  update 语句的语法解析树*/
       $$->update.relation_name = $2;
       $$->update.attribute_name = $4;
       $$->update.value = *$6;
+      delete $6;
       if ($7 != nullptr) {
         $$->update.conditions.swap(*$7);
         delete $7;
@@ -789,4 +792,4 @@ int sql_parse(const char *s, ParsedSqlResult *sql_result) {
 
   yylex_destroy(scanner);
   return result;
-}
\ No newline at end of file
+}
diff --git a/src/observer/sql/stmt/drop_table_stmt.cpp b/src/observer/sql/stmt/drop_table_stmt.cpp
new file mode 100644
index 000000000..8ba49a1aa
--- /dev/null
+++ b/src/observer/sql/stmt/drop_table_stmt.cpp
@@ -0,0 +1,23 @@
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+#include "sql/stmt/drop_table_stmt.h"
+
+#include "storage/db/db.h"
+
+RC DropTableStmt::create(Db *db, const DropTableSqlNode &drop_table, Stmt *&stmt)
+{
+  if (db == nullptr || db->find_table(drop_table.relation_name.c_str()) == nullptr) {
+    return RC::SCHEMA_TABLE_NOT_EXIST;
+  }
+
+  stmt = new DropTableStmt(drop_table.relation_name);
+  return RC::SUCCESS;
+}
diff --git a/src/observer/sql/stmt/drop_table_stmt.h b/src/observer/sql/stmt/drop_table_stmt.h
new file mode 100644
index 000000000..ec0ad218b
--- /dev/null
+++ b/src/observer/sql/stmt/drop_table_stmt.h
@@ -0,0 +1,34 @@
+/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
+miniob is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details. */
+
+#pragma once
+
+#include "sql/stmt/stmt.h"
+
+class Db;
+
+/**
+ * @brief Statement used to drop a table.
+ */
+class DropTableStmt : public Stmt
+{
+public:
+  explicit DropTableStmt(const string &table_name) : table_name_(table_name) {}
+  ~DropTableStmt() override = default;
+
+  StmtType type() const override { return StmtType::DROP_TABLE; }
+
+  const string &table_name() const { return table_name_; }
+
+  static RC create(Db *db, const DropTableSqlNode &drop_table, Stmt *&stmt);
+
+private:
+  string table_name_;
+};
diff --git a/src/observer/sql/stmt/stmt.cpp b/src/observer/sql/stmt/stmt.cpp
index 2664aaf60..572c9d2e2 100644
--- a/src/observer/sql/stmt/stmt.cpp
+++ b/src/observer/sql/stmt/stmt.cpp
@@ -20,6 +20,7 @@ See the Mulan PSL v2 for more details. */
 #include "sql/stmt/create_table_stmt.h"
 #include "sql/stmt/delete_stmt.h"
 #include "sql/stmt/desc_table_stmt.h"
+#include "sql/stmt/drop_table_stmt.h"
 #include "sql/stmt/exit_stmt.h"
 #include "sql/stmt/explain_stmt.h"
 #include "sql/stmt/help_stmt.h"
@@ -30,6 +31,7 @@ See the Mulan PSL v2 for more details. */
 #include "sql/stmt/show_tables_stmt.h"
 #include "sql/stmt/trx_begin_stmt.h"
 #include "sql/stmt/trx_end_stmt.h"
+#include "sql/stmt/update_stmt.h"
 
 bool stmt_type_ddl(StmtType type)
 {
@@ -56,6 +58,9 @@ RC Stmt::create_stmt(Db *db, ParsedSqlNode &sql_node, Stmt *&stmt)
     case SCF_DELETE: {
       return DeleteStmt::create(db, sql_node.deletion, stmt);
     }
+    case SCF_UPDATE: {
+      return UpdateStmt::create(db, sql_node.update, stmt);
+    }
     case SCF_SELECT: {
       return SelectStmt::create(db, sql_node.selection, stmt);
     }
@@ -72,6 +77,10 @@ RC Stmt::create_stmt(Db *db, ParsedSqlNode &sql_node, Stmt *&stmt)
       return CreateTableStmt::create(db, sql_node.create_table, stmt);
     }
 
+    case SCF_DROP_TABLE: {
+      return DropTableStmt::create(db, sql_node.drop_table, stmt);
+    }
+
     case SCF_DESC_TABLE: {
       return DescTableStmt::create(db, sql_node.desc_table, stmt);
     }
diff --git a/src/observer/sql/stmt/update_stmt.cpp b/src/observer/sql/stmt/update_stmt.cpp
index 92df6a6dc..e607e2b3c 100644
--- a/src/observer/sql/stmt/update_stmt.cpp
+++ b/src/observer/sql/stmt/update_stmt.cpp
@@ -13,14 +13,63 @@ See the Mulan PSL v2 for more details. */
 //
 
 #include "sql/stmt/update_stmt.h"
+#include "common/log/log.h"
+#include "sql/stmt/filter_stmt.h"
+#include "storage/db/db.h"
+#include "storage/table/table.h"
 
-UpdateStmt::UpdateStmt(Table *table, Value *values, int value_amount)
-    : table_(table), values_(values), value_amount_(value_amount)
+UpdateStmt::UpdateStmt(Table *table, const FieldMeta *field_meta, const Value &value, FilterStmt *filter_stmt)
+    : table_(table), field_meta_(field_meta), value_(value), filter_stmt_(filter_stmt)
 {}
 
+UpdateStmt::~UpdateStmt()
+{
+  delete filter_stmt_;
+  filter_stmt_ = nullptr;
+}
+
 RC UpdateStmt::create(Db *db, const UpdateSqlNode &update, Stmt *&stmt)
 {
-  // TODO
-  stmt = nullptr;
-  return RC::INTERNAL;
+  if (db == nullptr || update.relation_name.empty() || update.attribute_name.empty()) {
+    return RC::INVALID_ARGUMENT;
+  }
+
+  Table *table = db->find_table(update.relation_name.c_str());
+  if (table == nullptr) {
+    LOG_WARN("no such table. db=%s, table_name=%s", db->name(), update.relation_name.c_str());
+    return RC::SCHEMA_TABLE_NOT_EXIST;
+  }
+
+  const FieldMeta *field_meta = table->table_meta().field(update.attribute_name.c_str());
+  if (field_meta == nullptr) {
+    LOG_WARN("no such field. table=%s, field=%s", table->name(), update.attribute_name.c_str());
+    return RC::SCHEMA_FIELD_NOT_EXIST;
+  }
+
+  Value value = update.value;
+  if (value.attr_type() != field_meta->type()) {
+    Value cast_value;
+    RC rc = Value::cast_to(value, field_meta->type(), cast_value);
+    if (OB_FAIL(rc)) {
+      LOG_WARN("failed to cast update value. table=%s, field=%s, from=%s, to=%s, rc=%s",
+          table->name(), field_meta->name(), attr_type_to_string(value.attr_type()),
+          attr_type_to_string(field_meta->type()), strrc(rc));
+      return RC::SCHEMA_FIELD_TYPE_MISMATCH;
+    }
+    value = cast_value;
+  }
+
+  unordered_map<string, Table *> table_map;
+  table_map.emplace(update.relation_name, table);
+
+  FilterStmt *filter_stmt = nullptr;
+  RC rc = FilterStmt::create(db, table, &table_map, update.conditions.data(),
+      static_cast<int>(update.conditions.size()), filter_stmt);
+  if (OB_FAIL(rc)) {
+    LOG_WARN("failed to create filter statement for update. rc=%s", strrc(rc));
+    return rc;
+  }
+
+  stmt = new UpdateStmt(table, field_meta, value, filter_stmt);
+  return RC::SUCCESS;
 }
diff --git a/src/observer/sql/stmt/update_stmt.h b/src/observer/sql/stmt/update_stmt.h
index 28e13268e..1ad4ec72a 100644
--- a/src/observer/sql/stmt/update_stmt.h
+++ b/src/observer/sql/stmt/update_stmt.h
@@ -18,6 +18,8 @@ See the Mulan PSL v2 for more details. */
 #include "sql/stmt/stmt.h"
 
 class Table;
+class FieldMeta;
+class FilterStmt;
 
 /**
  * @brief 更新语句
@@ -26,19 +28,23 @@ class Table;
 class UpdateStmt : public Stmt
 {
 public:
-  UpdateStmt() = default;
-  UpdateStmt(Table *table, Value *values, int value_amount);
+  UpdateStmt(Table *table, const FieldMeta *field_meta, const Value &value, FilterStmt *filter_stmt);
+  ~UpdateStmt() override;
+
+  StmtType type() const override { return StmtType::UPDATE; }
 
 public:
   static RC create(Db *db, const UpdateSqlNode &update_sql, Stmt *&stmt);
 
 public:
-  Table *table() const { return table_; }
-  Value *values() const { return values_; }
-  int    value_amount() const { return value_amount_; }
+  Table           *table() const { return table_; }
+  const FieldMeta *field_meta() const { return field_meta_; }
+  const Value     &value() const { return value_; }
+  FilterStmt      *filter_stmt() const { return filter_stmt_; }
 
 private:
-  Table *table_        = nullptr;
-  Value *values_       = nullptr;
-  int    value_amount_ = 0;
+  Table           *table_       = nullptr;
+  const FieldMeta *field_meta_  = nullptr;
+  Value            value_;
+  FilterStmt      *filter_stmt_ = nullptr;
 };
diff --git a/src/observer/storage/common/codec.h b/src/observer/storage/common/codec.h
index ed373de70..9f5a50bf8 100644
--- a/src/observer/storage/common/codec.h
+++ b/src/observer/storage/common/codec.h
@@ -418,6 +418,7 @@ class Codec
   {
     RC rc = RC::SUCCESS;
     switch (val.attr_type()) {
+      case AttrType::DATES:
       case AttrType::INTS:
         if (OB_FAIL(OrderedCode::append(dst, (int64_t)val.get_int()))) {
           LOG_WARN("append failed");
diff --git a/src/observer/storage/db/db.cpp b/src/observer/storage/db/db.cpp
index a0661fa00..ce4c405f1 100644
--- a/src/observer/storage/db/db.cpp
+++ b/src/observer/storage/db/db.cpp
@@ -176,6 +176,65 @@ RC Db::create_table(const char *table_name, span<const AttrInfoSqlNode> attribut
   return RC::SUCCESS;
 }
 
+RC Db::drop_table(const char *table_name)
+{
+  if (common::is_blank(table_name)) {
+    return RC::INVALID_ARGUMENT;
+  }
+
+  auto table_iter = opened_tables_.find(table_name);
+  if (table_iter == opened_tables_.end()) {
+    return RC::SCHEMA_TABLE_NOT_EXIST;
+  }
+
+  Table *table = table_iter->second;
+  vector<string> index_files;
+  const TableMeta &table_meta = table->table_meta();
+  for (int i = 0; i < table_meta.index_num(); i++) {
+    const IndexMeta *index_meta = table_meta.index(i);
+    index_files.emplace_back(table_index_file(path_.c_str(), table_name, index_meta->name()));
+  }
+
+  const filesystem::path meta_file = table_meta_file(path_.c_str(), table_name);
+  filesystem::path dropped_meta_file = meta_file;
+  dropped_meta_file += ".drop";
+
+  error_code ec;
+  filesystem::rename(meta_file, dropped_meta_file, ec);
+  if (ec) {
+    LOG_ERROR("Failed to mark table metadata as dropped. table=%s, file=%s, error=%s",
+        table_name, meta_file.c_str(), ec.message().c_str());
+    return RC::IOERR_WRITE;
+  }
+
+  opened_tables_.erase(table_iter);
+  delete table;
+
+  vector<filesystem::path> files_to_remove;
+  files_to_remove.emplace_back(table_data_file(path_.c_str(), table_name));
+  files_to_remove.emplace_back(table_lob_file(path_.c_str(), table_name));
+  for (const string &index_file : index_files) {
+    files_to_remove.emplace_back(index_file);
+  }
+  files_to_remove.emplace_back(dropped_meta_file);
+
+  RC rc = RC::SUCCESS;
+  for (const filesystem::path &file : files_to_remove) {
+    ec.clear();
+    filesystem::remove(file, ec);
+    if (ec) {
+      LOG_ERROR("Failed to remove table file. table=%s, file=%s, error=%s",
+          table_name, file.c_str(), ec.message().c_str());
+      rc = RC::IOERR_WRITE;
+    }
+  }
+
+  if (OB_SUCC(rc)) {
+    LOG_INFO("Drop table success. table=%s", table_name);
+  }
+  return rc;
+}
+
 Table *Db::find_table(const char *table_name) const
 {
   unordered_map<string, Table *>::const_iterator iter = opened_tables_.find(table_name);
diff --git a/src/observer/storage/db/db.h b/src/observer/storage/db/db.h
index 51afce10a..0a32f1096 100644
--- a/src/observer/storage/db/db.h
+++ b/src/observer/storage/db/db.h
@@ -70,6 +70,11 @@ class Db
   RC create_table(const char *table_name, span<const AttrInfoSqlNode> attributes, const vector<string> &primary_keys,
       const StorageFormat storage_format = StorageFormat::ROW_FORMAT);
 
+  /**
+   * @brief Drop a table and remove all of its persistent files.
+   */
+  RC drop_table(const char *table_name);
+
   /**
    * @brief 根据表名查找表
    */
diff --git a/src/observer/storage/default/default_handler.cpp b/src/observer/storage/default/default_handler.cpp
index 48f9e0098..1a68dfc34 100644
--- a/src/observer/storage/default/default_handler.cpp
+++ b/src/observer/storage/default/default_handler.cpp
@@ -142,7 +142,14 @@ RC DefaultHandler::create_table(const char *dbname, const char *relation_name, s
   return db->create_table(relation_name, attributes, {});
 }
 
-RC DefaultHandler::drop_table(const char *dbname, const char *relation_name) { return RC::UNIMPLEMENTED; }
+RC DefaultHandler::drop_table(const char *dbname, const char *relation_name)
+{
+  Db *db = find_db(dbname);
+  if (db == nullptr) {
+    return RC::SCHEMA_DB_NOT_OPENED;
+  }
+  return db->drop_table(relation_name);
+}
 
 Db *DefaultHandler::find_db(const char *dbname) const
 {
diff --git a/src/observer/storage/table/heap_table_engine.cpp b/src/observer/storage/table/heap_table_engine.cpp
index 6348c60e8..33071d779 100644
--- a/src/observer/storage/table/heap_table_engine.cpp
+++ b/src/observer/storage/table/heap_table_engine.cpp
@@ -103,6 +103,41 @@ RC HeapTableEngine::delete_record(const Record &record)
   return rc;
 }
 
+RC HeapTableEngine::update_record_with_trx(const Record &old_record, const Record &new_record, Trx *trx)
+{
+  if (old_record.rid() != new_record.rid() || old_record.len() != new_record.len()) {
+    return RC::INVALID_ARGUMENT;
+  }
+
+  RC rc = delete_entry_of_indexes(old_record.data(), old_record.rid(), true);
+  if (OB_FAIL(rc)) {
+    LOG_WARN("failed to delete old index entries while updating. table=%s, rid=%s, rc=%s",
+        table_meta_->name(), old_record.rid().to_string().c_str(), strrc(rc));
+    return rc;
+  }
+
+  rc = insert_entry_of_indexes(new_record.data(), new_record.rid());
+  if (OB_FAIL(rc)) {
+    delete_entry_of_indexes(new_record.data(), new_record.rid(), false);
+    insert_entry_of_indexes(old_record.data(), old_record.rid());
+    LOG_WARN("failed to insert new index entries while updating. table=%s, rid=%s, rc=%s",
+        table_meta_->name(), old_record.rid().to_string().c_str(), strrc(rc));
+    return rc;
+  }
+
+  rc = record_handler_->visit_record(old_record.rid(), [&new_record](Record &record) {
+    memcpy(record.data(), new_record.data(), new_record.len());
+    return true;
+  });
+  if (OB_FAIL(rc)) {
+    delete_entry_of_indexes(new_record.data(), new_record.rid(), false);
+    insert_entry_of_indexes(old_record.data(), old_record.rid());
+    LOG_WARN("failed to update record data. table=%s, rid=%s, rc=%s",
+        table_meta_->name(), old_record.rid().to_string().c_str(), strrc(rc));
+  }
+  return rc;
+}
+
 RC HeapTableEngine::get_record_scanner(RecordScanner *&scanner, Trx *trx, ReadWriteMode mode)
 {
   scanner = new HeapRecordScanner(table_, *data_buffer_pool_, trx, db_->log_handler(), mode, nullptr);
diff --git a/src/observer/storage/table/heap_table_engine.h b/src/observer/storage/table/heap_table_engine.h
index 6726f86dc..4399fb29b 100644
--- a/src/observer/storage/table/heap_table_engine.h
+++ b/src/observer/storage/table/heap_table_engine.h
@@ -31,10 +31,7 @@ class HeapTableEngine : public TableEngine
   RC delete_record(const Record &record) override;
   RC insert_record_with_trx(Record &record, Trx *trx) override { return RC::UNSUPPORTED; }
   RC delete_record_with_trx(const Record &record, Trx *trx) override { return RC::UNSUPPORTED; }
-  RC update_record_with_trx(const Record &old_record, const Record &new_record, Trx *trx) override
-  {
-    return RC::UNSUPPORTED;
-  }
+  RC update_record_with_trx(const Record &old_record, const Record &new_record, Trx *trx) override;
   RC get_record(const RID &rid, Record &record) override;
 
   RC create_index(Trx *trx, const FieldMeta *field_meta, const char *index_name) override;
diff --git a/src/observer/storage/table/table.cpp b/src/observer/storage/table/table.cpp
index 14093fc06..cd1c10497 100644
--- a/src/observer/storage/table/table.cpp
+++ b/src/observer/storage/table/table.cpp
@@ -264,6 +264,33 @@ RC Table::set_value_to_record(char *record_data, const Value &value, const Field
   return RC::SUCCESS;
 }
 
+RC Table::make_updated_record(
+    const Record &old_record, const FieldMeta &field_meta, const Value &value, Record &new_record)
+{
+  if (old_record.len() != table_meta_.record_size()) {
+    return RC::INVALID_ARGUMENT;
+  }
+
+  Value real_value;
+  if (value.attr_type() == field_meta.type()) {
+    real_value = value;
+  } else {
+    RC rc = Value::cast_to(value, field_meta.type(), real_value);
+    if (OB_FAIL(rc)) {
+      return rc;
+    }
+  }
+
+  RC rc = new_record.copy_data(old_record.data(), old_record.len());
+  if (OB_FAIL(rc)) {
+    return rc;
+  }
+  new_record.set_rid(old_record.rid());
+
+  memset(new_record.data() + field_meta.offset(), 0, field_meta.len());
+  return set_value_to_record(new_record.data(), real_value, &field_meta);
+}
+
 RC Table::get_record_scanner(RecordScanner *&scanner, Trx *trx, ReadWriteMode mode)
 {
   return engine_->get_record_scanner(scanner, trx, mode);
diff --git a/src/observer/storage/table/table.h b/src/observer/storage/table/table.h
index 0e5f89014..81c4b0d1a 100644
--- a/src/observer/storage/table/table.h
+++ b/src/observer/storage/table/table.h
@@ -78,6 +78,9 @@ class Table
    */
   RC make_record(int value_num, const Value *values, Record &record);
 
+  RC make_updated_record(
+      const Record &old_record, const FieldMeta &field_meta, const Value &value, Record &new_record);
+
   /**
    * @brief 在当前的表中插入一条记录
    * @details 在表文件和索引中插入关联数据。这里只管在表中插入数据，不关心事务相关操作。
diff --git a/src/observer/storage/trx/vacuous_trx.cpp b/src/observer/storage/trx/vacuous_trx.cpp
index 737f243b7..3a817607c 100644
--- a/src/observer/storage/trx/vacuous_trx.cpp
+++ b/src/observer/storage/trx/vacuous_trx.cpp
@@ -34,6 +34,11 @@ RC VacuousTrx::insert_record(Table *table, Record &record) { return table->inser
 
 RC VacuousTrx::delete_record(Table *table, Record &record) { return table->delete_record(record); }
 
+RC VacuousTrx::update_record(Table *table, Record &old_record, Record &new_record)
+{
+  return table->update_record_with_trx(old_record, new_record, this);
+}
+
 RC VacuousTrx::visit_record(Table *table, Record &record, ReadWriteMode) { return RC::SUCCESS; }
 
 RC VacuousTrx::start_if_need() { return RC::SUCCESS; }
diff --git a/src/observer/storage/trx/vacuous_trx.h b/src/observer/storage/trx/vacuous_trx.h
index 92adde90f..eb99a5f14 100644
--- a/src/observer/storage/trx/vacuous_trx.h
+++ b/src/observer/storage/trx/vacuous_trx.h
@@ -45,7 +45,7 @@ class VacuousTrx : public Trx
 
   RC insert_record(Table *table, Record &record) override;
   RC delete_record(Table *table, Record &record) override;
-  RC update_record(Table *table, Record &old_record, Record &new_record) override { return RC::UNIMPLEMENTED; }
+  RC update_record(Table *table, Record &old_record, Record &new_record) override;
   RC visit_record(Table *table, Record &record, ReadWriteMode mode) override;
   RC start_if_need() override;
   RC commit() override;
@@ -63,4 +63,4 @@ class VacuousTrxLogReplayer : public LogReplayer
   virtual ~VacuousTrxLogReplayer() = default;
 
   RC replay(const LogEntry &) override { return RC::SUCCESS; }
-};
\ No newline at end of file
+};