diff --git a/deps/common/lang/array.h b/deps/common/lang/array.h
new file mode 100644
index 000000000..3f92304f9
--- /dev/null
+++ b/deps/common/lang/array.h
@@ -0,0 +1,15 @@
+/* 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 <array>
+
+using std::array;
\ No newline at end of file
diff --git a/deps/common/lang/comparator.cpp b/deps/common/lang/comparator.cpp
index a7c28d7ae..1f76a2af8 100644
--- a/deps/common/lang/comparator.cpp
+++ b/deps/common/lang/comparator.cpp
@@ -53,15 +53,15 @@ int compare_string(void *arg1, int arg1_max_length, void *arg2, int arg2_max_len
   int         maxlen = min(arg1_max_length, arg2_max_length);
   int         result = strncmp(s1, s2, maxlen);
   if (0 != result) {
-    return result;
+    return result < 0 ? -1 : 1;
   }
 
   if (arg1_max_length > maxlen) {
-    return s1[maxlen] - 0;
+    return 1;
   }
 
   if (arg2_max_length > maxlen) {
-    return 0 - s2[maxlen];
+    return -1;
   }
   return 0;
 }
diff --git a/deps/common/lang/string.h b/deps/common/lang/string.h
index 31683cf5b..9bdf4aea3 100644
--- a/deps/common/lang/string.h
+++ b/deps/common/lang/string.h
@@ -18,6 +18,7 @@ See the Mulan PSL v2 for more details. */
 #include <cxxabi.h>
 #include <signal.h>
 #include <stdlib.h>
+#include <string.h>
 
 #include <cstdlib>
 #include <string>
diff --git a/src/observer/common/rc.h b/src/observer/common/rc.h
index ed592dbc3..d465c7cfb 100644
--- a/src/observer/common/rc.h
+++ b/src/observer/common/rc.h
@@ -22,7 +22,7 @@ See the Mulan PSL v2 for more details. */
 #define DEFINE_RCS                       \
   DEFINE_RC(SUCCESS)                     \
   DEFINE_RC(INVALID_ARGUMENT)            \
-  DEFINE_RC(UNIMPLENMENT)                \
+  DEFINE_RC(UNIMPLEMENTED)               \
   DEFINE_RC(SQL_SYNTAX)                  \
   DEFINE_RC(INTERNAL)                    \
   DEFINE_RC(NOMEM)                       \
@@ -75,7 +75,8 @@ See the Mulan PSL v2 for more details. */
   DEFINE_RC(VARIABLE_NOT_VALID)          \
   DEFINE_RC(LOGBUF_FULL)                 \
   DEFINE_RC(LOG_FILE_FULL)               \
