diff --git a/src/observer/sql/executor/execute_stage.cpp b/src/observer/sql/executor/execute_stage.cpp
index d738c34b076096cc3f5db3d325cad4981e169b82..dfdc4f0267b1c00c0e821fa5bf586f73b4ee4fd6 100644
--- a/src/observer/sql/executor/execute_stage.cpp
+++ b/src/observer/sql/executor/execute_stage.cpp
@@ -25,7 +25,7 @@ See the Mulan PSL v2 for more details. */
 #include "event/storage_event.h"
 #include "event/sql_event.h"
 #include "event/session_event.h"
-#include "sql/executor/tuple.h"
+#include "sql/expr/tuple.h"
 #include "sql/operator/table_scan_operator.h"
 #include "sql/operator/predicate_operator.h"
 #include "sql/operator/delete_operator.h"
@@ -210,18 +210,18 @@ void end_trx_if_need(Session *session, Trx *trx, bool all_right)
   }
 }
 
-void print_tuple_header(std::ostream &os, const Operator &oper)
+void print_tuple_header(std::ostream &os, const ProjectOperator &oper)
 {
   const int cell_num = oper.tuple_cell_num();
-  TupleCellSpec cell_spec;
+  const TupleCellSpec *cell_spec = nullptr;
   for (int i = 0; i < cell_num; i++) {
     oper.tuple_cell_spec_at(i, cell_spec);
     if (i != 0) {
       os << " | ";
     }
 
-    if (cell_spec.alias()) {
-      os << cell_spec.alias();
+    if (cell_spec->alias()) {
+      os << cell_spec->alias();
     }
   }
 
diff --git a/src/observer/sql/expr/expression.cpp b/src/observer/sql/expr/expression.cpp
new file mode 100644
index 0000000000000000000000000000000000000000..0d50b2e1d73908fa3588179bc2048dd8e06e23ed
--- /dev/null
+++ b/src/observer/sql/expr/expression.cpp
@@ -0,0 +1,12 @@
+#include "sql/expr/tuple.h"
+
+
+RC FieldExpr::get_value(const Tuple &tuple, TupleCell &cell) const
+{
+  return tuple.find_cell(field_, cell);
+}
+
+RC ValueExpr::get_value(const Tuple &tuple, TupleCell & cell) const
+{
+  cell = tuple_cell;
+}
diff --git a/src/observer/sql/expr/expression.h b/src/observer/sql/expr/expression.h
new file mode 100644
index 0000000000000000000000000000000000000000..319caa8724e3bef633abe2a52ae09bb139c30d63
--- /dev/null
+++ b/src/observer/sql/expr/expression.h
@@ -0,0 +1,64 @@
+#pragma once
+
+#include "storage/common/field.h"
+#include "sql/expr/tuple_cell.h"
+
+class Tuple;
+
+class Expression
+{
+public: 
+  Expression() = default;
+  virtual ~Expression() = default;
+  
+  virtual RC get_value(const Tuple &tuple, TupleCell &cell) const = 0;
+};
+
+class FieldExpr : public Expression
+{
+public:
+  FieldExpr() = default;
+  FieldExpr(const Table *table, const FieldMeta *field) : field_(table, field)
+  {}
+
+  virtual ~FieldExpr() = default;
+
+  Field &field()
+  {
+    return field_;
+  }
+
+  const Field &field() const
+  {
+    return field_;
+  }
+
+  const char *table_name() const
+  {
+    return field_.table_name();
+  }
+
+  const char *field_name() const
+  {
+    return field_.field_name();
+  }
+
+  RC get_value(const Tuple &tuple, TupleCell &cell) const override;
+private:
+  Field field_;
+};
+
+class ValueExpr : public Expression
+{
+public:
+  ValueExpr() = default;
+  ValueExpr(const Value &value) : tuple_cell(value.type, (char *)value.data)
+  {}
+
+  virtual ~ValueExpr() = default;
+
+  RC get_value(const Tuple &tuple, TupleCell & cell) const override;
+
+private:
+  TupleCell tuple_cell;
+};
diff --git a/src/observer/sql/executor/tuple.h b/src/observer/sql/expr/tuple.h
similarity index 55%
rename from src/observer/sql/executor/tuple.h
rename to src/observer/sql/expr/tuple.h
index c6216356d97be9c1a3973a4b8863d7ca1eb409ce..e670fd25ad91bf3970baac41045cdc854bc56653 100644
--- a/src/observer/sql/executor/tuple.h
+++ b/src/observer/sql/expr/tuple.h
@@ -19,7 +19,8 @@ See the Mulan PSL v2 for more details. */
 
 #include "common/log/log.h"
 #include "sql/parser/parse.h"
-#include "storage/common/field.h"
+#include "sql/expr/tuple_cell.h"
+#include "sql/expr/expression.h"
 #include "storage/common/record.h"
 
 class Table;
@@ -28,48 +29,35 @@ class TupleCellSpec
 {
 public:
   TupleCellSpec() = default;
-  TupleCellSpec(const char *table_name, const char *field_name, const char *alias)
-    : table_name_(table_name), field_name_(field_name), alias_(alias)
+  TupleCellSpec(Expression *expr) : expression_(expr)
   {}
 
-  void set_table_name(const char *table_name)
+  ~TupleCellSpec()
   {
-    this->table_name_ = table_name;
-  }
-
-  void set_field_name(const char *field_name)
-  {
-    this->field_name_ = field_name;
+    if (expression_) {
+      delete expression_;
+      expression_ = nullptr;
+    }
   }
 
   void set_alias(const char *alias)
   {
     this->alias_ = alias;
   }
-  const char *table_name() const
-  {
-    return table_name_;
-  }
-  const char *field_name() const
-  {
-    return field_name_;
-  }
   const char *alias() const
   {
     return alias_;
   }
 
-  bool is_same_cell(const TupleCellSpec &other) const
+  Expression *expression() const
   {
-    return 0 == strcmp(this->table_name_, other.table_name_) &&
-      0 == strcmp(this->field_name_, other.field_name_);
+    return expression_;
   }
 
 private:
   // TODO table and field cannot describe all scenerio, should be expression
-  const char *table_name_ = nullptr;
-  const char *field_name_ = nullptr;
   const char *alias_ = nullptr;
+  Expression *expression_ = nullptr;
 };
 
 class Tuple
@@ -80,78 +68,82 @@ public:
 
   virtual int cell_num() const = 0; 
   virtual RC  cell_at(int index, TupleCell &cell) const = 0;
-  virtual RC  find_cell(const TupleCellSpec &spec, TupleCell &cell) const = 0;
+  virtual RC  find_cell(const Field &field, TupleCell &cell) const = 0;
 
-  virtual RC  cell_spec_at(int index, TupleCellSpec &spec) const = 0;
+  virtual RC  cell_spec_at(int index, const TupleCellSpec *&spec) const = 0;
 };
 
 class RowTuple : public Tuple
 {
 public:
   RowTuple() = default;
-  virtual ~RowTuple() = default;
+  virtual ~RowTuple()
+  {
+    for (TupleCellSpec *spec : speces_) {
+      delete spec;
+    }
+    speces_.clear();
+  }
   
   void set_record(Record *record)
   {
     this->record_ = record;
   }
 
-  void set_table(Table *table)
+  void set_schema(Table *table, const std::vector<FieldMeta> *fields)
   {
-    this->table_ = table;
-  }
-
-  void set_schema(const std::vector<FieldMeta> *fields)
-  {
-    this->fields_ = fields;
+    table_ = table;
+    this->speces_.reserve(fields->size());
+    for (const FieldMeta &field : *fields) {
+      speces_.push_back(new TupleCellSpec(new FieldExpr(table, &field)));
+    }
   }
 
   int cell_num() const override
   {
-    return fields_->size();
+    return speces_.size();
   }
 
   RC cell_at(int index, TupleCell &cell) const override
   {
-    if (index < 0 || index >= fields_->size()) {
+    if (index < 0 || index >= speces_.size()) {
       LOG_WARN("invalid argument. index=%d", index);
       return RC::INVALID_ARGUMENT;
     }
 
-    const FieldMeta &field_meta = (*fields_)[index];
-    cell.set_table(table_);
-    cell.set_type(field_meta.type());
-    cell.set_data(this->record_->data() + field_meta.offset());
+    const TupleCellSpec *spec = speces_[index];
+    FieldExpr *field_expr = (FieldExpr *)spec->expression();
+    const FieldMeta *field_meta = field_expr->field().meta();
+    cell.set_type(field_meta->type());
+    cell.set_data(this->record_->data() + field_meta->offset());
     return RC::SUCCESS;
   }
 
-  RC find_cell(const TupleCellSpec &spec, TupleCell &cell) const override
+  RC find_cell(const Field &field, TupleCell &cell) const override
   {
-    const char *table_name = spec.table_name();
+    const char *table_name = field.table_name();
     if (0 != strcmp(table_name, table_->name())) {
       return RC::NOTFOUND;
     }
 
-    const char *field_name = spec.field_name();
-    for (int i = 0; i < fields_->size(); ++i) {
-      const FieldMeta &field_meta = (*fields_)[i];
-      if (0 == strcmp(field_name, field_meta.name())) {
+    const char *field_name = field.field_name();
+    for (int i = 0; i < speces_.size(); ++i) {
+      const FieldExpr * field_expr = (const FieldExpr *)speces_[i]->expression();
+      const Field &field = field_expr->field();
+      if (0 == strcmp(field_name, field.field_name())) {
 	return cell_at(i, cell);
       }
     }
     return RC::NOTFOUND;
   }
 
-  RC cell_spec_at(int index, TupleCellSpec &spec) const override
+  RC cell_spec_at(int index, const TupleCellSpec *&spec) const override
   {
-    if (index < 0 || index >= fields_->size()) {
+    if (index < 0 || index >= speces_.size()) {
       LOG_WARN("invalid argument. index=%d", index);
       return RC::INVALID_ARGUMENT;
     }
-    const FieldMeta &field_meta = (*fields_)[index];
-    spec.set_table_name(table_->name());
-    spec.set_field_name(field_meta.name());
-    spec.set_alias(field_meta.name());
+    spec = speces_[index];
     return RC::SUCCESS;
   }
 
@@ -167,7 +159,7 @@ public:
 private:
   Record *record_ = nullptr;
   Table *table_ = nullptr;
-  const std::vector<FieldMeta> *fields_ = nullptr;
+  std::vector<TupleCellSpec *> speces_;
 };
 
 /*
@@ -186,13 +178,20 @@ class ProjectTuple : public Tuple
 {
 public:
   ProjectTuple() = default;
+  virtual ~ProjectTuple()
+  {
+    for (TupleCellSpec *spec : speces_) {
+      delete spec;
+    }
+    speces_.clear();
+  }
 
   void set_tuple(Tuple *tuple)
   {
     this->tuple_ = tuple;
   }
 
-  void add_cell_spec(const TupleCellSpec &spec)
+  void add_cell_spec(TupleCellSpec *spec)
   {
     speces_.push_back(spec);
   }
@@ -210,15 +209,15 @@ public:
       return RC::GENERIC_ERROR;
     }
 
-    const TupleCellSpec &spec = speces_[index]; // TODO better: add mapping between projection and raw tuple
-    return tuple_->find_cell(spec, cell);
+    const TupleCellSpec *spec = speces_[index];
+    return spec->expression()->get_value(*tuple_, cell);
   }
 
-  RC find_cell(const TupleCellSpec &spec, TupleCell &cell) const override
+  RC find_cell(const Field &field, TupleCell &cell) const override
   {
-    return tuple_->find_cell(spec, cell);
+    return tuple_->find_cell(field, cell);
   }
-  RC cell_spec_at(int index, TupleCellSpec &spec) const override
+  RC cell_spec_at(int index, const TupleCellSpec *&spec) const override
   {
     if (index < 0 || index >= speces_.size()) {
       return RC::NOTFOUND;
@@ -227,6 +226,6 @@ public:
     return RC::SUCCESS;
   }
 private:
-  std::vector<TupleCellSpec> speces_;
+  std::vector<TupleCellSpec *> speces_;
   Tuple *tuple_ = nullptr;
 };
diff --git a/src/observer/sql/expr/tuple_cell.cpp b/src/observer/sql/expr/tuple_cell.cpp
new file mode 100644
index 0000000000000000000000000000000000000000..0552a4e8ee95135792c1b0359a678dc7a4b21741
--- /dev/null
+++ b/src/observer/sql/expr/tuple_cell.cpp
@@ -0,0 +1,49 @@
+#include "sql/expr/tuple_cell.h"
+#include "storage/common/field.h"
+#include "common/log/log.h"
+#include "util/comparator.h"
+
+void TupleCell::to_string(std::ostream &os) const
+{
+  switch (attr_type_) {
+  case INTS: {
+    os << *(int *)data_;
+  } break;
+  case FLOATS: {
+    os << *(float *)data_;
+  } break;
+  case CHARS: {
+    for (int i = 0; i < 4; i++) { // the max length of CHARS is 4
+      if (data_[i] == '\0') {
+	break;
+      }
+      os << data_[i];
+    }
+  } break;
+  default: {
+    LOG_WARN("unsupported attr type: %d", attr_type_);
+  } break;
+  }
+}
+
+int TupleCell::compare(const TupleCell &other) const
+{
+  if (this->attr_type_ == other.attr_type_) {
+    switch (this->attr_type_) {
+    case INTS: return compare_int(this->data_, other.data_);
+    case FLOATS: return compare_float(this->data_, other.data_);
+    case CHARS: return compare_string(this->data_, other.data_, 4);
+    default: {
+      LOG_WARN("unsupported type: %d", this->attr_type_);
+    }
+    }
+  } else if (this->attr_type_ == INTS && other.attr_type_ == FLOATS) {
+    float this_data = *(int *)data_;
+    return compare_float(&this_data, other.data_);
+  } else if (this->attr_type_ == FLOATS && other.attr_type_ == INTS) {
+    float other_data = *(int *)other.data_;
+    return compare_float(data_, &other_data);
+  }
+  LOG_WARN("not supported");
+  return -1; // TODO return rc?
+}
diff --git a/src/observer/sql/expr/tuple_cell.h b/src/observer/sql/expr/tuple_cell.h
new file mode 100644
index 0000000000000000000000000000000000000000..c3d5d7dbbf48de75d9d0a5ec6d84a12004a157ca
--- /dev/null
+++ b/src/observer/sql/expr/tuple_cell.h
@@ -0,0 +1,30 @@
+#pragma once
+
+#include <iostream>
+#include "storage/common/table.h"
+#include "storage/common/field_meta.h"
+
+class TupleCell
+{
+public: 
+  TupleCell() = default;
+  
+  TupleCell(FieldMeta *meta, char *data)
+    : TupleCell(meta->type(), data)
+  {}
+  TupleCell(AttrType attr_type, char *data)
+    : attr_type_(attr_type), data_(data)
+  {}
+
+  void set_type(AttrType type) { this->attr_type_ = type; }
+  void set_data(char *data) { this->data_ = data; }
+  void set_data(const char *data) { this->set_data(const_cast<char *>(data)); }
+
+  void to_string(std::ostream &os) const;
+
+  int compare(const TupleCell &other) const;
+
+private:
+  AttrType attr_type_ = UNDEFINED;
+  char *data_ = nullptr; // real data. no need to move to field_meta.offset
+};
diff --git a/src/observer/sql/operator/delete_operator.h b/src/observer/sql/operator/delete_operator.h
index 57ee275ab770d681dd990db0714da11408fee7df..4b4365c166abd76209204fceeb4817cb18474a5f 100644
--- a/src/observer/sql/operator/delete_operator.h
+++ b/src/observer/sql/operator/delete_operator.h
@@ -35,15 +35,8 @@ public:
   Tuple * current_tuple() override {
     return nullptr;
   }
-  int tuple_cell_num() const override
-  {
-    return 0;
-  }
-
-  RC tuple_cell_spec_at(int index, TupleCellSpec &spec) const override
-  {
-    return RC::NOTFOUND;
-  }
+  //int tuple_cell_num() const override
+  //RC tuple_cell_spec_at(int index, TupleCellSpec &spec) const override
 private:
   DeleteStmt *delete_stmt_ = nullptr;
 };
diff --git a/src/observer/sql/operator/operator.h b/src/observer/sql/operator/operator.h
index 16c3afde21956e298ae0b4bb86a058bd35383c7a..f9da28f92473d35b2513239d054d22901a27c77d 100644
--- a/src/observer/sql/operator/operator.h
+++ b/src/observer/sql/operator/operator.h
@@ -16,7 +16,7 @@ See the Mulan PSL v2 for more details. */
 
 #include <vector>
 #include "rc.h"
-#include "sql/executor/tuple.h"
+#include "sql/expr/tuple.h"
 
 class Record;
 class TupleCellSpec;
@@ -34,8 +34,8 @@ public:
   virtual RC close() = 0;
 
   virtual Tuple * current_tuple() = 0;
-  virtual int tuple_cell_num() const = 0;
-  virtual RC  tuple_cell_spec_at(int index, TupleCellSpec &spec) const = 0;
+  //virtual int tuple_cell_num() const = 0;
+  //virtual RC  tuple_cell_spec_at(int index, TupleCellSpec *&spec) const = 0;
 
   void add_child(Operator *oper) {
     children_.push_back(oper);
diff --git a/src/observer/sql/operator/predicate_operator.cpp b/src/observer/sql/operator/predicate_operator.cpp
index 3eee630f751968e228ca38a99ff3144a87ac6fa4..2874263227d6353e52fb2e96d0df20b8706c0bd9 100644
--- a/src/observer/sql/operator/predicate_operator.cpp
+++ b/src/observer/sql/operator/predicate_operator.cpp
@@ -59,31 +59,20 @@ Tuple * PredicateOperator::current_tuple()
   return children_[0]->current_tuple();
 }
 
-void get_cell(const RowTuple &tuple, const FilterItem &filter_item, TupleCell &cell)
-{
-  if (filter_item.is_attr()) {
-    cell.set_data(tuple.record().data() + filter_item.field().field()->offset());
-    cell.set_type(filter_item.field().field()->type());
-  } else {
-    cell.set_data((char *)filter_item.value().data);
-    cell.set_type(filter_item.value().type);
-  }
-}
-
 bool PredicateOperator::do_predicate(RowTuple &tuple)
 {
   if (filter_stmt_ == nullptr || filter_stmt_->filter_units().empty()) {
     return true;
   }
 
-  for (const FilterUnit &filter_unit : filter_stmt_->filter_units()) {
-    const FilterItem & left = filter_unit.left();
-    const FilterItem & right = filter_unit.right();
-    CompOp comp = filter_unit.comp();
+  for (const FilterUnit *filter_unit : filter_stmt_->filter_units()) {
+    Expression *left_expr = filter_unit->left();
+    Expression *right_expr = filter_unit->right();
+    CompOp comp = filter_unit->comp();
     TupleCell left_cell;
     TupleCell right_cell;
-    get_cell(tuple, left, left_cell);
-    get_cell(tuple, right, right_cell);
+    left_expr->get_value(tuple, left_cell);
+    right_expr->get_value(tuple, right_cell);
 
     const int compare = left_cell.compare(right_cell);
     bool filter_result = false;
@@ -117,11 +106,11 @@ bool PredicateOperator::do_predicate(RowTuple &tuple)
   return true;
 }
 
-int PredicateOperator::tuple_cell_num() const
-{
-  return children_[0]->tuple_cell_num();
-}
-RC PredicateOperator::tuple_cell_spec_at(int index, TupleCellSpec &spec) const
-{
-  return children_[0]->tuple_cell_spec_at(index, spec);
-}
+// int PredicateOperator::tuple_cell_num() const
+// {
+//   return children_[0]->tuple_cell_num();
+// }
+// RC PredicateOperator::tuple_cell_spec_at(int index, TupleCellSpec &spec) const
+// {
+//   return children_[0]->tuple_cell_spec_at(index, spec);
+// }
diff --git a/src/observer/sql/operator/predicate_operator.h b/src/observer/sql/operator/predicate_operator.h
index 1c375aafcfcca076e80b0551f6430868340f1991..f92b486159bc3dfbe0deee2b9474f28d593dae01 100644
--- a/src/observer/sql/operator/predicate_operator.h
+++ b/src/observer/sql/operator/predicate_operator.h
@@ -17,6 +17,11 @@ See the Mulan PSL v2 for more details. */
 #include "sql/operator/operator.h"
 
 class FilterStmt;
+
+/**
+ * PredicateOperator 用于单个表中的记录过滤
+ * 如果是多个表数据过滤，比如join条件的过滤，需要设计新的predicate或者扩展:w
+ */
 class PredicateOperator : public Operator
 {
 public:
@@ -31,8 +36,8 @@ public:
   RC close() override;
 
   Tuple * current_tuple() override;
-  int tuple_cell_num() const override;
-  RC tuple_cell_spec_at(int index, TupleCellSpec &spec) const override;
+  //int tuple_cell_num() const override;
+  //RC tuple_cell_spec_at(int index, TupleCellSpec &spec) const override;
 private:
   bool do_predicate(RowTuple &tuple);
 private:
diff --git a/src/observer/sql/operator/project_operator.cpp b/src/observer/sql/operator/project_operator.cpp
index 62688e1d9660b3b477d82cf28c26a7d8782e62ee..3d7878264ae05099aa03a73fb73972d7355f924f 100644
--- a/src/observer/sql/operator/project_operator.cpp
+++ b/src/observer/sql/operator/project_operator.cpp
@@ -52,11 +52,14 @@ Tuple *ProjectOperator::current_tuple()
 
 void ProjectOperator::add_projection(const Table *table, const FieldMeta *field_meta)
 {
-  TupleCellSpec spec(table->name(), field_meta->name(), field_meta->name());
+  // 对单表来说，展示的(alias) 字段总是字段名称，
+  // 对多表查询来说，展示的alias 需要带表名字
+  TupleCellSpec *spec = new TupleCellSpec(new FieldExpr(table, field_meta));
+  spec->set_alias(field_meta->name());
   tuple_.add_cell_spec(spec);
 }
 
-RC ProjectOperator::tuple_cell_spec_at(int index, TupleCellSpec &spec) const
+RC ProjectOperator::tuple_cell_spec_at(int index, const TupleCellSpec *&spec) const
 {
   return tuple_.cell_spec_at(index, spec);
 }
diff --git a/src/observer/sql/operator/project_operator.h b/src/observer/sql/operator/project_operator.h
index 599625eca406184c8c8ad184250de7fd6631b1a4..64e9d99b784f689caf1dd584111c97f7d2f4cc94 100644
--- a/src/observer/sql/operator/project_operator.h
+++ b/src/observer/sql/operator/project_operator.h
@@ -31,12 +31,12 @@ public:
   RC next() override;
   RC close() override;
 
-  int tuple_cell_num() const override
+  int tuple_cell_num() const
   {
     return tuple_.cell_num();
   }
 
-  RC tuple_cell_spec_at(int index, TupleCellSpec &spec) const override;
+  RC tuple_cell_spec_at(int index, const TupleCellSpec *&spec) const;
 
   Tuple * current_tuple() override;
 private:
diff --git a/src/observer/sql/operator/table_scan_operator.cpp b/src/observer/sql/operator/table_scan_operator.cpp
index 637fc778043581c9227b5830007ae217f31b90ca..cdd0424ce27dd0e25d07c2049749021788ac8e48 100644
--- a/src/observer/sql/operator/table_scan_operator.cpp
+++ b/src/observer/sql/operator/table_scan_operator.cpp
@@ -20,8 +20,7 @@ RC TableScanOperator::open()
 {
   RC rc = table_->get_record_scanner(record_scanner_);
   if (rc == RC::SUCCESS) {
-    tuple_.set_table(table_);
-    tuple_.set_schema(table_->table_meta().field_metas());
+    tuple_.set_schema(table_, table_->table_meta().field_metas());
   }
   return rc;
 }
@@ -46,7 +45,7 @@ Tuple * TableScanOperator::current_tuple()
   tuple_.set_record(&current_record_);
   return &tuple_;
 }
-RC TableScanOperator::tuple_cell_spec_at(int index, TupleCellSpec &spec) const
-{
-  return tuple_.cell_spec_at(index, spec);
-}
+// RC TableScanOperator::tuple_cell_spec_at(int index, TupleCellSpec &spec) const
+// {
+//   return tuple_.cell_spec_at(index, spec);
+// }
diff --git a/src/observer/sql/operator/table_scan_operator.h b/src/observer/sql/operator/table_scan_operator.h
index 5848f67834235ec025f076046065db54788d94a1..5d7ef81e42a07b60210d348de4a597bf7a0d4862 100644
--- a/src/observer/sql/operator/table_scan_operator.h
+++ b/src/observer/sql/operator/table_scan_operator.h
@@ -35,12 +35,12 @@ public:
 
   Tuple * current_tuple() override;
 
-  int tuple_cell_num() const override
-  {
-    return tuple_.cell_num();
-  }
+  // int tuple_cell_num() const override
+  // {
+  //   return tuple_.cell_num();
+  // }
 
-  RC tuple_cell_spec_at(int index, TupleCellSpec &spec) const override;
+  // RC tuple_cell_spec_at(int index, TupleCellSpec &spec) const override;
 private:
   Table *table_ = nullptr;
   RecordFileScanner record_scanner_;
diff --git a/src/observer/sql/stmt/delete_stmt.cpp b/src/observer/sql/stmt/delete_stmt.cpp
index e5b552973ee592b79097e7be4f2ee736ff4b2935..114e4f85ce7a6fa9f708cf024fc72382c0e6de69 100644
--- a/src/observer/sql/stmt/delete_stmt.cpp
+++ b/src/observer/sql/stmt/delete_stmt.cpp
@@ -46,8 +46,12 @@ RC DeleteStmt::create(Db *db, const Deletes &delete_sql, Stmt *&stmt)
     return RC::SCHEMA_TABLE_NOT_EXIST;
   }
 
+  std::unordered_map<std::string_view, Table *> table_map;
+  table_map.insert(std::pair<std::string_view, Table *>(std::string_view(table_name), table));
+
   FilterStmt *filter_stmt = nullptr;
-  RC rc = FilterStmt::create(db, table, delete_sql.conditions, delete_sql.condition_num, filter_stmt);
+  RC rc = FilterStmt::create(db, table, &table_map,
+			     delete_sql.conditions, delete_sql.condition_num, filter_stmt);
   if (rc != RC::SUCCESS) {
     LOG_WARN("failed to create filter statement. rc=%d:%s", rc, strrc(rc));
     return rc;
diff --git a/src/observer/sql/stmt/filter_stmt.cpp b/src/observer/sql/stmt/filter_stmt.cpp
index 34194f0ff7e824c4f366374bb47ef46441bc0ad1..cbc4ece824a8e964152335e0ffd5e2676beee314 100644
--- a/src/observer/sql/stmt/filter_stmt.cpp
+++ b/src/observer/sql/stmt/filter_stmt.cpp
@@ -19,7 +19,15 @@ See the Mulan PSL v2 for more details. */
 #include "storage/common/db.h"
 #include "storage/common/table.h"
 
-RC FilterStmt::create(Db *db, Table *default_table,
+FilterStmt::~FilterStmt()
+{
+  for (FilterUnit *unit : filter_units_) {
+    delete unit;
+  }
+  filter_units_.clear();
+}
+
+RC FilterStmt::create(Db *db, Table *default_table, std::unordered_map<std::string_view, Table *> *tables,
 		      const Condition *conditions, int condition_num,
 		      FilterStmt *&stmt)
 {
@@ -28,8 +36,8 @@ RC FilterStmt::create(Db *db, Table *default_table,
 
   FilterStmt *tmp_stmt = new FilterStmt();
   for (int i = 0; i < condition_num; i++) {
-    FilterUnit filter_unit;
-    rc = create_filter_unit(db, default_table, conditions[i], filter_unit);
+    FilterUnit *filter_unit = nullptr;
+    rc = create_filter_unit(db, default_table, tables, conditions[i], filter_unit);
     if (rc != RC::SUCCESS) {
       delete tmp_stmt;
       LOG_WARN("failed to create filter unit. condition index=%d", i);
@@ -42,10 +50,16 @@ RC FilterStmt::create(Db *db, Table *default_table,
   return rc;
 }
 
-RC get_table_and_field(Db *db, Table *default_table, const RelAttr &attr, Table *&table, const FieldMeta *&field)
+RC get_table_and_field(Db *db, Table *default_table, std::unordered_map<std::string_view, Table *> *tables,
+		       const RelAttr &attr, Table *&table, const FieldMeta *&field)
 {
   if (common::is_blank(attr.relation_name)) {
     table = default_table;
+  } else if (nullptr != tables) {
+    auto iter = tables->find(std::string_view(attr.relation_name));
+    if (iter != tables->end()) {
+      table = iter->second;
+    }
   } else {
     table = db->find_table(attr.relation_name);
   }
@@ -64,8 +78,8 @@ RC get_table_and_field(Db *db, Table *default_table, const RelAttr &attr, Table
   return RC::SUCCESS;
 }
 
-RC FilterStmt::create_filter_unit(Db *db, Table *default_table,
-				  const Condition &condition, FilterUnit &filter_unit)
+RC FilterStmt::create_filter_unit(Db *db, Table *default_table, std::unordered_map<std::string_view, Table *> *tables,
+				  const Condition &condition, FilterUnit *&filter_unit)
 {
   RC rc = RC::SUCCESS;
   
@@ -75,38 +89,40 @@ RC FilterStmt::create_filter_unit(Db *db, Table *default_table,
     return RC::INVALID_ARGUMENT;
   }
 
-  filter_unit.set_comp(comp);
-  FilterItem &left_item = filter_unit.left();
-  FilterItem &right_item = filter_unit.right();
-  
+  Expression *left = nullptr;
+  Expression *right = nullptr;
   if (condition.left_is_attr) {
     Table *table = nullptr;
     const FieldMeta *field = nullptr;
-    rc = get_table_and_field(db, default_table, condition.left_attr, table, field);  
+    rc = get_table_and_field(db, default_table, tables, condition.left_attr, table, field);  
     if (rc != RC::SUCCESS) {
       LOG_WARN("cannot find attr");
       return rc;
     }
-    left_item.set_field(table, field);
+    left = new FieldExpr(table, field);
   } else {
-    left_item.set_value(condition.left_value);
+    left = new ValueExpr(condition.left_value);
   }
 
   if (condition.right_is_attr) {
     Table *table = nullptr;
     const FieldMeta *field = nullptr;
-    rc = get_table_and_field(db, default_table, condition.right_attr, table, field);  
+    rc = get_table_and_field(db, default_table, tables, condition.right_attr, table, field);  
     if (rc != RC::SUCCESS) {
       LOG_WARN("cannot find attr");
+      delete left;
       return rc;
     }
-    right_item.set_field(table, field);
+    right = new FieldExpr(table, field);
   } else {
-    right_item.set_value(condition.right_value);
+    right = new ValueExpr(condition.right_value);
   }
 
+  filter_unit = new FilterUnit;
+  filter_unit->set_comp(comp);
+  filter_unit->set_left(left);
+  filter_unit->set_right(right);
+
   // 检查两个类型是否能够比较
   return rc;
 }
-
-
diff --git a/src/observer/sql/stmt/filter_stmt.h b/src/observer/sql/stmt/filter_stmt.h
index a87462d00cb598b70a4452115f9c9bd881800c17..b5c6f64572e4abca6094845544ffdc4eae7d6ba5 100644
--- a/src/observer/sql/stmt/filter_stmt.h
+++ b/src/observer/sql/stmt/filter_stmt.h
@@ -15,79 +15,31 @@ See the Mulan PSL v2 for more details. */
 #pragma once
 
 #include <vector>
+#include <unordered_map>
 #include "rc.h"
 #include "sql/parser/parse_defs.h"
 #include "sql/stmt/stmt.h"
+#include "sql/expr/expression.h"
 
 class Db;
 class Table;
 class FieldMeta;
 
-class FilterField
-{
-public:
-  FilterField() = default;
-
-  FilterField(Table *table, FieldMeta *field) : table_(table), field_(field)
-  {}
-
-  Table *table() const {
-    return table_;
-  }
-
-  const FieldMeta *field() const {
-    return field_;
-  }
-
-  void set_table(Table *table) {
-    table_ = table;
-  }
-  void set_field(const FieldMeta *field) {
-    field_ = field;
-  }
-private:
-  Table *table_ = nullptr;
-  const FieldMeta *field_ = nullptr;
-};
-
-class FilterItem
-{
-public:
-  FilterItem() = default;
-
-  void set_field(Table *table, const FieldMeta *field) {
-    is_attr_ = true;
-    field_.set_table(table);
-    field_.set_field(field);
-  }
-
-  void set_value(const Value &value) {
-    is_attr_ = false;
-    value_ = value;
-  }
-
-  bool is_attr() const {
-    return is_attr_;
-  }
-
-  const FilterField &field() const {
-    return field_;
-  }
-
-  const Value &value() const {
-    return value_;
-  }
-
-private:
-  bool is_attr_ = false; // is an attribute or a value, or maybe an expression in future
-  FilterField field_;
-  Value value_;
-};
-
 class FilterUnit
 {
 public:
   FilterUnit() = default;
+  ~FilterUnit()
+  {
+    if (left_) {
+      delete left_;
+      left_ = nullptr;
+    }
+    if (right_) {
+      delete right_;
+      right_ = nullptr;
+    }
+  }
   
   void set_comp(CompOp comp) {
     comp_ = comp;
@@ -96,22 +48,28 @@ public:
   CompOp comp() const {
     return comp_;
   }
-  FilterItem &left() {
-    return left_;
+
+  void set_left(Expression *expr)
+  {
+    left_ = expr;
   }
-  FilterItem &right() {
-    return right_;
+  void set_right(Expression *expr)
+  {
+    right_ = expr;
   }
-  const FilterItem &left() const {
+  Expression *left() const
+  {
     return left_;
   }
-  const FilterItem &right() const {
+  Expression *right() const
+  {
     return right_;
   }
+
 private:
   CompOp comp_ = NO_OP;
-  FilterItem left_;
-  FilterItem right_;
+  Expression *left_ = nullptr;
+  Expression *right_ = nullptr;
 };
 
 class FilterStmt 
@@ -119,20 +77,22 @@ class FilterStmt
 public:
 
   FilterStmt() = default;
+  virtual ~FilterStmt();
 
 public:
-  const std::vector<FilterUnit> &filter_units() const
+  const std::vector<FilterUnit *> &filter_units() const
   {
     return filter_units_;
   }
 
 public:
-  static RC create(Db *db, Table *default_table,
+  static RC create(Db *db, Table *default_table, std::unordered_map<std::string_view, Table *> *tables,
 			const Condition *conditions, int condition_num,
 			FilterStmt *&stmt);
 
-  static RC create_filter_unit(Db *db, Table *default_table, const Condition &condition, FilterUnit &filter_unit);
+  static RC create_filter_unit(Db *db, Table *default_table, std::unordered_map<std::string_view, Table *> *tables,
+			       const Condition &condition, FilterUnit *&filter_unit);
 
 private:
-  std::vector<FilterUnit>  filter_units_; // 默认当前都是AND关系
+  std::vector<FilterUnit *>  filter_units_; // 默认当前都是AND关系
 };
diff --git a/src/observer/sql/stmt/select_stmt.cpp b/src/observer/sql/stmt/select_stmt.cpp
index 0d519cf5c386eeeacf0de8d8c207e22541f4991e..4bc7dfe9dc24bc0735a43379a5d3ce93cca4a7dd 100644
--- a/src/observer/sql/stmt/select_stmt.cpp
+++ b/src/observer/sql/stmt/select_stmt.cpp
@@ -43,8 +43,9 @@ RC SelectStmt::create(Db *db, const Selects &select_sql, Stmt *&stmt)
     return RC::INVALID_ARGUMENT;
   }
 
+  // collect tables in `from` statement
   std::vector<Table *> tables;
-  std::map<std::string, Table *> table_map;
+  std::unordered_map<std::string_view, Table *> table_map;
   for (int i = 0; i < select_sql.relation_num; i++) {
     const char *table_name = select_sql.relations[i];
     if (nullptr == table_name) {
@@ -62,8 +63,9 @@ RC SelectStmt::create(Db *db, const Selects &select_sql, Stmt *&stmt)
     table_map.insert(std::pair<std::string, Table*>(table_name, table));
   }
   
+  // collect query fields in `select` statement
   std::vector<FieldDesc> query_fields;
-  for (int i = 0; i < select_sql.attr_num; i++) {
+  for (int i = select_sql.attr_num - 1; i >= 0; i--) {
     const RelAttr &relation_attr = select_sql.attributes[i];
 
     if (common::is_blank(relation_attr.relation_name) && 0 == strcmp(relation_attr.attribute_name, "*")) {
@@ -127,34 +129,15 @@ RC SelectStmt::create(Db *db, const Selects &select_sql, Stmt *&stmt)
     default_table = tables[0];
   }
 
+  // create filter statement in `where` statement
   FilterStmt *filter_stmt = nullptr;
-  RC rc = FilterStmt::create(db, default_table, select_sql.conditions, select_sql.condition_num, filter_stmt);
+  RC rc = FilterStmt::create(db, default_table, &table_map,
+			     select_sql.conditions, select_sql.condition_num, filter_stmt);
   if (rc != RC::SUCCESS) {
     LOG_WARN("cannot construct filter stmt");
     return rc;
   }
 
-  // make sure all tables in predicate are exists in from stmt
-  const std::vector<FilterUnit> &filter_units = filter_stmt->filter_units();
-  for (const FilterUnit &filter_unit : filter_units) {
-    const FilterItem &left = filter_unit.left();
-    const FilterItem &right = filter_unit.right();
-    if (left.is_attr()) {
-      Table *table = left.field().table();
-      if (table_map.find(table->name()) == table_map.end()) {
-	LOG_WARN("the table in predicate is not in from stmt: %s", table->name());
-	return RC::SCHEMA_TABLE_NOT_EXIST;
-      }
-    }
-    if (right.is_attr()) {
-      Table *table = right.field().table();
-      if (table_map.find(table->name()) == table_map.end()) {
-	LOG_WARN("the table in predicate is not in from stmt: %s", table->name());
-	return RC::SCHEMA_TABLE_NOT_EXIST;
-      }
-    }
-  }
-
   // everything alright
   SelectStmt *select_stmt = new SelectStmt();
   select_stmt->tables_.swap(tables);
diff --git a/src/observer/storage/common/field.cpp b/src/observer/storage/common/field.cpp
index 8f662926a4b0677c6564381dc9bf94d66030d6c5..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 100644
--- a/src/observer/storage/common/field.cpp
+++ b/src/observer/storage/common/field.cpp
@@ -1,48 +0,0 @@
-#include "storage/common/field.h"
-#include "common/log/log.h"
-#include "util/comparator.h"
-
-void TupleCell::to_string(std::ostream &os) const
-{
-  switch (attr_type_) {
-  case INTS: {
-    os << *(int *)data_;
-  } break;
-  case FLOATS: {
-    os << *(float *)data_;
-  } break;
-  case CHARS: {
-    for (int i = 0; i < 4; i++) { // the max length of CHARS is 4
-      if (data_[i] == '\0') {
-	break;
-      }
-      os << data_[i];
-    }
-  } break;
-  default: {
-    LOG_WARN("unsupported attr type: %d", attr_type_);
-  } break;
-  }
-}
-
-int TupleCell::compare(const TupleCell &other) const
-{
-  if (this->attr_type_ == other.attr_type_) {
-    switch (this->attr_type_) {
-    case INTS: return compare_int(this->data_, other.data_);
-    case FLOATS: return compare_float(this->data_, other.data_);
-    case CHARS: return compare_string(this->data_, other.data_, 4);
-    default: {
-      LOG_WARN("unsupported type: %d", this->attr_type_);
-    }
-    }
-  } else if (this->attr_type_ == INTS && other.attr_type_ == FLOATS) {
-    float this_data = *(int *)data_;
-    return compare_float(&this_data, other.data_);
-  } else if (this->attr_type_ == FLOATS && other.attr_type_ == INTS) {
-    float other_data = *(int *)other.data_;
-    return compare_float(data_, &other_data);
-  }
-  LOG_WARN("not supported");
-  return -1; // TODO return rc?
-}
diff --git a/src/observer/storage/common/field.h b/src/observer/storage/common/field.h
index 317dae70f2df280ebba6ea1b96f3e00d9fb7c82d..befacbc8d70412f87fff7439a4849775c58168a7 100644
--- a/src/observer/storage/common/field.h
+++ b/src/observer/storage/common/field.h
@@ -1,31 +1,35 @@
 #pragma once
 
-#include <iostream>
 #include "storage/common/table.h"
 #include "storage/common/field_meta.h"
 
-class TupleCell // TODO rename to Cell
+class Field
 {
-public: 
-  TupleCell() = default;
-  
-  TupleCell(FieldMeta *meta, char *data) : TupleCell(nullptr, meta, data)
-  {}
-  TupleCell(Table *table, FieldMeta *meta, char *data)
-    : table_(table), attr_type_(meta->type()), data_(data)
+public:
+  Field() = default;
+  Field(const Table *table, const FieldMeta *field) : table_(table), field_(field)
   {}
 
-  void set_table(Table *table) { this->table_ = table; }
-  void set_type(AttrType type) { this->attr_type_ = type; }
-  void set_data(char *data) { this->data_ = data; }
-  void set_data(const char *data) { this->set_data(const_cast<char *>(data)); }
+  const Table *table() const { return table_; }
+  const FieldMeta *meta() const { return field_; }
 
-  void to_string(std::ostream &os) const;
+  AttrType attr_type() const
+  {
+    return field_->type();
+  }
 
-  int compare(const TupleCell &other) const;
+  const char *table_name() const { return table_->name(); }
+  const char *field_name() const { return field_->name(); }
 
+  void set_table(const Table *table)
+  {
+    this->table_ = table;
+  }
+  void set_field(const FieldMeta *field)
+  {
+    this->field_ = field;
+  }
 private:
-  Table *table_ = nullptr;
-  AttrType attr_type_ = UNDEFINED;
-  char *data_ = nullptr; // real data. no need to move to field_meta.offset
+  const Table *table_ = nullptr;
+  const FieldMeta *field_ = nullptr;
 };