[FEAT MERGE] transaction pdml support optimization

Co-authored-by: yyy-hust <yyy.hust@gmail.com>
Co-authored-by: SanmuWangZJU <sanmuwang.ws@gmail.com>

src/sql/das/ob_das_context.cpp

@@ -571,6 +571,7 @@ OB_DEF_SERIALIZE(ObDASCtx)
   OB_UNIS_ENCODE(flags_);
   OB_UNIS_ENCODE(snapshot_);
   OB_UNIS_ENCODE(location_router_);
+  OB_UNIS_ENCODE(write_branch_id_);
   return ret;
 }
 
@@ -601,6 +602,7 @@ OB_DEF_DESERIALIZE(ObDASCtx)
     LOG_WARN("rebuild tablet loc reference failed", K(ret));
   }
   OB_UNIS_DECODE(location_router_);
+  OB_UNIS_DECODE(write_branch_id_);
   return ret;
 }
 
@@ -615,6 +617,7 @@ OB_DEF_SERIALIZE_SIZE(ObDASCtx)
   OB_UNIS_ADD_LEN(flags_);
   OB_UNIS_ADD_LEN(snapshot_);
   OB_UNIS_ADD_LEN(location_router_);
+  OB_UNIS_ADD_LEN(write_branch_id_);
   return len;
 }
 }  // namespace sql

src/sql/das/ob_das_context.h

@@ -50,6 +50,7 @@ public:
       allocator_(allocator),
       snapshot_(),
       savepoint_(),
+      write_branch_id_(0),
       del_ctx_list_(allocator),
       jump_read_group_id_(-1),
       flags_(0)
@@ -89,6 +90,8 @@ public:
   transaction::ObTxReadSnapshot &get_snapshot() { return snapshot_; }
   transaction::ObTxSEQ get_savepoint() const { return savepoint_; }
   void set_savepoint(const transaction::ObTxSEQ savepoint) { savepoint_ = savepoint; }
+  void set_write_branch_id(const int16_t branch_id) { write_branch_id_ = branch_id; }
+  int16_t get_write_branch_id() const { return write_branch_id_; }
   ObDASLocationRouter &get_location_router() { return location_router_; }
   int build_related_tablet_loc(ObDASTabletLoc &tablet_loc);
   int build_related_table_loc(ObDASTableLoc &table_loc);
@@ -114,7 +117,8 @@ public:
                K_(external_table_locs),
                K_(is_fk_cascading),
                K_(snapshot),
-               K_(savepoint));
+               K_(savepoint),
+               K_(write_branch_id));
 private:
   int check_same_server(const ObDASTabletLoc *tablet_loc);
 private:
@@ -131,6 +135,11 @@ private:
   common::ObIAllocator &allocator_;
   transaction::ObTxReadSnapshot snapshot_;           // Mvcc snapshot
   transaction::ObTxSEQ savepoint_;                   // DML savepoint
+  // for DML like `insert update` and `replace`, which use savepoint to
+  // resolve conflicts and when these DML executed under partition-wise
+  // style, they need rollback their own writes but not all, we assign
+  // id to data writes by different writer thread (named branch)
+  int16_t write_branch_id_;
   //@todo: save snapshot version
   DASDelCtxList del_ctx_list_;
 public:

src/sql/das/ob_das_delete_op.cpp

@@ -98,6 +98,7 @@ int ObDASDeleteOp::open_op()
   ObDASIndexDMLAdaptor<DAS_OP_TABLE_DELETE, ObDASDMLIterator> del_adaptor;
   del_adaptor.tx_desc_ = trans_desc_;
   del_adaptor.snapshot_ = snapshot_;
+  del_adaptor.write_branch_id_ = write_branch_id_;
   del_adaptor.ctdef_ = del_ctdef_;
   del_adaptor.rtdef_ = del_rtdef_;
   del_adaptor.related_ctdefs_ = &related_ctdefs_;

src/sql/das/ob_das_insert_op.cpp

@@ -127,6 +127,7 @@ int ObDASInsertOp::insert_rows()
   ObDASIndexDMLAdaptor<DAS_OP_TABLE_INSERT, ObDASDMLIterator> ins_adaptor;
   ins_adaptor.tx_desc_ = trans_desc_;
   ins_adaptor.snapshot_ = snapshot_;
+  ins_adaptor.write_branch_id_ = write_branch_id_;
   ins_adaptor.ctdef_ = ins_ctdef_;
   ins_adaptor.rtdef_ = ins_rtdef_;
   ins_adaptor.related_ctdefs_ = &related_ctdefs_;
