update yangqi demo0.2

src/observer/sql/operator/logical_operator.h

@@ -40,6 +40,7 @@ enum class LogicalOperatorType
   INSERT,     ///< 插入
   DELETE,     ///< 删除，删除可能会有子查询
   EXPLAIN,    ///< 查看执行计划
+  UPDATE,     
 };
 
 /**

src/observer/sql/operator/physical_operator.h

@@ -47,6 +47,7 @@ enum class PhysicalOperatorType
   STRING_LIST,
   DELETE,
   INSERT,
+  UPDATE,
 };
 
 /**

src/observer/sql/operator/update_logical_operator.cpp

@@ -0,0 +1,6 @@
+#include "sql/operator/update_logical_operator.h"
+
+UpdateLogicalOperator::UpdateLogicalOperator(Table *table, Value *value)
+    : table_(table), value_(value)
+{
+}

src/observer/sql/operator/update_logical_operator.h

@@ -0,0 +1,25 @@
+#pragma once
+
+
+#include "sql/operator/logical_operator.h"
+#include "sql/parser/parse_defs.h"
+
+class UpdateLogicalOperator : public LogicalOperator
+{
+public:
+  UpdateLogicalOperator(Table *table, Value* value);
+  virtual ~UpdateLogicalOperator() = default;
+
+  LogicalOperatorType type() const override
+  {
+    return LogicalOperatorType::UPDATE;
+  }
+
+  Table *table() const { return table_; }
+  const Value *values() const { return value_; }
+  Value *values() { return value_; }
+
+private:
+  Table *table_ = nullptr;
+  Value* value_;
+};

src/observer/sql/operator/update_physical_operator.cpp

@@ -0,0 +1,62 @@
+#include "sql/operator/update_physical_operator.h"
+#include "sql/stmt/update_stmt.h"
+#include "storage/table/table.h"
+#include "storage/trx/trx.h"
+
+using namespace std;
+
+UpdatePhysicalOperator::UpdatePhysicalOperator(Table *table, Value *value)
+    : table_(table), value_(value)
+{}
+RC UpdatePhysicalOperator::open(Trx *trx)
+{
+  if (children_.empty()) {
+    return RC::SUCCESS;
+  }
+
+  std::unique_ptr<PhysicalOperator> &child = children_[0];
+  RC rc = child->open(trx);
+  if (rc != RC::SUCCESS) {
+    LOG_WARN("failed to open child operator: %s", strrc(rc));
+    return rc;
+  }
+
+  trx_ = trx;
+
+  return RC::SUCCESS;
+}
+
+RC UpdatePhysicalOperator::next()
+{
+  RC rc = RC::SUCCESS;
+  if (children_.empty()) {
+    return RC::RECORD_EOF;
+  }
+
+  PhysicalOperator *child = children_[0].get();
+  while (RC::SUCCESS == (rc = child->next())) {
+    Tuple *tuple = child->current_tuple();
+    if (nullptr == tuple) {
+      LOG_WARN("failed to get current record: %s", strrc(rc));
+      return rc;
+    }
+
+    RowTuple *row_tuple = static_cast<RowTuple *>(tuple);
+    Record &record = row_tuple->record();
+    rc = trx_->update_record(table_, record);
+    if (rc != RC::SUCCESS) {
+      LOG_WARN("failed to delete record: %s", strrc(rc));
+      return rc;
+    }
+  }
+
+  return RC::RECORD_EOF;
+}
+
+RC UpdatePhysicalOperator::close()
+{
+  if (!children_.empty()) {
+    children_[0]->close();
+  }
+  return RC::SUCCESS;
+}

src/observer/sql/operator/update_physical_operator.h

@@ -0,0 +1,30 @@
+#pragma once
+
+#include "sql/operator/physical_operator.h"
+#include "sql/parser/parse.h"
+
+class UpdateStmt;
+
+class UpdatePhysicalOperator : public PhysicalOperator
+{
+public:
+  UpdatePhysicalOperator(Table *table, Value *value);
+
+  virtual ~UpdatePhysicalOperator() = default;
+
+  PhysicalOperatorType type() const override
+  {
+    return PhysicalOperatorType::UPDATE;
+  }
+
+  RC open(Trx *trx) override;
+  RC next() override;
+  RC close() override;
+
+  Tuple *current_tuple() override { return nullptr; }
+
+private:
+  Table *table_ = nullptr;
+  Value *value_ = nullptr;
+  Trx *trx_ = nullptr;
+};

src/observer/storage/table/table.cpp

@@ -533,6 +533,7 @@ RC Table::delete_entry_of_indexes(const char *record, const RID &rid, bool error
 }
 RC Table::update_entry_of_indexes(const char *new_record, const char *old_record, const RID &rid)
 {
+  // TODO
   RC rc = RC::SUCCESS;
   for (Index *index : indexes_) {
     // 首先删除旧记录

src/observer/storage/trx/mvcc_trx.cpp

@@ -189,8 +189,15 @@ RC MvccTrx::update_record(Table * table, Record &record)
   Field begin_field;
   Field end_field;
   trx_fields(table, begin_field, end_field);
-  begin_field.set_int(record, -trx_id_);
-  end_field.set_int(record, trx_kit_.max_trx_id());
+
+  [[maybe_unused]] int32_t end_xid = end_field.get_int(record);
+  /// 在修改之前，第一次获取record时，就已经对record做了对应的检查，并且保证不会有其它的事务来访问这条数据
+  ASSERT(end_xid > 0, "concurrency conflit: other transaction is updating this record. end_xid=%d, current trx id=%d, rid=%s",
+         end_xid, trx_id_, record.rid().to_string().c_str());
+  if (end_xid != trx_kit_.max_trx_id()) {
+    // 当前不是多版本数据中的最新记录，不需要修改
+    return RC::SUCCESS;
+  }
 
   RC rc = table->update_record(record);
   if (rc != RC::SUCCESS) {