[fix](merge-on-write) cherry pick prs from master (apache#23575)

cherry-pick from: apache#22044 apache#22154 apache#22437

be/src/olap/delta_writer.cpp

@@ -113,7 +113,13 @@ Status DeltaWriter::init() {
  if (_tablet->enable_unique_key_merge_on_write()) {
  std::lock_guard<std::shared_mutex> lck(_tablet->get_header_lock());
  _cur_max_version = _tablet->max_version_unlocked().second;
- _rowset_ids = _tablet->all_rs_id(_cur_max_version);
+ // tablet is under alter process. The delete bitmap will be calculated after conversion.
+ if (_tablet->tablet_state() == TABLET_NOTREADY &&
+ SchemaChangeHandler::tablet_in_converting(_tablet->tablet_id())) {
+ _rowset_ids.clear();
+ } else {
+ _rowset_ids = _tablet->all_rs_id(_cur_max_version);
+ }
  }
 
  // check tablet version number
@@ -399,19 +405,22 @@ Status DeltaWriter::close_wait(const PSlaveTabletNodes& slave_tablet_nodes,
  return res;
  }
  if (_tablet->enable_unique_key_merge_on_write()) {
- auto beta_rowset = reinterpret_cast<BetaRowset*>(_cur_rowset.get());
- std::vector<segment_v2::SegmentSharedPtr> segments;
- RETURN_IF_ERROR(beta_rowset->load_segments(&segments));
  // tablet is under alter process. The delete bitmap will be calculated after conversion.
  if (_tablet->tablet_state() == TABLET_NOTREADY &&
  SchemaChangeHandler::tablet_in_converting(_tablet->tablet_id())) {
- return Status::OK();
- }
- if (segments.size() > 1) {
- RETURN_IF_ERROR(_tablet->calc_delete_bitmap(beta_rowset->rowset_id(), segments, nullptr,
- _delete_bitmap, _cur_max_version, true));
+ LOG(INFO) << "tablet is under alter process, delete bitmap will be calculated later, "
+ "tablet_id: "
+ << _tablet->tablet_id() << " txn_id: " << _req.txn_id;
+ } else {
+ auto beta_rowset = reinterpret_cast<BetaRowset*>(_cur_rowset.get());
+ std::vector<segment_v2::SegmentSharedPtr> segments;
+ RETURN_IF_ERROR(beta_rowset->load_segments(&segments));
+ if (segments.size() > 1) {
+ RETURN_IF_ERROR(_tablet->calc_delete_bitmap(beta_rowset->rowset_id(), segments,
+ nullptr, _delete_bitmap,
+ _cur_max_version, true));
+ }
  }
-
  _storage_engine->txn_manager()->set_txn_related_delete_bitmap(
  _req.partition_id, _req.txn_id, _tablet->tablet_id(), _tablet->schema_hash(),
  _tablet->tablet_uid(), true, _delete_bitmap, _rowset_ids);

be/src/olap/schema_change.cpp

@@ -2067,7 +2067,7 @@ Status SchemaChangeHandler::_do_process_alter_tablet_v2(const TAlterTabletReqV2&
  std::lock_guard<std::mutex> rwlock(new_tablet->get_rowset_update_lock());
  std::lock_guard<std::shared_mutex> new_wlock(new_tablet->get_header_lock());
  SCOPED_SIMPLE_TRACE_IF_TIMEOUT(TRACE_TABLET_LOCK_THRESHOLD);
- int64_t new_max_version = new_tablet->max_version().second;
+ int64_t new_max_version = new_tablet->max_version_unlocked().second;
  rowsets.clear();
  if (max_version < new_max_version) {
  LOG(INFO)
@@ -2377,6 +2377,12 @@ Status SchemaChangeHandler::_parse_request(const SchemaChangeParams& sc_params,
  return Status::OK();
  }
 
+ if (new_tablet->enable_unique_key_merge_on_write() &&
+ new_tablet->num_key_columns() > base_tablet_schema->num_key_columns()) {
+ *sc_directly = true;
+ return Status::OK();
+ }
+
  if (base_tablet_schema->num_short_key_columns() != new_tablet->num_short_key_columns()) {
  // the number of short_keys changed, can't do linked schema change
  *sc_directly = true;

be/src/olap/tablet_meta.cpp

@@ -487,17 +487,26 @@ void TabletMeta::init_from_pb(const TabletMetaPB& tablet_meta_pb) {
  // init _schema
  _schema->init_from_pb(tablet_meta_pb.schema());
 
+ if (tablet_meta_pb.has_enable_unique_key_merge_on_write()) {
+ _enable_unique_key_merge_on_write = tablet_meta_pb.enable_unique_key_merge_on_write();
+ }
+
  // init _rs_metas
  for (auto& it : tablet_meta_pb.rs_metas()) {
  RowsetMetaSharedPtr rs_meta(new RowsetMeta());
  rs_meta->init_from_pb(it);
  _rs_metas.push_back(std::move(rs_meta));
  }
 
- for (auto& it : tablet_meta_pb.stale_rs_metas()) {
- RowsetMetaSharedPtr rs_meta(new RowsetMeta());
- rs_meta->init_from_pb(it);
- _stale_rs_metas.push_back(std::move(rs_meta));
+ // For mow table, delete bitmap of stale rowsets has not been persisted.
+ // When be restart, query should not read the stale rowset, otherwise duplicate keys
+ // will be read out. Therefore, we don't add them to _stale_rs_meta for mow table.
+ if (!_enable_unique_key_merge_on_write) {
+ for (auto& it : tablet_meta_pb.stale_rs_metas()) {
+ RowsetMetaSharedPtr rs_meta(new RowsetMeta());
+ rs_meta->init_from_pb(it);
+ _stale_rs_metas.push_back(std::move(rs_meta));
+ }
  }
 
  if (tablet_meta_pb.has_in_restore_mode()) {
@@ -509,9 +518,6 @@ void TabletMeta::init_from_pb(const TabletMetaPB& tablet_meta_pb) {
  }
 
  _storage_policy = tablet_meta_pb.storage_policy();
- if (tablet_meta_pb.has_enable_unique_key_merge_on_write()) {
- _enable_unique_key_merge_on_write = tablet_meta_pb.enable_unique_key_merge_on_write();
- }
 
  if (tablet_meta_pb.has_delete_bitmap()) {
  int rst_ids_size = tablet_meta_pb.delete_bitmap().rowset_ids_size();