-  DEFINE_RC(LOG_ENTRY_INVALID)
+  DEFINE_RC(LOG_ENTRY_INVALID)           \
+  DEFINE_RC(UNSUPPORTED)
 
 enum class RC
 {
diff --git a/src/observer/common/type/attr_type.cpp b/src/observer/common/type/attr_type.cpp
new file mode 100644
index 000000000..88206cfba
--- /dev/null
+++ b/src/observer/common/type/attr_type.cpp
@@ -0,0 +1,33 @@
+/* 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/lang/string.h"
+#include "common/type/attr_type.h"
+
+const char *ATTR_TYPE_NAME[] = {"undefined", "chars", "ints", "floats", "booleans"};
+
+const char *attr_type_to_string(AttrType type)
+{
+  if (type >= AttrType::UNDEFINED && type < AttrType::MAXTYPE) {
+    return ATTR_TYPE_NAME[static_cast<int>(type)];
+  }
+  return "unknown";
+}
+
+AttrType attr_type_from_string(const char *s)
+{
+  for (unsigned int i = 0; i < sizeof(ATTR_TYPE_NAME) / sizeof(ATTR_TYPE_NAME[0]); i++) {
+    if (0 == strcasecmp(ATTR_TYPE_NAME[i], s)) {
+      return (AttrType)i;
+    }
+  }
+  return AttrType::UNDEFINED;
+}
diff --git a/src/observer/common/type/attr_type.h b/src/observer/common/type/attr_type.h
new file mode 100644
index 000000000..55a0e64ef
--- /dev/null
+++ b/src/observer/common/type/attr_type.h
@@ -0,0 +1,28 @@
+/* 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
+
+/**
+ * @brief 属性的类型
+ * @details AttrType 枚举列出了属性的各种数据类型。
+ */
+enum class AttrType
+{
+  UNDEFINED,
+  CHARS,     ///< 字符串类型
+  INTS,      ///< 整数类型(4字节)
+  FLOATS,    ///< 浮点数类型(4字节)
+  BOOLEANS,  ///< boolean类型，当前不是由parser解析出来的，是程序内部使用的
+  MAXTYPE,   ///< 请在 UNDEFINED 与 MAXTYPE 之间增加新类型
+};
+
+const char *attr_type_to_string(AttrType type);
+AttrType    attr_type_from_string(const char *s);
diff --git a/src/observer/common/type/char_type.cpp b/src/observer/common/type/char_type.cpp
new file mode 100644
index 000000000..6c93114a9
--- /dev/null
+++ b/src/observer/common/type/char_type.cpp
@@ -0,0 +1,51 @@
+/* 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/lang/comparator.h"
+#include "common/log/log.h"
+#include "common/type/char_type.h"
+#include "common/value.h"
+
+int CharType::compare(const Value &left, const Value &right) const
+{
+  ASSERT(left.attr_type() == AttrType::CHARS && right.attr_type() == AttrType::CHARS, "invalid type");
+  return common::compare_string(
+      (void *)left.value_.pointer_value_, left.length_, (void *)right.value_.pointer_value_, right.length_);
+}
+
+RC CharType::set_value_from_str(Value &val, const string &data) const
+{
+  val.set_string(data.c_str());
+  return RC::SUCCESS;
+}
+
+RC CharType::cast_to(const Value &val, AttrType type, Value &result) const
+{
+  switch (type) {
+    default: return RC::UNIMPLEMENTED;
+  }
+  return RC::SUCCESS;
+}
+
+int CharType::cast_cost(AttrType type)
+{
+  if (type == AttrType::CHARS) {
+    return 0;
+  }
+  return INT32_MAX;
+}
+
+RC CharType::to_string(const Value &val, string &result) const
+{
+  stringstream ss;
+  ss << val.value_.pointer_value_;
+  result = ss.str();
+  return RC::SUCCESS;
+}
\ No newline at end of file
diff --git a/src/observer/common/type/char_type.h b/src/observer/common/type/char_type.h
new file mode 100644
index 000000000..a56101438
--- /dev/null
+++ b/src/observer/common/type/char_type.h
@@ -0,0 +1,36 @@
+/* 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/rc.h"
+#include "common/type/data_type.h"
+
+/**
+ * @brief 固定长度的字符串类型
+ * @ingroup DataType
+ */
+class CharType : public DataType
+{
+public:
+  CharType() : DataType(AttrType::CHARS) {}
+
+  virtual ~CharType() = default;
+
+  int compare(const Value &left, const Value &right) const override;
+
+  RC cast_to(const Value &val, AttrType type, Value &result) const override;
+
+  RC set_value_from_str(Value &val, const string &data) const override;
+
+  int cast_cost(AttrType type) override;
+
+  RC to_string(const Value &val, string &result) const override;
+};
\ No newline at end of file
diff --git a/src/observer/common/type/data_type.cpp b/src/observer/common/type/data_type.cpp
new file mode 100644
index 000000000..021a30e82
--- /dev/null
+++ b/src/observer/common/type/data_type.cpp
@@ -0,0 +1,22 @@
+/* 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/char_type.h"
+#include "common/type/float_type.h"
+#include "common/type/integer_type.h"
+#include "common/type/data_type.h"
+
+array<unique_ptr<DataType>, static_cast<int>(AttrType::MAXTYPE)> DataType::type_instances_ = {
+    make_unique<DataType>(AttrType::UNDEFINED),
+    make_unique<CharType>(),
+    make_unique<IntegerType>(),
+    make_unique<FloatType>(),
+    make_unique<DataType>(AttrType::BOOLEANS),
+};
\ No newline at end of file
diff --git a/src/observer/common/type/data_type.h b/src/observer/common/type/data_type.h
new file mode 100644
index 000000000..3b8c9fde9
--- /dev/null
+++ b/src/observer/common/type/data_type.h
@@ -0,0 +1,102 @@
+/* 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/lang/array.h"
+#include "common/lang/memory.h"
+#include "common/lang/string.h"
+#include "common/rc.h"
+#include "common/type/attr_type.h"
+
+class Value;
+
+/**
+ * @brief 定义了数据类型相关的操作，比如比较运算、算术运算等
+ * @defgroup 数据类型
+ * @details 数据类型定义的算术运算中，比如 add、subtract 等，将按照当前数据类型设置最终结果值的类型。
+ * 参与运算的参数类型不一定相同，不同的类型进行运算是否能够支持需要参考各个类型的实现。
+ */
+class DataType
+{
+public:
+  explicit DataType(AttrType attr_type) : attr_type_(attr_type) {}
+
+  virtual ~DataType() = default;
+
+  inline static DataType *type_instance(AttrType attr_type)
+  {
+    return type_instances_.at(static_cast<int>(attr_type)).get();
+  }
+
+  inline AttrType get_attr_type() const { return attr_type_; }
+
+  /**
+   * @return
+   *  -1 表示 left < right
+   *  0 表示 left = right
+   *  1 表示 left > right
+   *  INT32_MAX 表示未实现的比较
+   */
+  virtual int compare(const Value &left, const Value &right) const { return INT32_MAX; }
+
+  /**
+   * @brief 计算 left + right，并将结果保存到 result 中
+   */
+  virtual RC add(const Value &left, const Value &right, Value &result) const { return RC::UNSUPPORTED; }
+
+  /**
+   * @brief 计算 left - right，并将结果保存到 result 中
+   */
+  virtual RC subtract(const Value &left, const Value &right, Value &result) const { return RC::UNSUPPORTED; }
+
+  /**
+   * @brief 计算 left * right，并将结果保存到 result 中
+   */
+  virtual RC multiply(const Value &left, const Value &right, Value &result) const { return RC::UNSUPPORTED; }
+
+  /**
+   * @brief 计算 left / right，并将结果保存到 result 中
+   */
+  virtual RC divide(const Value &left, const Value &right, Value &result) const { return RC::UNSUPPORTED; }
+
+  /**
+   * @brief 计算 -val，并将结果保存到 result 中
+   */
+  virtual RC negative(const Value &val, Value &result) const { return RC::UNSUPPORTED; }
+
+  /**
+   * @brief 将 val 转换为 type 类型，并将结果保存到 result 中
+   */
+  virtual RC cast_to(const Value &val, AttrType type, Value &result) const { return RC::UNSUPPORTED; }
+
+  /**
+   * @brief 将 val 转换为 string，并将结果保存到 result 中
+   */
+  virtual RC to_string(const Value &val, string &result) const { return RC::UNSUPPORTED; }
+
+  /**
+   * @brief 计算从 type 到 attr_type 的隐式转换的 cost，如果无法转换，返回 INT32_MAX
+   */
+  virtual int cast_cost(AttrType type)
+  {
+    if (type == attr_type_) {
+      return 0;
+    }
+    return INT32_MAX;
+  }
+
+  virtual RC set_value_from_str(Value &val, const string &data) const { return RC::UNSUPPORTED; }
+
+protected:
+  AttrType attr_type_;
+
+  static array<unique_ptr<DataType>, static_cast<int>(AttrType::MAXTYPE)> type_instances_;
+};
diff --git a/src/observer/common/type/float_type.cpp b/src/observer/common/type/float_type.cpp
new file mode 100644
index 000000000..fcf89a7cc
--- /dev/null
+++ b/src/observer/common/type/float_type.cpp
@@ -0,0 +1,85 @@
+/* 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/lang/comparator.h"
+#include "common/lang/sstream.h"
+#include "common/log/log.h"
+#include "common/type/float_type.h"
+#include "common/value.h"
+#include "common/lang/limits.h"
+#include "common/value.h"
+
+int FloatType::compare(const Value &left, const Value &right) const
+{
+  ASSERT(left.attr_type() == AttrType::FLOATS, "left type is not integer");
+  ASSERT(right.attr_type() == AttrType::INTS || right.attr_type() == AttrType::FLOATS, "right type is not numeric");
+  float left_val  = left.get_float();
+  float right_val = right.get_float();
+  return common::compare_float((void *)&left_val, (void *)&right_val);
+}
+
+RC FloatType::add(const Value &left, const Value &right, Value &result) const
+{
+  result.set_float(left.get_float() + right.get_float());
+  return RC::SUCCESS;
+}
+RC FloatType::subtract(const Value &left, const Value &right, Value &result) const
+{
+  result.set_float(left.get_float() - right.get_float());
+  return RC::SUCCESS;
+}
+RC FloatType::multiply(const Value &left, const Value &right, Value &result) const
+{
+  result.set_float(left.get_float() * right.get_float());
+  return RC::SUCCESS;
+}
+
+RC FloatType::divide(const Value &left, const Value &right, Value &result) const
+{
+  if (right.get_float() > -EPSILON && right.get_float() < EPSILON) {
+    // NOTE:
+    // 设置为浮点数最大值是不正确的。通常的做法是设置为NULL，但是当前的miniob没有NULL概念，所以这里设置为浮点数最大值。
+    result.set_float(numeric_limits<float>::max());
+  } else {
+    result.set_float(left.get_float() / right.get_float());
+  }
+  return RC::SUCCESS;
+}
+
+RC FloatType::negative(const Value &val, Value &result) const
+{
+  result.set_float(-val.get_float());
+  return RC::SUCCESS;
+}
+
+RC FloatType::set_value_from_str(Value &val, const string &data) const
+{
+  RC                rc = RC::SUCCESS;
+  stringstream deserialize_stream;
+  deserialize_stream.clear();
+  deserialize_stream.str(data);
+
+  float float_value;
+  deserialize_stream >> float_value;
+  if (!deserialize_stream || !deserialize_stream.eof()) {
+    rc = RC::SCHEMA_FIELD_TYPE_MISMATCH;
+  } else {
+    val.set_float(float_value);
+  }
+  return rc;
+}
+
+RC FloatType::to_string(const Value &val, string &result) const
+{
+  stringstream ss;
+  ss << common::double_to_str(val.value_.float_value_);
+  result = ss.str();
+  return RC::SUCCESS;
+}
diff --git a/src/observer/common/type/float_type.h b/src/observer/common/type/float_type.h
new file mode 100644
index 000000000..fd4a7fcb5
--- /dev/null
+++ b/src/observer/common/type/float_type.h
@@ -0,0 +1,36 @@
+/* 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"
+
+/**
+ * @brief 浮点型数据类型
+ * @ingroup DataType
+ */
+class FloatType : public DataType
+{
+public:
+  FloatType() : DataType(AttrType::FLOATS) {}
+  virtual ~FloatType() = default;
+
+  int compare(const Value &left, const Value &right) const override;
+
+  RC add(const Value &left, const Value &right, Value &result) const override;
+  RC subtract(const Value &left, const Value &right, Value &result) const override;
+  RC multiply(const Value &left, const Value &right, Value &result) const override;
+  RC divide(const Value &left, const Value &right, Value &result) const override;
+  RC negative(const Value &val, Value &result) const override;
+
+  RC set_value_from_str(Value &val, const string &data) const override;
+
+  RC to_string(const Value &val, string &result) const override;
+};
\ No newline at end of file
diff --git a/src/observer/common/type/integer_type.cpp b/src/observer/common/type/integer_type.cpp
new file mode 100644
index 000000000..34931e0bb
--- /dev/null
+++ b/src/observer/common/type/integer_type.cpp
@@ -0,0 +1,75 @@
+/* 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/lang/comparator.h"
+#include "common/lang/sstream.h"
+#include "common/log/log.h"
+#include "common/type/integer_type.h"
+#include "common/value.h"
+
+int IntegerType::compare(const Value &left, const Value &right) const
+{
+  ASSERT(left.attr_type() == AttrType::INTS, "left type is not integer");
+  ASSERT(right.attr_type() == AttrType::INTS || right.attr_type() == AttrType::FLOATS, "right type is not numeric");
+  if (right.attr_type() == AttrType::INTS) {
+    return common::compare_int((void *)&left.value_.int_value_, (void *)&right.value_.int_value_);
+  } else if (right.attr_type() == AttrType::FLOATS) {
+    return common::compare_float((void *)&left.value_.int_value_, (void *)&right.value_.int_value_);
+  }
+  return INT32_MAX;
+}
+
+RC IntegerType::add(const Value &left, const Value &right, Value &result) const
+{
+  result.set_int(left.get_int() + right.get_int());
+  return RC::SUCCESS;
+}
+
+RC IntegerType::subtract(const Value &left, const Value &right, Value &result) const
+{
+  result.set_int(left.get_int() - right.get_int());
+  return RC::SUCCESS;
+}
+
+RC IntegerType::multiply(const Value &left, const Value &right, Value &result) const
+{
+  result.set_int(left.get_int() * right.get_int());
+  return RC::SUCCESS;
+}
+
+RC IntegerType::negative(const Value &val, Value &result) const
+{
+  result.set_int(-val.get_int());
+  return RC::SUCCESS;
+}
+
+RC IntegerType::set_value_from_str(Value &val, const string &data) const
+{
+  RC                rc = RC::SUCCESS;
+  stringstream deserialize_stream;
+  deserialize_stream.clear();  // 清理stream的状态，防止多次解析出现异常
+  deserialize_stream.str(data);
+  int int_value;
+  deserialize_stream >> int_value;
+  if (!deserialize_stream || !deserialize_stream.eof()) {
+    rc = RC::SCHEMA_FIELD_TYPE_MISMATCH;
+  } else {
+    val.set_int(int_value);
+  }
+  return rc;
+}
+
+RC IntegerType::to_string(const Value &val, string &result) const
+{
+  stringstream ss;
+  ss << val.value_.int_value_;
+  result = ss.str();
+  return RC::SUCCESS;
+}
\ No newline at end of file
diff --git a/src/observer/common/type/integer_type.h b/src/observer/common/type/integer_type.h
new file mode 100644
index 000000000..dd8c48e5c
--- /dev/null
+++ b/src/observer/common/type/integer_type.h
@@ -0,0 +1,35 @@
+/* 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"
+
+/**
+ * @brief 整型类型
+ * @ingroup DataType
+ */
+class IntegerType : public DataType
+{
+public:
+  IntegerType() : DataType(AttrType::INTS) {}
+  virtual ~IntegerType() {}
+
+  int compare(const Value &left, const Value &right) const override;
+
+  RC add(const Value &left, const Value &right, Value &result) const override;
+  RC subtract(const Value &left, const Value &right, Value &result) const override;
+  RC multiply(const Value &left, const Value &right, Value &result) const override;
+  RC negative(const Value &val, Value &result) const override;
+
+  RC set_value_from_str(Value &val, const string &data) const override;
+
+  RC to_string(const Value &val, string &result) const override;
+};
\ No newline at end of file
diff --git a/src/observer/common/type/vector_type.h b/src/observer/common/type/vector_type.h
new file mode 100644
index 000000000..e06d8057a
--- /dev/null
+++ b/src/observer/common/type/vector_type.h
@@ -0,0 +1,13 @@
+/* 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"
\ No newline at end of file
diff --git a/src/observer/common/value.cpp b/src/observer/common/value.cpp
new file mode 100644
index 000000000..b8f04914b
--- /dev/null
+++ b/src/observer/common/value.cpp
@@ -0,0 +1,329 @@
+/* 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. */
+
+//
+// Created by WangYunlai on 2023/06/28.
+//
+
+#include "common/value.h"
+
+#include "common/lang/comparator.h"
+#include "common/lang/exception.h"
+#include "common/lang/sstream.h"
+#include "common/lang/string.h"
+#include "common/log/log.h"
+
+Value::Value(int val) { set_int(val); }
+
+Value::Value(float val) { set_float(val); }
+
+Value::Value(bool val) { set_boolean(val); }
+
+Value::Value(const char *s, int len /*= 0*/) { set_string(s, len); }
+
+Value::Value(const Value &other)
+{
+  this->attr_type_ = other.attr_type_;
+  this->length_    = other.length_;
+  this->own_data_  = other.own_data_;
+  switch (this->attr_type_) {
+    case AttrType::CHARS: {
+      set_string_from_other(other);
+    } break;
+
+    default: {
+      this->value_ = other.value_;
+    } break;
+  }
+}
+
+Value::Value(Value &&other)
+{
+  this->attr_type_ = other.attr_type_;
+  this->length_    = other.length_;
+  this->own_data_  = other.own_data_;
+  this->value_     = other.value_;
+  other.own_data_  = false;
+  other.length_    = 0;
+}
+
+Value &Value::operator=(const Value &other)
+{
+  if (this == &other) {
+    return *this;
+  }
+  reset();
+  this->attr_type_ = other.attr_type_;
+  this->length_    = other.length_;
+  this->own_data_  = other.own_data_;
+  switch (this->attr_type_) {
+    case AttrType::CHARS: {
+      set_string_from_other(other);
+    } break;
+
+    default: {
+      this->value_ = other.value_;
+    } break;
+  }
+  return *this;
+}
+
+Value &Value::operator=(Value &&other)
+{
+  if (this == &other) {
+    return *this;
+  }
+  reset();
+  this->attr_type_ = other.attr_type_;
+  this->length_    = other.length_;
+  this->own_data_  = other.own_data_;
+  this->value_     = other.value_;
+  other.own_data_  = false;
+  other.length_    = 0;
+  return *this;
+}
+
+void Value::reset()
+{
+  switch (attr_type_) {
+    case AttrType::CHARS:
+      if (own_data_ && value_.pointer_value_ != nullptr) {
+        delete[] value_.pointer_value_;
+        value_.pointer_value_ = nullptr;
+      }
+      break;
+    default: break;
+  }
+
+  attr_type_ = AttrType::UNDEFINED;
+  length_    = 0;
+  own_data_  = false;
+}
+
+void Value::set_data(char *data, int length)
+{
+  switch (attr_type_) {
+    case AttrType::CHARS: {
+      set_string(data, length);
+    } break;
+    case AttrType::INTS: {
+      value_.int_value_ = *(int *)data;
+      length_           = length;
+    } break;
+    case AttrType::FLOATS: {
+      value_.float_value_ = *(float *)data;
+      length_             = length;
+    } break;
+    case AttrType::BOOLEANS: {
+      value_.bool_value_ = *(int *)data != 0;
+      length_            = length;
+    } break;
+    default: {
+      LOG_WARN("unknown data type: %d", attr_type_);
+    } break;
+  }
+}
+
+void Value::set_int(int val)
+{
+  reset();
+  attr_type_        = AttrType::INTS;
+  value_.int_value_ = val;
+  length_           = sizeof(val);
+}
+
+void Value::set_float(float val)
+{
+  reset();
+  attr_type_          = AttrType::FLOATS;
+  value_.float_value_ = val;
+  length_             = sizeof(val);
+}
+void Value::set_boolean(bool val)
+{
+  reset();
+  attr_type_         = AttrType::BOOLEANS;
+  value_.bool_value_ = val;
+  length_            = sizeof(val);
+}
+
+void Value::set_string(const char *s, int len /*= 0*/)
+{
+  reset();
+  attr_type_ = AttrType::CHARS;
+  if (s == nullptr) {
+    value_.pointer_value_ = nullptr;
+    length_               = 0;
+  } else {
+    own_data_ = true;
+    if (len > 0) {
+      len = strnlen(s, len);
+    } else {
+      len = strlen(s);
+    }
+    value_.pointer_value_ = new char[len + 1];
+    length_               = len;
+    memcpy(value_.pointer_value_, s, len);
+    value_.pointer_value_[len] = '\0';
+  }
+}
+
+void Value::set_value(const Value &value)
+{
+  switch (value.attr_type_) {
+    case AttrType::INTS: {
+      set_int(value.get_int());
+    } break;
+    case AttrType::FLOATS: {
+      set_float(value.get_float());
+    } break;
+    case AttrType::CHARS: {
+      set_string(value.get_string().c_str());
+    } break;
+    case AttrType::BOOLEANS: {
+      set_boolean(value.get_boolean());
+    } break;
+    default: {
+      ASSERT(false, "got an invalid value type");
+    } break;
+  }
+}
+
+void Value::set_string_from_other(const Value &other)
+{
+  ASSERT(attr_type_ == AttrType::CHARS, "attr type is not CHARS");
+  if (own_data_ && other.value_.pointer_value_ != nullptr && length_ != 0) {
+    this->value_.pointer_value_ = new char[this->length_ + 1];
+    memcpy(this->value_.pointer_value_, other.value_.pointer_value_, this->length_);
+    this->value_.pointer_value_[this->length_] = '\0';
+  }
+}
+
+const char *Value::data() const
+{
+  switch (attr_type_) {
+    case AttrType::CHARS: {
+      return value_.pointer_value_;
+    } break;
+    default: {
+      return (const char *)&value_;
+    } break;
+  }
+}
+
+string Value::to_string() const
+{
+  string res;
+  RC     rc = DataType::type_instance(this->attr_type_)->to_string(*this, res);
+  if (OB_FAIL(rc)) {
+    LOG_WARN("failed to convert value to string. type=%s", attr_type_to_string(this->attr_type_));
+    return "";
+  }
+  return res;
+}
+
+int Value::compare(const Value &other) const { return DataType::type_instance(this->attr_type_)->compare(*this, other); }
+
+int Value::get_int() const
+{
+  switch (attr_type_) {
+    case AttrType::CHARS: {
+      try {
+        return (int)(std::stol(value_.pointer_value_));
+      } catch (exception const &ex) {
+        LOG_TRACE("failed to convert string to number. s=%s, ex=%s", value_.pointer_value_, ex.what());
+        return 0;
+      }
+    }
+    case AttrType::INTS: {
+      return value_.int_value_;
+    }
+    case AttrType::FLOATS: {
+      return (int)(value_.float_value_);
+    }
+    case AttrType::BOOLEANS: {
+      return (int)(value_.bool_value_);
+    }
+    default: {
+      LOG_WARN("unknown data type. type=%d", attr_type_);
+      return 0;
+    }
+  }
+  return 0;
+}
+
+float Value::get_float() const
+{
+  switch (attr_type_) {
+    case AttrType::CHARS: {
+      try {
+        return std::stof(value_.pointer_value_);
+      } catch (exception const &ex) {
+        LOG_TRACE("failed to convert string to float. s=%s, ex=%s", value_.pointer_value_, ex.what());
+        return 0.0;
+      }
+    } break;
+    case AttrType::INTS: {
+      return float(value_.int_value_);
+    } break;
+    case AttrType::FLOATS: {
+      return value_.float_value_;
+    } break;
+    case AttrType::BOOLEANS: {
+      return float(value_.bool_value_);
+    } break;
+    default: {
+      LOG_WARN("unknown data type. type=%d", attr_type_);
+      return 0;
+    }
+  }
+  return 0;
+}
+
+string Value::get_string() const { return this->to_string(); }
+
+bool Value::get_boolean() const
+{
+  switch (attr_type_) {
+    case AttrType::CHARS: {
+      try {
+        float val = std::stof(value_.pointer_value_);
+        if (val >= EPSILON || val <= -EPSILON) {
+          return true;
+        }
+
+        int int_val = std::stol(value_.pointer_value_);
+        if (int_val != 0) {
+          return true;
+        }
+
+        return value_.pointer_value_ != nullptr;
+      } catch (exception const &ex) {
+        LOG_TRACE("failed to convert string to float or integer. s=%s, ex=%s", value_.pointer_value_, ex.what());
+        return value_.pointer_value_ != nullptr;
+      }
+    } break;
+    case AttrType::INTS: {
+      return value_.int_value_ != 0;
+    } break;
+    case AttrType::FLOATS: {
+      float val = value_.float_value_;
+      return val >= EPSILON || val <= -EPSILON;
+    } break;
+    case AttrType::BOOLEANS: {
+      return value_.bool_value_;
+    } break;
+    default: {
+      LOG_WARN("unknown data type. type=%d", attr_type_);
+      return false;
+    }
+  }
+  return false;
+}
diff --git a/src/observer/common/value.h b/src/observer/common/value.h
new file mode 100644
index 000000000..ebbc70a3a
--- /dev/null
+++ b/src/observer/common/value.h
@@ -0,0 +1,132 @@
+/* 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. */
+
+//
+// Created by Wangyunlai 2023/6/27
+//
+
+#pragma once
+
+#include "common/lang/string.h"
+#include "common/lang/memory.h"
+#include "common/type/attr_type.h"
+#include "common/type/data_type.h"
+
+/**
+ * @brief 属性的值
+ * @ingroup DataType
+ * @details 与DataType，就是数据类型，配套完成各种算术运算、比较、类型转换等操作。这里同时记录了数据的值与类型。
+ * 当需要对值做运算时，建议使用类似 Value::add 的操作而不是 DataType::add。在进行运算前，应该设置好结果的类型，
+ * 比如进行两个INT类型的除法运算时，结果类型应该设置为FLOAT。
+ */
+class Value final
+{
+public:
+  friend class DataType;
+  friend class IntegerType;
+  friend class FloatType;
+  friend class BooleanType;
+  friend class CharType;
+
+  Value() = default;
+
+  ~Value() { reset(); }
+
+  Value(AttrType attr_type, char *data, int length = 4) : attr_type_(attr_type) { this->set_data(data, length); }
+
+  explicit Value(int val);
+  explicit Value(float val);
+  explicit Value(bool val);
+  explicit Value(const char *s, int len = 0);
+
+  Value(const Value &other);
+  Value(Value &&other);
+
+  Value &operator=(const Value &other);
+  Value &operator=(Value &&other);
+
+  void reset();
+
+  static RC add(const Value &left, const Value &right, Value &result)
+  {
+    return DataType::type_instance(result.attr_type())->add(left, right, result);
+  }
+
+  static RC subtract(const Value &left, const Value &right, Value &result)
+  {
+    return DataType::type_instance(result.attr_type())->subtract(left, right, result);
+  }
+
+  static RC multiply(const Value &left, const Value &right, Value &result)
+  {
+    return DataType::type_instance(result.attr_type())->multiply(left, right, result);
+  }
+
+  static RC divide(const Value &left, const Value &right, Value &result)
+  {
+    return DataType::type_instance(result.attr_type())->divide(left, right, result);
+  }
+
+  static RC negative(const Value &value, Value &result)
+  {
+    return DataType::type_instance(result.attr_type())->negative(value, result);
+  }
+
+  static RC cast_to(const Value &value, AttrType to_type, Value &result)
+  {
+    return DataType::type_instance(value.attr_type())->cast_to(value, to_type, result);
+  }
+
+  void set_type(AttrType type) { this->attr_type_ = type; }
+  void set_data(char *data, int length);
+  void set_data(const char *data, int length) { this->set_data(const_cast<char *>(data), length); }
+  void set_value(const Value &value);
+  void set_boolean(bool val);
+
+  string to_string() const;
+
+  int compare(const Value &other) const;
+
+  const char *data() const;
+
+  int      length() const { return length_; }
+  AttrType attr_type() const { return attr_type_; }
+
+public:
+  /**
+   * 获取对应的值
+   * 如果当前的类型与期望获取的类型不符，就会执行转换操作
+   */
+  int    get_int() const;
+  float  get_float() const;
+  string get_string() const;
+  bool   get_boolean() const;
+
+private:
+  void set_int(int val);
+  void set_float(float val);
+  void set_string(const char *s, int len = 0);
+  void set_string_from_other(const Value &other);
+
+private:
+  AttrType attr_type_ = AttrType::UNDEFINED;
+  int      length_    = 0;
+
+  union Val
+  {
+    int32_t int_value_;
+    float   float_value_;
+    bool    bool_value_;
+    char   *pointer_value_;
+  } value_ = {.int_value_ = 0};
+
+  /// 是否申请并占有内存, 目前对于 CHARS 类型 own_data_ 为true, 其余类型 own_data_ 为false
+  bool own_data_ = false;
+};
diff --git a/src/observer/net/sql_task_handler.cpp b/src/observer/net/sql_task_handler.cpp
index 1086c02db..90f0609b4 100644
--- a/src/observer/net/sql_task_handler.cpp
+++ b/src/observer/net/sql_task_handler.cpp
@@ -76,7 +76,7 @@ RC SqlTaskHandler::handle_sql(SQLStageEvent *sql_event)
   }
 
   rc = optimize_stage_.handle_request(sql_event);