@@ -159,7 +160,7 @@ int ObDASInsertOp::insert_row_with_fetch()
   if (ins_ctdef_->table_rowkey_types_.empty()) {
     ret = OB_ERR_UNEXPECTED;
     LOG_WARN("table_rowkey_types is invalid", K(ret));
-  } else if (OB_FAIL(ObDMLService::init_dml_param(*ins_ctdef_, *ins_rtdef_, *snapshot_, op_alloc_, dml_param))) {
+  } else if (OB_FAIL(ObDMLService::init_dml_param(*ins_ctdef_, *ins_rtdef_, *snapshot_, write_branch_id_, op_alloc_, dml_param))) {
     LOG_WARN("init dml param failed", K(ret), KPC_(ins_ctdef), KPC_(ins_rtdef));
   } else if (OB_ISNULL(buf = op_alloc_.alloc(sizeof(ObDASConflictIterator)))) {
     ret = OB_ERR_UNEXPECTED;
@@ -210,6 +211,7 @@ int ObDASInsertOp::insert_row_with_fetch()
     } else if (OB_FAIL(ObDMLService::init_dml_param(*index_ins_ctdef,
                                                     *index_ins_rtdef,
                                                     *snapshot_,
+                                                    write_branch_id_,
                                                     op_alloc_,
                                                     dml_param))) {
       LOG_WARN("init index dml param failed", K(ret), KPC(index_ins_ctdef), KPC(index_ins_rtdef));

src/sql/das/ob_das_lock_op.cpp

@@ -55,7 +55,7 @@ int ObDASLockOp::open_op()
 
   ObDASDMLIterator dml_iter(lock_ctdef_, lock_buffer_, op_alloc_);
   ObAccessService *as = MTL(ObAccessService *);
-  if (OB_FAIL(ObDMLService::init_dml_param(*lock_ctdef_, *lock_rtdef_, *snapshot_, op_alloc_, dml_param))) {
+  if (OB_FAIL(ObDMLService::init_dml_param(*lock_ctdef_, *lock_rtdef_, *snapshot_, write_branch_id_, op_alloc_, dml_param))) {
     LOG_WARN("init dml param failed", K(ret));
   } else if (OB_FAIL(as->lock_rows(ls_id_,
                                    tablet_id_,

src/sql/das/ob_das_ref.cpp

@@ -515,6 +515,7 @@ int ObDASRef::create_das_task(const ObDASTabletLoc *tablet_loc,
   } else {
     task_op->set_trans_desc(session->get_tx_desc());
     task_op->set_snapshot(&get_exec_ctx().get_das_ctx().get_snapshot());
+    task_op->set_write_branch_id(get_exec_ctx().get_das_ctx().get_write_branch_id());
     task_op->set_tenant_id(session->get_effective_tenant_id());
     task_op->set_task_id(task_id);
     task_op->in_stmt_retry_ = session->get_is_in_retry();

src/sql/das/ob_das_task.h

@@ -111,6 +111,7 @@ public:
       task_flag_(0),
       trans_desc_(nullptr),
       snapshot_(nullptr),
+      write_branch_id_(0),
       tablet_loc_(nullptr),
       op_alloc_(op_alloc),
       related_ctdefs_(op_alloc),
@@ -192,6 +193,8 @@ public:
   transaction::ObTxDesc *get_trans_desc() { return trans_desc_; }
   void set_snapshot(transaction::ObTxReadSnapshot *snapshot) { snapshot_ = snapshot; }
   transaction::ObTxReadSnapshot *get_snapshot() { return snapshot_; }
+  int16_t get_write_branch_id() const { return write_branch_id_; }
+  void set_write_branch_id(const int16_t branch_id) { write_branch_id_ = branch_id; }
   bool is_local_task() const { return task_started_; }
   void set_can_part_retry(const bool flag) { can_part_retry_ = flag; }
   bool can_part_retry() const { return can_part_retry_; }
@@ -245,6 +248,7 @@ protected:
   };
   transaction::ObTxDesc *trans_desc_; //trans desc，事务是全局信息，由RPC框架管理，这里不维护其内存
   transaction::ObTxReadSnapshot *snapshot_; // Mvcc snapshot
+  int16_t write_branch_id_;  // branch id for parallel write, required for partially rollback
   common::ObTabletID tablet_id_;
   share::ObLSID ls_id_;
   const ObDASTabletLoc *tablet_loc_; //does not need serialize it

src/sql/das/ob_das_update_op.cpp

@@ -318,6 +318,7 @@ int ObDASUpdateOp::open_op()
   ObDASIndexDMLAdaptor<DAS_OP_TABLE_UPDATE, ObDASUpdIterator> upd_adaptor;
   upd_adaptor.tx_desc_ = trans_desc_;
   upd_adaptor.snapshot_ = snapshot_;
+  upd_adaptor.write_branch_id_ = write_branch_id_;
   upd_adaptor.ctdef_ = upd_ctdef_;
   upd_adaptor.rtdef_ = upd_rtdef_;
   upd_adaptor.related_ctdefs_ = &related_ctdefs_;