-  if (rc != RC::UNIMPLENMENT && rc != RC::SUCCESS) {
+  if (rc != RC::UNIMPLEMENTED && rc != RC::SUCCESS) {
     LOG_TRACE("failed to do optimize. rc=%s", strrc(rc));
     return rc;
   }
diff --git a/src/observer/session/session_stage.cpp b/src/observer/session/session_stage.cpp
index 8ca555e53..eed271357 100644
--- a/src/observer/session/session_stage.cpp
+++ b/src/observer/session/session_stage.cpp
@@ -98,7 +98,7 @@ RC SessionStage::handle_sql(SQLStageEvent *sql_event)
   }
 
   rc = optimize_stage_.handle_request(sql_event);
-  if (rc != RC::UNIMPLENMENT && rc != RC::SUCCESS) {
+  if (rc != RC::UNIMPLEMENTED && rc != RC::SUCCESS) {
     LOG_TRACE("failed to do optimize. rc=%s", strrc(rc));
     return rc;
   }
diff --git a/src/observer/sql/executor/command_executor.cpp b/src/observer/sql/executor/command_executor.cpp
index c598940e3..0da5c4e0f 100644
--- a/src/observer/sql/executor/command_executor.cpp
+++ b/src/observer/sql/executor/command_executor.cpp
@@ -84,7 +84,7 @@ RC CommandExecutor::execute(SQLStageEvent *sql_event)
 
     default: {
       LOG_ERROR("unknown command: %d", static_cast<int>(stmt->type()));
-      rc = RC::UNIMPLENMENT;
+      rc = RC::UNIMPLEMENTED;
     } break;
   }
 
diff --git a/src/observer/sql/executor/load_data_executor.cpp b/src/observer/sql/executor/load_data_executor.cpp
index bdb685572..c3d6fbe67 100644
--- a/src/observer/sql/executor/load_data_executor.cpp
+++ b/src/observer/sql/executor/load_data_executor.cpp
@@ -61,45 +61,7 @@ RC insert_record_from_file(
     if (field->type() != AttrType::CHARS) {
       common::strip(file_value);
     }
-
-    switch (field->type()) {
-      case AttrType::INTS: {
-        deserialize_stream.clear();  // 清理stream的状态，防止多次解析出现异常
-        deserialize_stream.str(file_value);
-
-        int int_value;
-        deserialize_stream >> int_value;
-        if (!deserialize_stream || !deserialize_stream.eof()) {
-          errmsg << "need an integer but got '" << file_values[i] << "' (field index:" << i << ")";
-
-          rc = RC::SCHEMA_FIELD_TYPE_MISMATCH;
-        } else {
-          record_values[i].set_int(int_value);
-        }
-      }
-
-      break;
-      case AttrType::FLOATS: {
-        deserialize_stream.clear();
-        deserialize_stream.str(file_value);
-
-        float float_value;
-        deserialize_stream >> float_value;
-        if (!deserialize_stream || !deserialize_stream.eof()) {
-          errmsg << "need a float number but got '" << file_values[i] << "'(field index:" << i << ")";
-          rc = RC::SCHEMA_FIELD_TYPE_MISMATCH;
-        } else {
-          record_values[i].set_float(float_value);
-        }
-      } break;
-      case AttrType::CHARS: {
-        record_values[i].set_string(file_value.c_str());
-      } break;
-      default: {
-        errmsg << "Unsupported field type to loading: " << attr_type_to_string(field->type());
-        rc = RC::SCHEMA_FIELD_TYPE_MISMATCH;
-      } break;
-    }
+    rc = DataType::type_instance(field->type())->set_value_from_str(record_values[i], file_value);
   }
 
   if (RC::SUCCESS == rc) {
diff --git a/src/observer/sql/expr/aggregator.cpp b/src/observer/sql/expr/aggregator.cpp
index 35d8573ca..29ba468a7 100644
--- a/src/observer/sql/expr/aggregator.cpp
+++ b/src/observer/sql/expr/aggregator.cpp
@@ -25,18 +25,7 @@ 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()));
   
-  switch (value.attr_type())
-  {
-    case AttrType::INTS: {
-      value_.set_int(value.get_int() + value_.get_int());
-    } break;
-    case AttrType::FLOATS: {
-      value_.set_float(value.get_float() + value_.get_float());
-    } break;
-    default: {
-      return RC::INTERNAL;
-    }
-  }
+  Value::add(value, value_, value_);
   return RC::SUCCESS;
 }
 
diff --git a/src/observer/sql/expr/aggregator.h b/src/observer/sql/expr/aggregator.h
index 3b3ac74c1..c1af3d659 100644
--- a/src/observer/sql/expr/aggregator.h
+++ b/src/observer/sql/expr/aggregator.h
@@ -14,7 +14,7 @@ See the Mulan PSL v2 for more details. */
 
 #pragma once
 
-#include "sql/parser/value.h"
+#include "common/value.h"
 #include "common/rc.h"
 
 class Aggregator
diff --git a/src/observer/sql/expr/expression.cpp b/src/observer/sql/expr/expression.cpp
index 0d5b45333..1c1aa48c1 100644
--- a/src/observer/sql/expr/expression.cpp
+++ b/src/observer/sql/expr/expression.cpp
@@ -84,38 +84,30 @@ RC CastExpr::cast(const Value &value, Value &cast_value) const
     cast_value = value;
     return rc;
   }
-
-  switch (cast_type_) {
-    case AttrType::BOOLEANS: {
-      bool val = value.get_boolean();
-      cast_value.set_boolean(val);
-    } break;
-    default: {
-      rc = RC::INTERNAL;
-      LOG_WARN("unsupported convert from type %d to %d", child_->value_type(), cast_type_);
-    }
-  }
+  rc = Value::cast_to(value, cast_type_, cast_value);
   return rc;
 }
 
-RC CastExpr::get_value(const Tuple &tuple, Value &cell) const
+RC CastExpr::get_value(const Tuple &tuple, Value &result) const
 {
-  RC rc = child_->get_value(tuple, cell);
+  Value value;
+  RC rc = child_->get_value(tuple, value);
   if (rc != RC::SUCCESS) {
     return rc;
   }
 
-  return cast(cell, cell);
+  return cast(value, result);
 }
 
-RC CastExpr::try_get_value(Value &value) const
+RC CastExpr::try_get_value(Value &result) const
 {
+  Value value;
   RC rc = child_->try_get_value(value);
   if (rc != RC::SUCCESS) {
     return rc;
   }
 
-  return cast(value, value);
+  return cast(value, result);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -328,58 +320,27 @@ RC ArithmeticExpr::calc_value(const Value &left_value, const Value &right_value,
   RC rc = RC::SUCCESS;
 
   const AttrType target_type = value_type();
+  value.set_type(target_type);
 
   switch (arithmetic_type_) {
     case Type::ADD: {
-      if (target_type == AttrType::INTS) {
-        value.set_int(left_value.get_int() + right_value.get_int());
-      } else {
-        value.set_float(left_value.get_float() + right_value.get_float());
-      }
+      Value::add(left_value, right_value, value);
     } break;
 
     case Type::SUB: {
-      if (target_type == AttrType::INTS) {
-        value.set_int(left_value.get_int() - right_value.get_int());
-      } else {
-        value.set_float(left_value.get_float() - right_value.get_float());
-      }
+      Value::subtract(left_value, right_value, value);
     } break;
 
     case Type::MUL: {
-      if (target_type == AttrType::INTS) {
-        value.set_int(left_value.get_int() * right_value.get_int());
-      } else {
-        value.set_float(left_value.get_float() * right_value.get_float());
-      }
+      Value::multiply(left_value, right_value, value);
     } break;
 
     case Type::DIV: {
-      if (target_type == AttrType::INTS) {
-        if (right_value.get_int() == 0) {
-          // NOTE:
-          // 设置为整数最大值是不正确的。通常的做法是设置为NULL，但是当前的miniob没有NULL概念，所以这里设置为整数最大值。
-          value.set_int(numeric_limits<int>::max());
-        } else {
-          value.set_int(left_value.get_int() / right_value.get_int());
-        }
-      } else {
-        if (right_value.get_float() > -EPSILON && right_value.get_float() < EPSILON) {
-          // NOTE:
-          // 设置为浮点数最大值是不正确的。通常的做法是设置为NULL，但是当前的miniob没有NULL概念，所以这里设置为浮点数最大值。
-          value.set_float(numeric_limits<float>::max());
-        } else {
-          value.set_float(left_value.get_float() / right_value.get_float());
-        }
-      }
+      Value::divide(left_value, right_value, value);
     } break;
 
     case Type::NEGATIVE: {
-      if (target_type == AttrType::INTS) {
-        value.set_int(-left_value.get_int());
-      } else {
-        value.set_float(-left_value.get_float());
-      }
+      Value::negative(left_value, value);
     } break;
 
     default: {
@@ -404,7 +365,7 @@ RC ArithmeticExpr::execute_calc(
         binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, AddOperator>(
             (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
       } else {
-        rc = RC::UNIMPLENMENT;
+        rc = RC::UNIMPLEMENTED;
       }
     } break;
     case Type::SUB:
@@ -415,7 +376,7 @@ RC ArithmeticExpr::execute_calc(
         binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, SubtractOperator>(
             (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
       } else {
-        rc = RC::UNIMPLENMENT;
+        rc = RC::UNIMPLEMENTED;
       }
       break;
     case Type::MUL:
@@ -426,7 +387,7 @@ RC ArithmeticExpr::execute_calc(
         binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, MultiplyOperator>(
             (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
       } else {
-        rc = RC::UNIMPLENMENT;
+        rc = RC::UNIMPLEMENTED;
       }
       break;
     case Type::DIV:
@@ -437,7 +398,7 @@ RC ArithmeticExpr::execute_calc(
         binary_operator<LEFT_CONSTANT, RIGHT_CONSTANT, float, DivideOperator>(
             (float *)left.data(), (float *)right.data(), (float *)result.data(), result.capacity());
       } else {
-        rc = RC::UNIMPLENMENT;
+        rc = RC::UNIMPLEMENTED;
       }
       break;
     case Type::NEGATIVE:
@@ -447,10 +408,10 @@ RC ArithmeticExpr::execute_calc(
         unary_operator<LEFT_CONSTANT, float, NegateOperator>(
             (float *)left.data(), (float *)result.data(), result.capacity());
       } else {
-        rc = RC::UNIMPLENMENT;
+        rc = RC::UNIMPLEMENTED;
       }
       break;
-    default: rc = RC::UNIMPLENMENT; break;
+    default: rc = RC::UNIMPLEMENTED; break;
   }
   if (rc == RC::SUCCESS) {
     result.set_count(result.capacity());
diff --git a/src/observer/sql/expr/expression.h b/src/observer/sql/expr/expression.h
index 278b1a78b..7058fda4b 100644
--- a/src/observer/sql/expr/expression.h
+++ b/src/observer/sql/expr/expression.h
@@ -17,7 +17,7 @@ See the Mulan PSL v2 for more details. */
 #include <memory>
 #include <string>
 
-#include "sql/parser/value.h"
+#include "common/value.h"
 #include "storage/field/field.h"
 #include "sql/expr/aggregator.h"
 #include "storage/common/chunk.h"
@@ -81,12 +81,12 @@ class Expression
    * @brief 在没有实际运行的情况下，也就是无法获取tuple的情况下，尝试获取表达式的值
    * @details 有些表达式的值是固定的，比如ValueExpr，这种情况下可以直接获取值
    */
-  virtual RC try_get_value(Value &value) const { return RC::UNIMPLENMENT; }
+  virtual RC try_get_value(Value &value) const { return RC::UNIMPLEMENTED; }
 
   /**
    * @brief 从 `chunk` 中获取表达式的计算结果 `column`
    */
-  virtual RC get_column(Chunk &chunk, Column &column) { return RC::UNIMPLENMENT; }
+  virtual RC get_column(Chunk &chunk, Column &column) { return RC::UNIMPLEMENTED; }
 
   /**
    * @brief 表达式的类型
@@ -120,7 +120,7 @@ class Expression
   /**
    * @brief 用于 ComparisonExpr 获得比较结果 `select`。
    */
-  virtual RC eval(Chunk &chunk, std::vector<uint8_t> &select) { return RC::UNIMPLENMENT; }
+  virtual RC eval(Chunk &chunk, std::vector<uint8_t> &select) { return RC::UNIMPLEMENTED; }
 
 protected:
   /**
@@ -145,7 +145,7 @@ class StarExpr : public Expression
   ExprType type() const override { return ExprType::STAR; }
   AttrType value_type() const override { return AttrType::UNDEFINED; }
 
-  RC get_value(const Tuple &tuple, Value &value) const override { return RC::UNIMPLENMENT; }  // 不需要实现
+  RC get_value(const Tuple &tuple, Value &value) const override { return RC::UNIMPLEMENTED; }  // 不需要实现
 
   const char *table_name() const { return table_name_.c_str(); }
 
diff --git a/src/observer/sql/expr/expression_tuple.h b/src/observer/sql/expr/expression_tuple.h
index 36a26aeda..21f34fdb7 100644
--- a/src/observer/sql/expr/expression_tuple.h
+++ b/src/observer/sql/expr/expression_tuple.h
@@ -17,7 +17,7 @@ See the Mulan PSL v2 for more details. */
 #include <vector>
 
 #include "sql/expr/tuple.h"
-#include "sql/parser/value.h"
+#include "common/value.h"
 #include "common/rc.h"
 
 template <typename ExprPointerType>
diff --git a/src/observer/sql/expr/tuple.h b/src/observer/sql/expr/tuple.h
index 7290073ec..a813770c3 100644
--- a/src/observer/sql/expr/tuple.h
+++ b/src/observer/sql/expr/tuple.h
@@ -22,7 +22,7 @@ See the Mulan PSL v2 for more details. */
 #include "sql/expr/expression.h"
 #include "sql/expr/tuple_cell.h"
 #include "sql/parser/parse.h"
-#include "sql/parser/value.h"
+#include "common/value.h"
 #include "storage/record/record.h"
 
 class Table;
diff --git a/src/observer/sql/operator/explain_physical_operator.cpp b/src/observer/sql/operator/explain_physical_operator.cpp
index bda6ef329..da5573703 100644
--- a/src/observer/sql/operator/explain_physical_operator.cpp
+++ b/src/observer/sql/operator/explain_physical_operator.cpp
@@ -53,8 +53,7 @@ RC ExplainPhysicalOperator::next()
   generate_physical_plan();
 
   vector<Value> cells;
-  Value         cell;
-  cell.set_string(physical_plan_.c_str());
+  Value         cell(physical_plan_.c_str());
   cells.emplace_back(cell);
   tuple_.set_cells(cells);
   return RC::SUCCESS;
@@ -67,8 +66,7 @@ RC ExplainPhysicalOperator::next(Chunk &chunk)
   }
   generate_physical_plan();
 
-  Value         cell;
-  cell.set_string(physical_plan_.c_str());
+  Value         cell(physical_plan_.c_str());
   auto column = make_unique<Column>();
   column->init(cell);
   chunk.add_column(std::move(column), 0);
diff --git a/src/observer/sql/operator/group_by_vec_physical_operator.h b/src/observer/sql/operator/group_by_vec_physical_operator.h
index be8eef8e1..e78469b4e 100644
--- a/src/observer/sql/operator/group_by_vec_physical_operator.h
+++ b/src/observer/sql/operator/group_by_vec_physical_operator.h
@@ -27,9 +27,9 @@ class GroupByVecPhysicalOperator : public PhysicalOperator
 
   PhysicalOperatorType type() const override { return PhysicalOperatorType::GROUP_BY_VEC; }
 
-  RC open(Trx *trx) override { return RC::UNIMPLENMENT; }
-  RC next(Chunk &chunk) override { return RC::UNIMPLENMENT; }
-  RC close() override { return RC::UNIMPLENMENT; }
+  RC open(Trx *trx) override { return RC::UNIMPLEMENTED; }
+  RC next(Chunk &chunk) override { return RC::UNIMPLEMENTED; }
+  RC close() override { return RC::UNIMPLEMENTED; }
 
 private:
 };
\ No newline at end of file
diff --git a/src/observer/sql/operator/physical_operator.h b/src/observer/sql/operator/physical_operator.h
index 6ed13a20d..a3b20ed34 100644
--- a/src/observer/sql/operator/physical_operator.h
+++ b/src/observer/sql/operator/physical_operator.h
@@ -77,13 +77,13 @@ class PhysicalOperator
   virtual PhysicalOperatorType type() const = 0;
 
   virtual RC open(Trx *trx) = 0;
-  virtual RC next() { return RC::UNIMPLENMENT; }
-  virtual RC next(Chunk &chunk) { return RC::UNIMPLENMENT; }
+  virtual RC next() { return RC::UNIMPLEMENTED; }
+  virtual RC next(Chunk &chunk) { return RC::UNIMPLEMENTED; }
   virtual RC close() = 0;
 
   virtual Tuple *current_tuple() { return nullptr; }
 
-  virtual RC tuple_schema(TupleSchema &schema) const { return RC::UNIMPLENMENT; }
+  virtual RC tuple_schema(TupleSchema &schema) const { return RC::UNIMPLEMENTED; }
 
   void add_child(std::unique_ptr<PhysicalOperator> oper) { children_.emplace_back(std::move(oper)); }
 
diff --git a/src/observer/sql/operator/string_list_physical_operator.h b/src/observer/sql/operator/string_list_physical_operator.h
index b363b3be0..86ffd3fc4 100644
--- a/src/observer/sql/operator/string_list_physical_operator.h
+++ b/src/observer/sql/operator/string_list_physical_operator.h
@@ -74,8 +74,7 @@ class StringListPhysicalOperator : public PhysicalOperator
     std::vector<Value> cells;
     for (const std::string &s : string_list) {
 
-      Value value;
-      value.set_string(s.c_str());
+      Value value(s.c_str());
       cells.push_back(value);
     }
     tuple_.set_cells(cells);
diff --git a/src/observer/sql/optimizer/logical_plan_generator.cpp b/src/observer/sql/optimizer/logical_plan_generator.cpp
index af6c33bb6..8c418b320 100644
--- a/src/observer/sql/optimizer/logical_plan_generator.cpp
+++ b/src/observer/sql/optimizer/logical_plan_generator.cpp
@@ -74,7 +74,7 @@ RC LogicalPlanGenerator::create(Stmt *stmt, unique_ptr<LogicalOperator> &logical
       rc = create_plan(explain_stmt, logical_operator);
     } break;
     default: {
-      rc = RC::UNIMPLENMENT;
+      rc = RC::UNIMPLEMENTED;
     }
   }
   return rc;
@@ -149,6 +149,7 @@ RC LogicalPlanGenerator::create_plan(SelectStmt *select_stmt, unique_ptr<Logical
 
 RC LogicalPlanGenerator::create_plan(FilterStmt *filter_stmt, unique_ptr<LogicalOperator> &logical_operator)
 {
+  RC                                  rc = RC::SUCCESS;
   std::vector<unique_ptr<Expression>> cmp_exprs;
   const std::vector<FilterUnit *>    &filter_units = filter_stmt->filter_units();
   for (const FilterUnit *filter_unit : filter_units) {
@@ -163,6 +164,45 @@ RC LogicalPlanGenerator::create_plan(FilterStmt *filter_stmt, unique_ptr<Logical
                                      ? static_cast<Expression *>(new FieldExpr(filter_obj_right.field))
                                      : static_cast<Expression *>(new ValueExpr(filter_obj_right.value)));
 
+    if (left->value_type() != right->value_type()) {
+      auto left_to_right_cost = implicit_cast_cost(left->value_type(), right->value_type());
+      auto right_to_left_cost = implicit_cast_cost(right->value_type(), left->value_type());
+      if (left_to_right_cost <= right_to_left_cost && left_to_right_cost != INT32_MAX) {
+        ExprType left_type = left->type();
+        auto cast_expr = make_unique<CastExpr>(std::move(left), right->value_type());
+        if (left_type == ExprType::VALUE) {
+          Value left_val;
+          if (OB_FAIL(rc = cast_expr->try_get_value(left_val)))
+          {
+            LOG_WARN("failed to get value from left child", strrc(rc));
+            return rc;
+          }
+          left = make_unique<ValueExpr>(left_val);
+        } else {
+          left = std::move(cast_expr);
+        }
+      } else if (right_to_left_cost < left_to_right_cost && right_to_left_cost != INT32_MAX) {
+        ExprType right_type = right->type();
+        auto cast_expr = make_unique<CastExpr>(std::move(right), left->value_type());
+        if (right_type == ExprType::VALUE) {
+          Value right_val;
+          if (OB_FAIL(rc = cast_expr->try_get_value(right_val)))
+          {
+            LOG_WARN("failed to get value from right child", strrc(rc));
+            return rc;
+          }
+          right = make_unique<ValueExpr>(right_val);
+        } else {
+          right = std::move(cast_expr);
+        }
+
+      } else {
+        rc = RC::UNSUPPORTED;
+        LOG_WARN("unsupported cast from %s to %s", attr_type_to_string(left->value_type()), attr_type_to_string(right->value_type()));
+        return rc;
+      }
+    }
+
     ComparisonExpr *cmp_expr = new ComparisonExpr(filter_unit->comp(), std::move(left), std::move(right));
     cmp_exprs.emplace_back(cmp_expr);
   }
@@ -174,7 +214,15 @@ RC LogicalPlanGenerator::create_plan(FilterStmt *filter_stmt, unique_ptr<Logical
   }
 
   logical_operator = std::move(predicate_oper);
-  return RC::SUCCESS;
+  return rc;
+}
+
+int LogicalPlanGenerator::implicit_cast_cost(AttrType from, AttrType to)
+{
+  if (from == to) {
+    return 0;
+  }
+  return DataType::type_instance(from)->cast_cost(to);
 }
 
 RC LogicalPlanGenerator::create_plan(InsertStmt *insert_stmt, unique_ptr<LogicalOperator> &logical_operator)
diff --git a/src/observer/sql/optimizer/logical_plan_generator.h b/src/observer/sql/optimizer/logical_plan_generator.h
index 951ae8f47..d02cb1816 100644
--- a/src/observer/sql/optimizer/logical_plan_generator.h
+++ b/src/observer/sql/optimizer/logical_plan_generator.h
@@ -17,6 +17,7 @@ See the Mulan PSL v2 for more details. */
 #include <memory>
 
 #include "common/rc.h"
+#include "common/type/attr_type.h"
 
 class Stmt;
 class CalcStmt;
@@ -44,4 +45,6 @@ class LogicalPlanGenerator
   RC create_plan(ExplainStmt *explain_stmt, std::unique_ptr<LogicalOperator> &logical_operator);
 
   RC create_group_by_plan(SelectStmt *select_stmt, std::unique_ptr<LogicalOperator> &logical_operator);
+
+  int implicit_cast_cost(AttrType from, AttrType to);
 };
\ No newline at end of file
diff --git a/src/observer/sql/optimizer/optimize_stage.cpp b/src/observer/sql/optimizer/optimize_stage.cpp
index 19847fcb5..902c2ce86 100644
--- a/src/observer/sql/optimizer/optimize_stage.cpp
+++ b/src/observer/sql/optimizer/optimize_stage.cpp
@@ -35,7 +35,7 @@ RC OptimizeStage::handle_request(SQLStageEvent *sql_event)
 
   RC rc = create_logical_plan(sql_event, logical_operator);
   if (rc != RC::SUCCESS) {
-    if (rc != RC::UNIMPLENMENT) {
+    if (rc != RC::UNIMPLEMENTED) {
       LOG_WARN("failed to create logical plan. rc=%s", strrc(rc));
     }
     return rc;
@@ -113,7 +113,7 @@ RC OptimizeStage::create_logical_plan(SQLStageEvent *sql_event, unique_ptr<Logic
 {
   Stmt *stmt = sql_event->stmt();
   if (nullptr == stmt) {
-    return RC::UNIMPLENMENT;
+    return RC::UNIMPLEMENTED;
   }
 
   return logical_plan_generator_.create(stmt, logical_operator);
diff --git a/src/observer/sql/optimizer/predicate_pushdown_rewriter.cpp b/src/observer/sql/optimizer/predicate_pushdown_rewriter.cpp
index f206342fc..342218f73 100644
--- a/src/observer/sql/optimizer/predicate_pushdown_rewriter.cpp
+++ b/src/observer/sql/optimizer/predicate_pushdown_rewriter.cpp
@@ -97,7 +97,7 @@ RC PredicatePushdownRewriter::get_exprs_can_pushdown(
     // 或 操作的比较，太复杂，现在不考虑
     if (conjunction_expr->conjunction_type() == ConjunctionExpr::Type::OR) {
       LOG_WARN("unsupported or operation");
-      rc = RC::UNIMPLENMENT;
+      rc = RC::UNIMPLEMENTED;
       return rc;
     }
 
diff --git a/src/observer/sql/parser/parse_defs.h b/src/observer/sql/parser/parse_defs.h
index 4672bc1e8..9593d57ab 100644
--- a/src/observer/sql/parser/parse_defs.h
+++ b/src/observer/sql/parser/parse_defs.h
@@ -18,7 +18,7 @@ See the Mulan PSL v2 for more details. */
 #include <vector>
 #include <memory>
 
-#include "sql/parser/value.h"
+#include "common/value.h"
 
 class Expression;
 
diff --git a/src/observer/sql/parser/resolve_stage.cpp b/src/observer/sql/parser/resolve_stage.cpp
index e9681fafc..d04337b93 100644
--- a/src/observer/sql/parser/resolve_stage.cpp
+++ b/src/observer/sql/parser/resolve_stage.cpp
@@ -47,7 +47,7 @@ RC ResolveStage::handle_request(SQLStageEvent *sql_event)
   Stmt          *stmt     = nullptr;
 
   rc = Stmt::create_stmt(db, *sql_node, stmt);
-  if (rc != RC::SUCCESS && rc != RC::UNIMPLENMENT) {
+  if (rc != RC::SUCCESS && rc != RC::UNIMPLEMENTED) {
     LOG_WARN("failed to create stmt. rc=%d:%s", rc, strrc(rc));
     sql_result->set_return_code(rc);
     return rc;
diff --git a/src/observer/sql/parser/value.cpp b/src/observer/sql/parser/value.cpp
deleted file mode 100644
index 65c9ba4be..000000000
--- a/src/observer/sql/parser/value.cpp
+++ /dev/null
@@ -1,287 +0,0 @@
-/* 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. */
-
-//
-// Created by WangYunlai on 2023/06/28.
-//
-
-#include "sql/parser/value.h"
-#include "common/lang/comparator.h"
-#include "common/lang/string.h"
-#include "common/log/log.h"
-#include <sstream>
-
-const char *ATTR_TYPE_NAME[] = {"undefined", "chars", "ints", "floats", "booleans"};
-
-const char *attr_type_to_string(AttrType type)
-{
-  if (type >= AttrType::UNDEFINED && type <= AttrType::FLOATS) {
-    return ATTR_TYPE_NAME[static_cast<int>(type)];
-  }
-  return "unknown";
-}
-AttrType attr_type_from_string(const char *s)
-{
-  for (unsigned int i = 0; i < sizeof(ATTR_TYPE_NAME) / sizeof(ATTR_TYPE_NAME[0]); i++) {
-    if (0 == strcmp(ATTR_TYPE_NAME[i], s)) {
-      return (AttrType)i;
-    }
-  }
-  return AttrType::UNDEFINED;
-}
-
-Value::Value(int val) { set_int(val); }
-
-Value::Value(float val) { set_float(val); }
-
-Value::Value(bool val) { set_boolean(val); }
-
-Value::Value(const char *s, int len /*= 0*/) { set_string(s, len); }
-
-void Value::set_data(char *data, int length)
-{
-  switch (attr_type_) {
-    case AttrType::CHARS: {
-      set_string(data, length);
-    } break;
-    case AttrType::INTS: {
-      num_value_.int_value_ = *(int *)data;
-      length_               = length;
-    } break;
-    case AttrType::FLOATS: {
-      num_value_.float_value_ = *(float *)data;
-      length_                 = length;
-    } break;
-    case AttrType::BOOLEANS: {
-      num_value_.bool_value_ = *(int *)data != 0;
-      length_                = length;
-    } break;
-    default: {
-      LOG_WARN("unknown data type: %d", attr_type_);
-    } break;
-  }
-}
-void Value::set_int(int val)
-{
-  attr_type_            = AttrType::INTS;
-  num_value_.int_value_ = val;
-  length_               = sizeof(val);
-}
-
-void Value::set_float(float val)
-{
-  attr_type_              = AttrType::FLOATS;
-  num_value_.float_value_ = val;
-  length_                 = sizeof(val);
-}
-void Value::set_boolean(bool val)
-{
-  attr_type_             = AttrType::BOOLEANS;
-  num_value_.bool_value_ = val;
-  length_                = sizeof(val);
-}
-void Value::set_string(const char *s, int len /*= 0*/)
-{
-  attr_type_ = AttrType::CHARS;
-  if (len > 0) {
-    len = strnlen(s, len);
-    str_value_.assign(s, len);
-  } else {
-    str_value_.assign(s);
-  }
-  length_ = str_value_.length();
-}
-
-void Value::set_value(const Value &value)
-{
-  switch (value.attr_type_) {
-    case AttrType::INTS: {
-      set_int(value.get_int());
-    } break;
-    case AttrType::FLOATS: {
-      set_float(value.get_float());
-    } break;
-    case AttrType::CHARS: {
-      set_string(value.get_string().c_str());
-    } break;
-    case AttrType::BOOLEANS: {
-      set_boolean(value.get_boolean());
-    } break;
-    case AttrType::UNDEFINED: {
-      ASSERT(false, "got an invalid value type");
-    } break;
-  }
-}
-
-const char *Value::data() const
-{
-  switch (attr_type_) {
-    case AttrType::CHARS: {
-      return str_value_.c_str();
-    } break;
-    default: {
-      return (const char *)&num_value_;
-    } break;
-  }
-}
-
-std::string Value::to_string() const
-{
-  std::stringstream os;
-  switch (attr_type_) {
-    case AttrType::INTS: {
-      os << num_value_.int_value_;
-    } break;
-    case AttrType::FLOATS: {
-      os << common::double_to_str(num_value_.float_value_);
-    } break;
-    case AttrType::BOOLEANS: {
-      os << num_value_.bool_value_;
-    } break;
-    case AttrType::CHARS: {
-      os << str_value_;
-    } break;
-    default: {
-      LOG_WARN("unsupported attr type: %d", attr_type_);
-    } break;
-  }
-  return os.str();
-}
-
-int Value::compare(const Value &other) const
-{
-  if (this->attr_type_ == other.attr_type_) {
-    switch (this->attr_type_) {
-      case AttrType::INTS: {
-        return common::compare_int((void *)&this->num_value_.int_value_, (void *)&other.num_value_.int_value_);
-      } break;
-      case AttrType::FLOATS: {
-        return common::compare_float((void *)&this->num_value_.float_value_, (void *)&other.num_value_.float_value_);
-      } break;
-      case AttrType::CHARS: {
-        return common::compare_string((void *)this->str_value_.c_str(),
-            this->str_value_.length(),
-            (void *)other.str_value_.c_str(),
-            other.str_value_.length());
-      } break;
-      case AttrType::BOOLEANS: {
-        return common::compare_int((void *)&this->num_value_.bool_value_, (void *)&other.num_value_.bool_value_);
-      }
-      default: {
-        LOG_WARN("unsupported type: %d", this->attr_type_);
-      }
-    }
-  } else if (this->attr_type_ == AttrType::INTS && other.attr_type_ == AttrType::FLOATS) {
-    float this_data = this->num_value_.int_value_;
-    return common::compare_float((void *)&this_data, (void *)&other.num_value_.float_value_);
-  } else if (this->attr_type_ == AttrType::FLOATS && other.attr_type_ == AttrType::INTS) {
-    float other_data = other.num_value_.int_value_;
-    return common::compare_float((void *)&this->num_value_.float_value_, (void *)&other_data);
-  }
-  LOG_WARN("not supported");
-  return -1;  // TODO return rc?
-}
-
-int Value::get_int() const
-{
-  switch (attr_type_) {
-    case AttrType::CHARS: {
-      try {
-        return (int)(std::stol(str_value_));
-      } catch (std::exception const &ex) {
-        LOG_TRACE("failed to convert string to number. s=%s, ex=%s", str_value_.c_str(), ex.what());
-        return 0;
-      }
-    }
-    case AttrType::INTS: {
-      return num_value_.int_value_;
-    }
-    case AttrType::FLOATS: {
-      return (int)(num_value_.float_value_);
-    }
-    case AttrType::BOOLEANS: {
-      return (int)(num_value_.bool_value_);
-    }
-    default: {
-      LOG_WARN("unknown data type. type=%d", attr_type_);
-      return 0;
-    }
-  }
-  return 0;
-}
-
-float Value::get_float() const
-{
-  switch (attr_type_) {
-    case AttrType::CHARS: {
-      try {
-        return std::stof(str_value_);
-      } catch (std::exception const &ex) {
-        LOG_TRACE("failed to convert string to float. s=%s, ex=%s", str_value_.c_str(), ex.what());
-        return 0.0;
-      }
-    } break;
-    case AttrType::INTS: {
-      return float(num_value_.int_value_);
-    } break;
-    case AttrType::FLOATS: {
-      return num_value_.float_value_;
-    } break;
-    case AttrType::BOOLEANS: {
-      return float(num_value_.bool_value_);
-    } break;
-    default: {
-      LOG_WARN("unknown data type. type=%d", attr_type_);
-      return 0;
-    }
-  }
-  return 0;
-}
-
-std::string Value::get_string() const { return this->to_string(); }
-
-bool Value::get_boolean() const
-{
-  switch (attr_type_) {
-    case AttrType::CHARS: {
-      try {
-        float val = std::stof(str_value_);
-        if (val >= EPSILON || val <= -EPSILON) {
-          return true;
-        }
-
-        int int_val = std::stol(str_value_);
-        if (int_val != 0) {
-          return true;
-        }
-
-        return !str_value_.empty();
-      } catch (std::exception const &ex) {
-        LOG_TRACE("failed to convert string to float or integer. s=%s, ex=%s", str_value_.c_str(), ex.what());
-        return !str_value_.empty();
-      }
-    } break;
-    case AttrType::INTS: {
-      return num_value_.int_value_ != 0;
-    } break;
-    case AttrType::FLOATS: {
-      float val = num_value_.float_value_;
-      return val >= EPSILON || val <= -EPSILON;
-    } break;
-    case AttrType::BOOLEANS: {
-      return num_value_.bool_value_;
-    } break;
-    default: {
-      LOG_WARN("unknown data type. type=%d", attr_type_);
-      return false;
-    }
-  }
-  return false;
-}
diff --git a/src/observer/sql/parser/value.h b/src/observer/sql/parser/value.h
deleted file mode 100644
index 74d0eb590..000000000
--- a/src/observer/sql/parser/value.h
+++ /dev/null
@@ -1,93 +0,0 @@
-/* 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. */
-
-//
-// Created by Wangyunlai 2023/6/27
-//
-
-#pragma once
-
-#include <string>
-
-/**
- * @brief 属性的类型
- *
- */
-enum class AttrType
-{
-  UNDEFINED,
-  CHARS,     ///< 字符串类型
-  INTS,      ///< 整数类型(4字节)
-  FLOATS,    ///< 浮点数类型(4字节)
-  BOOLEANS,  ///< boolean类型，当前不是由parser解析出来的，是程序内部使用的
-};
-
-const char *attr_type_to_string(AttrType type);
-AttrType    attr_type_from_string(const char *s);
-
-/**
- * @brief 属性的值
- *
- */
-class Value final
-{
-public:
-  Value() = default;
-
-  Value(AttrType attr_type, char *data, int length = 4) : attr_type_(attr_type) { this->set_data(data, length); }
-
-  explicit Value(int val);
-  explicit Value(float val);
-  explicit Value(bool val);
-  explicit Value(const char *s, int len = 0);
-
-  Value(const Value &other)            = default;
-  Value &operator=(const Value &other) = default;
-
-  void set_type(AttrType type) { this->attr_type_ = type; }
-  void set_data(char *data, int length);
-  void set_data(const char *data, int length) { this->set_data(const_cast<char *>(data), length); }
-  void set_int(int val);
-  void set_float(float val);
-  void set_boolean(bool val);
-  void set_string(const char *s, int len = 0);
-  void set_value(const Value &value);
-
-  std::string to_string() const;
-
-  int compare(const Value &other) const;
-
-  const char *data() const;
-  int         length() const { return length_; }
-
-  AttrType attr_type() const { return attr_type_; }
-
-public:
-  /**
-   * 获取对应的值
-   * 如果当前的类型与期望获取的类型不符，就会执行转换操作
-   */
-  int         get_int() const;
-  float       get_float() const;
-  std::string get_string() const;
-  bool        get_boolean() const;
-
-private:
-  AttrType attr_type_ = AttrType::UNDEFINED;
-  int      length_    = 0;
-
-  union
-  {
-    int   int_value_;
-    float float_value_;
-    bool  bool_value_;
-  } num_value_;
-  std::string str_value_;
-};
\ No newline at end of file
diff --git a/src/observer/sql/stmt/insert_stmt.cpp b/src/observer/sql/stmt/insert_stmt.cpp
index 3f7afa415..96ced424f 100644
--- a/src/observer/sql/stmt/insert_stmt.cpp
+++ b/src/observer/sql/stmt/insert_stmt.cpp
@@ -47,19 +47,6 @@ RC InsertStmt::create(Db *db, const InsertSqlNode &inserts, Stmt *&stmt)
     return RC::SCHEMA_FIELD_MISSING;
   }
 
-  // check fields type
-  const int sys_field_num = table_meta.sys_field_num();
-  for (int i = 0; i < value_num; i++) {
-    const FieldMeta *field_meta = table_meta.field(i + sys_field_num);
-    const AttrType   field_type = field_meta->type();
-    const AttrType   value_type = values[i].attr_type();
-    if (field_type != value_type) {  // TODO try to convert the value type to field type
-      LOG_WARN("field type mismatch. table=%s, field=%s, field type=%d, value_type=%d",
-          table_name, field_meta->name(), field_type, value_type);
-      return RC::SCHEMA_FIELD_TYPE_MISMATCH;
-    }
-  }
-
   // everything alright
   stmt = new InsertStmt(table, values, value_num);
   return RC::SUCCESS;
diff --git a/src/observer/sql/stmt/stmt.cpp b/src/observer/sql/stmt/stmt.cpp
index 08a15d664..61be9c1b6 100644
--- a/src/observer/sql/stmt/stmt.cpp
+++ b/src/observer/sql/stmt/stmt.cpp
@@ -112,5 +112,5 @@ RC Stmt::create_stmt(Db *db, ParsedSqlNode &sql_node, Stmt *&stmt)
       LOG_INFO("Command::type %d doesn't need to create statement.", sql_node.flag);
     } break;
   }
-  return RC::UNIMPLENMENT;
+  return RC::UNIMPLEMENTED;
 }
diff --git a/src/observer/storage/common/condition_filter.cpp b/src/observer/storage/common/condition_filter.cpp
index 47caa6d1c..5a62bb10f 100644
--- a/src/observer/storage/common/condition_filter.cpp
+++ b/src/observer/storage/common/condition_filter.cpp
@@ -14,7 +14,7 @@ See the Mulan PSL v2 for more details. */
 
 #include "condition_filter.h"
 #include "common/log/log.h"
-#include "sql/parser/value.h"
+#include "common/value.h"
 #include "storage/record/record_manager.h"
 #include "storage/table/table.h"
 #include <math.h>
@@ -38,7 +38,7 @@ DefaultConditionFilter::~DefaultConditionFilter() {}
 
 RC DefaultConditionFilter::init(const ConDesc &left, const ConDesc &right, AttrType attr_type, CompOp comp_op)
 {
-  if (attr_type < AttrType::CHARS || attr_type > AttrType::FLOATS) {
+  if (attr_type <= AttrType::UNDEFINED || attr_type >= AttrType::MAXTYPE) {
     LOG_ERROR("Invalid condition with unsupported attribute type: %d", attr_type);
     return RC::INVALID_ARGUMENT;
   }
diff --git a/src/observer/storage/default/default_handler.cpp b/src/observer/storage/default/default_handler.cpp
index 3858702e6..e310aa916 100644
--- a/src/observer/storage/default/default_handler.cpp
+++ b/src/observer/storage/default/default_handler.cpp
@@ -132,7 +132,7 @@ RC DefaultHandler::open_db(const char *dbname)
   return ret;
 }
 
-RC DefaultHandler::close_db(const char *dbname) { return RC::UNIMPLENMENT; }
+RC DefaultHandler::close_db(const char *dbname) { return RC::UNIMPLEMENTED; }
 
 RC DefaultHandler::create_table(const char *dbname, const char *relation_name, span<const AttrInfoSqlNode> attributes)
 {
@@ -143,7 +143,7 @@ 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::UNIMPLENMENT; }
+RC DefaultHandler::drop_table(const char *dbname, const char *relation_name) { return RC::UNIMPLEMENTED; }
 
 Db *DefaultHandler::find_db(const char *dbname) const
 {
diff --git a/src/observer/storage/field/field.cpp b/src/observer/storage/field/field.cpp
index bfc0777da..b554787d8 100644
--- a/src/observer/storage/field/field.cpp
+++ b/src/observer/storage/field/field.cpp
@@ -14,7 +14,7 @@ See the Mulan PSL v2 for more details. */
 
 #include "storage/field/field.h"
 #include "common/log/log.h"
-#include "sql/parser/value.h"
+#include "common/value.h"
 #include "storage/record/record.h"
 
 void Field::set_int(Record &record, int value)
diff --git a/src/observer/storage/index/bplus_tree.h b/src/observer/storage/index/bplus_tree.h
index caea013eb..dd7520564 100644
--- a/src/observer/storage/index/bplus_tree.h
+++ b/src/observer/storage/index/bplus_tree.h
@@ -66,21 +66,14 @@ class AttrComparator
 
   int operator()(const char *v1, const char *v2) const
   {
-    switch (attr_type_) {
-      case AttrType::INTS: {
-        return common::compare_int((void *)v1, (void *)v2);
-      } break;
-      case AttrType::FLOATS: {
-        return common::compare_float((void *)v1, (void *)v2);
-      }
-      case AttrType::CHARS: {
-        return common::compare_string((void *)v1, attr_length_, (void *)v2, attr_length_);
-      }
-      default: {
-        ASSERT(false, "unknown attr type. %d", attr_type_);
-        return 0;
-      }
-    }
+    // TODO: optimized the comparison
+    Value left;
+    left.set_type(attr_type_);
+    left.set_data(v1, attr_length_);
+    Value right;
+    right.set_type(attr_type_);
+    right.set_data(v2, attr_length_);
+    return DataType::type_instance(attr_type_)->compare(left, right);
   }
 
 private:
diff --git a/src/observer/storage/record/record_manager.h b/src/observer/storage/record/record_manager.h
index 421e6e76c..810ec586d 100644
--- a/src/observer/storage/record/record_manager.h
+++ b/src/observer/storage/record/record_manager.h
@@ -179,7 +179,7 @@ class RecordPageHandler
    * @param data 要插入的记录
    * @param rid  如果插入成功，通过这个参数返回插入的位置
    */
-  virtual RC insert_record(const char *data, RID *rid) { return RC::UNIMPLENMENT; }
+  virtual RC insert_record(const char *data, RID *rid) { return RC::UNIMPLEMENTED; }
 
   /**
    * @brief 数据库恢复时，在指定位置插入数据
@@ -187,20 +187,20 @@ class RecordPageHandler
    * @param data 要插入的数据行
    * @param rid  插入的位置
    */
-  virtual RC recover_insert_record(const char *data, const RID &rid) { return RC::UNIMPLENMENT; }
+  virtual RC recover_insert_record(const char *data, const RID &rid) { return RC::UNIMPLEMENTED; }
 
   /**
    * @brief 删除指定的记录
    *
    * @param rid 要删除的记录标识
    */
-  virtual RC delete_record(const RID *rid) { return RC::UNIMPLENMENT; }
+  virtual RC delete_record(const RID *rid) { return RC::UNIMPLEMENTED; }
 
   /**
    * @brief
    *
    */
-  virtual RC update_record(const RID &rid, const char *data) { return RC::UNIMPLENMENT; }
+  virtual RC update_record(const RID &rid, const char *data) { return RC::UNIMPLEMENTED; }
 
   /**
    * @brief 获取指定位置的记录数据
@@ -208,7 +208,7 @@ class RecordPageHandler
    * @param rid 指定的位置
    * @param record 获取到的记录结果
    */
-  virtual RC get_record(const RID &rid, Record &record) { return RC::UNIMPLENMENT; }
+  virtual RC get_record(const RID &rid, Record &record) { return RC::UNIMPLEMENTED; }
 
   /**
    * @brief 获取整个页面中指定列的所有记录。
@@ -216,7 +216,7 @@ class RecordPageHandler
    * @param chunk 由 chunk.column(i).col_id() 指定列。
    * 只需由 PaxRecordPageHandler 实现。
    */
-  virtual RC get_chunk(Chunk &chunk) { return RC::UNIMPLENMENT; }
+  virtual RC get_chunk(Chunk &chunk) { return RC::UNIMPLEMENTED; }
 
   /**
    * @brief 返回该记录页的页号
diff --git a/src/observer/storage/table/table.cpp b/src/observer/storage/table/table.cpp
index 910dd259e..9ab04d7b9 100644
--- a/src/observer/storage/table/table.cpp
+++ b/src/observer/storage/table/table.cpp
@@ -257,6 +257,7 @@ const TableMeta &Table::table_meta() const { return table_meta_; }
 
 RC Table::make_record(int value_num, const Value *values, Record &record)
 {
+  RC rc = RC::SUCCESS;
   // 检查字段类型是否一致
   if (value_num + table_meta_.sys_field_num() != table_meta_.field_num()) {
     LOG_WARN("Input values don't match the table's schema, table name:%s", table_meta_.name());
@@ -264,37 +265,46 @@ RC Table::make_record(int value_num, const Value *values, Record &record)
   }
 
   const int normal_field_start_index = table_meta_.sys_field_num();
-  for (int i = 0; i < value_num; i++) {
-    const FieldMeta *field = table_meta_.field(i + normal_field_start_index);
-    const Value     &value = values[i];
-    if (field->type() != value.attr_type()) {
-      LOG_ERROR("Invalid value type. table name =%s, field name=%s, type=%d, but given=%d",
-                table_meta_.name(), field->name(), field->type(), value.attr_type());
-      return RC::SCHEMA_FIELD_TYPE_MISMATCH;
-    }
-  }
-
   // 复制所有字段的值
   int   record_size = table_meta_.record_size();
   char *record_data = (char *)malloc(record_size);
+  memset(record_data, 0, record_size);
 
   for (int i = 0; i < value_num; i++) {
-    const FieldMeta *field    = table_meta_.field(i + normal_field_start_index);
-    const Value     &value    = values[i];
-    size_t           copy_len = field->len();
-    if (field->type() == AttrType::CHARS) {
-      const size_t data_len = value.length();
-      if (copy_len > data_len) {
-        copy_len = data_len + 1;
+    const FieldMeta *field = table_meta_.field(i + normal_field_start_index);
+    const Value &    value = values[i];
+    if (field->type() != value.attr_type()) {
+      Value real_value;
+      rc = Value::cast_to(value, field->type(), real_value);
+      if (OB_FAIL(rc)) {
+        return rc;
       }
+      rc = set_value_to_record(record_data, real_value, field);
+    } else {
+      rc = set_value_to_record(record_data, value, field);
     }
-    memcpy(record_data + field->offset(), value.data(), copy_len);
+  }
+  if (OB_FAIL(rc)) {
+    return rc;
   }
 
   record.set_data_owner(record_data, record_size);
   return RC::SUCCESS;
 }
 
+RC Table::set_value_to_record(char *record_data, const Value &value, const FieldMeta *field)
+{
+  size_t       copy_len = field->len();
+  const size_t data_len = value.length();
+  if (field->type() == AttrType::CHARS) {
+    if (copy_len > data_len) {
+      copy_len = data_len + 1;
+    }
+  }
+  memcpy(record_data + field->offset(), value.data(), copy_len);
+  return RC::SUCCESS;
+}
+
 RC Table::init_record_handler(const char *base_dir)
 {
   string data_file = table_data_file(base_dir, table_meta_.name());
diff --git a/src/observer/storage/table/table.h b/src/observer/storage/table/table.h
index 5fc14021c..089e13ae4 100644
--- a/src/observer/storage/table/table.h
+++ b/src/observer/storage/table/table.h
@@ -113,6 +113,7 @@ class Table
 private:
   RC insert_entry_of_indexes(const char *record, const RID &rid);
   RC delete_entry_of_indexes(const char *record, const RID &rid, bool error_on_not_exists);
+  RC set_value_to_record(char *record_data, const Value &value, const FieldMeta *field);
 
 private:
   RC init_record_handler(const char *base_dir);
diff --git a/test/case/result/primary-date.result b/test/case/result/primary-date.result
index 11e7408c3..1485ec30c 100644
--- a/test/case/result/primary-date.result
+++ b/test/case/result/primary-date.result
@@ -1,69 +1,69 @@
 INITIALIZATION
-CREATE TABLE date_table(id int, u_date date);
+CREATE TABLE DATE_TABLE(ID INT, U_DATE DATE);
 SUCCESS
-CREATE INDEX index_id on date_table(u_date);
+CREATE INDEX INDEX_ID ON DATE_TABLE(U_DATE);
 SUCCESS
 
 1. INSERT NORMAL DATE DATA
-INSERT INTO date_table VALUES (1,'2020-01-21');
+INSERT INTO DATE_TABLE VALUES (1,'2020-01-21');
 SUCCESS
-INSERT INTO date_table VALUES (2,'2020-10-21');
+INSERT INTO DATE_TABLE VALUES (2,'2020-10-21');
 SUCCESS
-INSERT INTO date_table VALUES (3,'2020-1-01');
+INSERT INTO DATE_TABLE VALUES (3,'2020-1-01');
 SUCCESS
-INSERT INTO date_table VALUES (4,'2020-01-1');
+INSERT INTO DATE_TABLE VALUES (4,'2020-01-1');
 SUCCESS
-INSERT INTO date_table VALUES (5,'2019-12-21');
+INSERT INTO DATE_TABLE VALUES (5,'2019-12-21');
 SUCCESS
-INSERT INTO date_table VALUES (6,'2016-2-29');
+INSERT INTO DATE_TABLE VALUES (6,'2016-2-29');
 SUCCESS
-INSERT INTO date_table VALUES (7,'1970-1-1');
+INSERT INTO DATE_TABLE VALUES (7,'1970-1-1');
 SUCCESS
-INSERT INTO date_table VALUES (8,'2000-01-01');
+INSERT INTO DATE_TABLE VALUES (8,'2000-01-01');
 SUCCESS
-INSERT INTO date_table VALUES (9,'2038-1-19');
+INSERT INTO DATE_TABLE VALUES (9,'2038-1-19');
 SUCCESS
 
 2. COMPARE DATE DATA
-SELECT * FROM date_table WHERE u_date>'2020-1-20';
+SELECT * FROM DATE_TABLE WHERE U_DATE>'2020-1-20';
 1 | 2020-01-21
 2 | 2020-10-21
 9 | 2038-01-19
 ID | U_DATE
-SELECT * FROM date_table WHERE u_date<'2019-12-31';
+SELECT * FROM DATE_TABLE WHERE U_DATE<'2019-12-31';
 5 | 2019-12-21
 6 | 2016-02-29
 7 | 1970-01-01
 8 | 2000-01-01
 ID | U_DATE
-SELECT * FROM date_table WHERE u_date='2020-1-1';
+SELECT * FROM DATE_TABLE WHERE U_DATE='2020-1-1';
 3 | 2020-01-01
 4 | 2020-01-01
 ID | U_DATE
 
 3. DELETE DATA
-DELETE FROM date_table WHERE u_date>'2012-2-29';
+DELETE FROM DATE_TABLE WHERE U_DATE>'2012-2-29';
 SUCCESS
-SELECT * FROM date_table;
+SELECT * FROM DATE_TABLE;
 7 | 1970-01-01
 8 | 2000-01-01
 ID | U_DATE
 
 4. CHECK INVALID DATE DATA
-SELECT * FROM date_table WHERE u_date='2017-2-29';
+SELECT * FROM DATE_TABLE WHERE U_DATE='2017-2-29';
 FAILURE
-SELECT * FROM date_table WHERE u_date='2017-21-29';
+SELECT * FROM DATE_TABLE WHERE U_DATE='2017-21-29';
 FAILURE
-SELECT * FROM date_table WHERE u_date='2017-12-32';
+SELECT * FROM DATE_TABLE WHERE U_DATE='2017-12-32';
 FAILURE
-SELECT * FROM date_table WHERE u_date='2017-11-31';
+SELECT * FROM DATE_TABLE WHERE U_DATE='2017-11-31';
 FAILURE
 
-INSERT INTO date_table VALUES (10,'2017-2-29');
+INSERT INTO DATE_TABLE VALUES (10,'2017-2-29');
 FAILURE
-INSERT INTO date_table VALUES (11,'2017-21-29');
+INSERT INTO DATE_TABLE VALUES (11,'2017-21-29');
 FAILURE
-INSERT INTO date_table VALUES (12,'2017-12-32');
+INSERT INTO DATE_TABLE VALUES (12,'2017-12-32');
 FAILURE
-INSERT INTO date_table VALUES (13,'2017-11-31');
+INSERT INTO DATE_TABLE VALUES (13,'2017-11-31');
 FAILURE
diff --git a/test/case/test/primary-date.test b/test/case/test/primary-date.test
index 5a6c1a0c1..552f542f1 100644
--- a/test/case/test/primary-date.test
+++ b/test/case/test/primary-date.test
@@ -23,10 +23,10 @@ DELETE FROM date_table WHERE u_date>'2012-2-29';
 -- sort SELECT * FROM date_table;
 
 -- echo 4. check invalid date data
-SELECT * FROM date_table WHERE u_date='2017-2-29';
-SELECT * FROM date_table WHERE u_date='2017-21-29';
-SELECT * FROM date_table WHERE u_date='2017-12-32';
-SELECT * FROM date_table WHERE u_date='2017-11-31';
+--sort SELECT * FROM date_table WHERE u_date='2017-2-29';
+--sort SELECT * FROM date_table WHERE u_date='2017-21-29';
+--sort SELECT * FROM date_table WHERE u_date='2017-12-32';
+--sort SELECT * FROM date_table WHERE u_date='2017-11-31';
 
 INSERT INTO date_table VALUES (10,'2017-2-29');
 INSERT INTO date_table VALUES (11,'2017-21-29');
diff --git a/unittest/observer/expression_test.cpp b/unittest/observer/expression_test.cpp
index 94d85a3b1..9b8bb6972 100644
--- a/unittest/observer/expression_test.cpp
+++ b/unittest/observer/expression_test.cpp
@@ -252,7 +252,7 @@ TEST(CastExpr, cast_expr_test)
   Value                  result;
   cast_expr.get_value(tuple, result);
   ASSERT_EQ(result.get_int(), 1);
-  result.set_int(0);
+  result.reset();
   cast_expr.try_get_value(result);
   ASSERT_EQ(result.get_int(), 1);
 }
diff --git a/unittest/observer/mvcc_trx_log_test.cpp b/unittest/observer/mvcc_trx_log_test.cpp
index bed22cbef..1657d795b 100644
--- a/unittest/observer/mvcc_trx_log_test.cpp
+++ b/unittest/observer/mvcc_trx_log_test.cpp
@@ -18,6 +18,9 @@ See the Mulan PSL v2 for more details. */
 #include <vector>
 #include <string>
 
+#define private public
+#include "common/value.h"
+#undef private
 #include "gtest/gtest.h"
 #include "storage/db/db.h"
 #include "storage/table/table.h"