Adding upstream version 18.2.3.upstream/18.2.3upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

48 files changed, 593 insertions, 385 deletions

diff --git a/src/rgw/driver/rados/rgw_bucket.cc b/src/rgw/driver/rados/rgw_bucket.cc
index 32cd1ccf9..d2b4a6b34 100644
--- a/src/rgw/driver/rados/rgw_bucket.cc
+++ b/src/rgw/driver/rados/rgw_bucket.cc
@@ -158,7 +158,7 @@ int rgw_remove_object(const DoutPrefixProvider *dpp, rgw::sal::Driver* driver, r
 
   std::unique_ptr<rgw::sal::Object> object = bucket->get_object(key);
 
-  return object->delete_object(dpp, null_yield);
+  return object->delete_object(dpp, null_yield, rgw::sal::FLAG_LOG_OP);
 }
 
 static void set_err_msg(std::string *sink, std::string msg)
diff --git a/src/rgw/driver/rados/rgw_cr_rados.cc b/src/rgw/driver/rados/rgw_cr_rados.cc
index d8e0ecba6..996a3758f 100644
--- a/src/rgw/driver/rados/rgw_cr_rados.cc
+++ b/src/rgw/driver/rados/rgw_cr_rados.cc
@@ -922,7 +922,7 @@ int RGWAsyncRemoveObj::_send_request(const DoutPrefixProvider *dpp)
   del_op->params.high_precision_time = true;
   del_op->params.zones_trace = &zones_trace;
 
-  ret = del_op->delete_obj(dpp, null_yield);
+  ret = del_op->delete_obj(dpp, null_yield, true);
   if (ret < 0) {
     ldpp_dout(dpp, 20) << __func__ << "(): delete_obj() obj=" << obj << " returned ret=" << ret << dendl;
   }
diff --git a/src/rgw/driver/rados/rgw_d3n_datacache.cc b/src/rgw/driver/rados/rgw_d3n_datacache.cc
index f1bf731ae..b744a16a3 100644
--- a/src/rgw/driver/rados/rgw_d3n_datacache.cc
+++ b/src/rgw/driver/rados/rgw_d3n_datacache.cc
@@ -105,7 +105,7 @@ void D3nDataCache::init(CephContext *_cct) {
   struct aioinit ainit{0};
   ainit.aio_threads = cct->_conf.get_val<int64_t>("rgw_d3n_libaio_aio_threads");
   ainit.aio_num = cct->_conf.get_val<int64_t>("rgw_d3n_libaio_aio_num");
-  ainit.aio_idle_time = 10;
+  ainit.aio_idle_time = 5;
   aio_init(&ainit);
 #endif
 }
diff --git a/src/rgw/driver/rados/rgw_datalog.cc b/src/rgw/driver/rados/rgw_datalog.cc
index 7ca37abf6..14994a615 100644
--- a/src/rgw/driver/rados/rgw_datalog.cc
+++ b/src/rgw/driver/rados/rgw_datalog.cc
@@ -720,7 +720,8 @@ int RGWDataChangesLog::add_entry(const DoutPrefixProvider *dpp,
     ldpp_dout(dpp, 20) << "RGWDataChangesLog::add_entry() sending update with now=" << now << " cur_expiration=" << expiration << dendl;
 
     auto be = bes->head();
-    ret = be->push(dpp, index, now, change.key, std::move(bl), y);
+    // TODO: pass y once we fix the deadlock from https://tracker.ceph.com/issues/63373
+    ret = be->push(dpp, index, now, change.key, std::move(bl), null_yield);
 
     now = real_clock::now();
 
diff --git a/src/rgw/driver/rados/rgw_object_expirer_core.cc b/src/rgw/driver/rados/rgw_object_expirer_core.cc
index ec1bf3fb6..d6beeeb06 100644
--- a/src/rgw/driver/rados/rgw_object_expirer_core.cc
+++ b/src/rgw/driver/rados/rgw_object_expirer_core.cc
@@ -219,7 +219,7 @@ int RGWObjectExpirer::garbage_single_object(const DoutPrefixProvider *dpp, objex
 
   std::unique_ptr<rgw::sal::Object> obj = bucket->get_object(key);
   obj->set_atomic();
-  ret = obj->delete_object(dpp, null_yield);
+  ret = obj->delete_object(dpp, null_yield, rgw::sal::FLAG_LOG_OP);
 
   return ret;
 }
diff --git a/src/rgw/driver/rados/rgw_pubsub_push.cc b/src/rgw/driver/rados/rgw_pubsub_push.cc
index bdb24ce9a..05dc9e65d 100644
--- a/src/rgw/driver/rados/rgw_pubsub_push.cc
+++ b/src/rgw/driver/rados/rgw_pubsub_push.cc
@@ -115,6 +115,55 @@ public:
   }
 };
 
+namespace {
+// this allows waiting untill "finish()" is called from a different thread
+// waiting could be blocking the waiting thread or yielding, depending
+// with compilation flag support and whether the optional_yield is set
+class Waiter {
+  using Signature = void(boost::system::error_code);
+  using Completion = ceph::async::Completion<Signature>;
+  using CompletionInit = boost::asio::async_completion<yield_context, Signature>;
+  std::unique_ptr<Completion> completion = nullptr;
+  int ret;
+
+  bool done = false;
+  mutable std::mutex lock;
+  mutable std::condition_variable cond;
+
+public:
+  int wait(optional_yield y) {
+    std::unique_lock l{lock};
+    if (done) {
+      return ret;
+    }
+    if (y) {
+      boost::system::error_code ec;
+      auto&& token = y.get_yield_context()[ec];
+      CompletionInit init(token);
+      completion = Completion::create(y.get_io_context().get_executor(),
+          std::move(init.completion_handler));
+      l.unlock();
+      init.result.get();
+      return -ec.value();
+    }
+    cond.wait(l, [this]{return (done==true);});
+    return ret;
+  }
+
+  void finish(int r) {
+    std::unique_lock l{lock};
+    ret = r;
+    done = true;
+    if (completion) {
+      boost::system::error_code ec(-ret, boost::system::system_category());
+      Completion::post(std::move(completion), ec);
+    } else {
+      cond.notify_all();
+    }
+  }
+};
+} // namespace
+
 #ifdef WITH_RADOSGW_AMQP_ENDPOINT
 class RGWPubSubAMQPEndpoint : public RGWPubSubEndpoint {
 private:
@@ -187,71 +236,17 @@ public:
     }
   }
 
-  // this allows waiting untill "finish()" is called from a different thread
-  // waiting could be blocking the waiting thread or yielding, depending
-  // with compilation flag support and whether the optional_yield is set
-  class Waiter {
-    using Signature = void(boost::system::error_code);
-    using Completion = ceph::async::Completion<Signature>;
-    std::unique_ptr<Completion> completion = nullptr;
-    int ret;
-
-    mutable std::atomic<bool> done = false;
-    mutable std::mutex lock;
-    mutable std::condition_variable cond;
-
-    template <typename ExecutionContext, typename CompletionToken>
-    auto async_wait(ExecutionContext& ctx, CompletionToken&& token) {
-      boost::asio::async_completion<CompletionToken, Signature> init(token);
-      auto& handler = init.completion_handler;
-      {
-        std::unique_lock l{lock};
-        completion = Completion::create(ctx.get_executor(), std::move(handler));
-      }
-      return init.result.get();
-    }
-
-  public:
-    int wait(optional_yield y) {
-      if (done) {
-        return ret;
-      }
-      if (y) {
-	auto& io_ctx = y.get_io_context();
-        auto& yield_ctx = y.get_yield_context();
-        boost::system::error_code ec;
-        async_wait(io_ctx, yield_ctx[ec]);
-        return -ec.value();
-      }
-      std::unique_lock l(lock);
-      cond.wait(l, [this]{return (done==true);});
-      return ret;
-    }
-
-    void finish(int r) {
-      std::unique_lock l{lock};
-      ret = r;
-      done = true;
-      if (completion) {
-        boost::system::error_code ec(-ret, boost::system::system_category());
-        Completion::post(std::move(completion), ec);
-      } else {
-        cond.notify_all();
-      }
-    }
-  };
-
   int send_to_completion_async(CephContext* cct, const rgw_pubsub_s3_event& event, optional_yield y) override {
     if (ack_level == ack_level_t::None) {
       return amqp::publish(conn_id, topic, json_format_pubsub_event(event));
     } else {
       // TODO: currently broker and routable are the same - this will require different flags but the same mechanism
-      // note: dynamic allocation of Waiter is needed when this is invoked from a beast coroutine
-      auto w = std::unique_ptr<Waiter>(new Waiter);
+      auto w = std::make_unique<Waiter>();
       const auto rc = amqp::publish_with_confirm(conn_id, 
         topic,
         json_format_pubsub_event(event),
-        std::bind(&Waiter::finish, w.get(), std::placeholders::_1));
+        [wp = w.get()](int r) { wp->finish(r);}
+      );
       if (rc < 0) {
         // failed to publish, does not wait for reply
         return rc;
@@ -314,70 +309,16 @@ public:
     }
   }
 
-  // this allows waiting untill "finish()" is called from a different thread
-  // waiting could be blocking the waiting thread or yielding, depending
-  // with compilation flag support and whether the optional_yield is set
-  class Waiter {
-    using Signature = void(boost::system::error_code);
-    using Completion = ceph::async::Completion<Signature>;
-    std::unique_ptr<Completion> completion = nullptr;
-    int ret;
-
-    mutable std::atomic<bool> done = false;
-    mutable std::mutex lock;
-    mutable std::condition_variable cond;
-
-    template <typename ExecutionContext, typename CompletionToken>
-    auto async_wait(ExecutionContext& ctx, CompletionToken&& token) {
-      boost::asio::async_completion<CompletionToken, Signature> init(token);
-      auto& handler = init.completion_handler;
-      {
-        std::unique_lock l{lock};
-        completion = Completion::create(ctx.get_executor(), std::move(handler));
-      }
-      return init.result.get();
-    }
-
-  public:
-    int wait(optional_yield y) {
-      if (done) {
-        return ret;
-      }
-      if (y) {
-        auto& io_ctx = y.get_io_context();
-        auto& yield_ctx = y.get_yield_context();
-        boost::system::error_code ec;
-        async_wait(io_ctx, yield_ctx[ec]);
-        return -ec.value();
-      }
-      std::unique_lock l(lock);
-      cond.wait(l, [this]{return (done==true);});
-      return ret;
-    }
-
-    void finish(int r) {
-      std::unique_lock l{lock};
-      ret = r;
-      done = true;
-      if (completion) {
-        boost::system::error_code ec(-ret, boost::system::system_category());
-        Completion::post(std::move(completion), ec);
-      } else {
-        cond.notify_all();
-      }
-    }
-  };
-
   int send_to_completion_async(CephContext* cct, const rgw_pubsub_s3_event& event, optional_yield y) override {
     if (ack_level == ack_level_t::None) {
       return kafka::publish(conn_name, topic, json_format_pubsub_event(event));
     } else {
-      // note: dynamic allocation of Waiter is needed when this is invoked from a beast coroutine
-      auto w = std::unique_ptr<Waiter>(new Waiter);
+      auto w = std::make_unique<Waiter>();
       const auto rc = kafka::publish_with_confirm(conn_name, 
         topic,
         json_format_pubsub_event(event),
-        std::bind(&Waiter::finish, w.get(), std::placeholders::_1));
+        [wp = w.get()](int r) { wp->finish(r); }
+      );
       if (rc < 0) {
         // failed to publish, does not wait for reply
         return rc;
diff --git a/src/rgw/driver/rados/rgw_putobj_processor.cc b/src/rgw/driver/rados/rgw_putobj_processor.cc
index e453db5a9..dc3b5c9ee 100644
--- a/src/rgw/driver/rados/rgw_putobj_processor.cc
+++ b/src/rgw/driver/rados/rgw_putobj_processor.cc
@@ -124,6 +124,11 @@ void RadosWriter::add_write_hint(librados::ObjectWriteOperation& op) {
   op.set_alloc_hint2(0, 0, alloc_hint_flags);
 }
 
+void RadosWriter::set_head_obj(const rgw_obj& head)
+{
+  head_obj = head;
+}
+
 int RadosWriter::set_stripe_obj(const rgw_raw_obj& raw_obj)
 {
   stripe_obj = store->svc.rados->obj(raw_obj);
@@ -339,7 +344,8 @@ int AtomicObjectProcessor::complete(size_t accounted_size,
                                     const char *if_nomatch,
                                     const std::string *user_data,
                                     rgw_zone_set *zones_trace,
-                                    bool *pcanceled, optional_yield y)
+                                    bool *pcanceled, optional_yield y,
+                                    uint32_t flags)
 {
   int r = writer.drain();
   if (r < 0) {
@@ -376,7 +382,8 @@ int AtomicObjectProcessor::complete(size_t accounted_size,
 
   read_cloudtier_info_from_attrs(attrs, obj_op.meta.category, manifest);
 
-  r = obj_op.write_meta(dpp, actual_size, accounted_size, attrs, y);
+  r = obj_op.write_meta(dpp, actual_size, accounted_size, attrs, y,
+                        flags & rgw::sal::FLAG_LOG_OP);
   if (r < 0) {
     if (r == -ETIMEDOUT) {
       // The head object write may eventually succeed, clear the set of objects for deletion. if it
@@ -451,6 +458,9 @@ int MultipartObjectProcessor::prepare_head()
   RGWSI_Tier_RADOS::raw_obj_to_obj(head_obj.bucket, stripe_obj, &head_obj);
   head_obj.index_hash_source = target_obj.key.name;
 
+  // point part uploads at the part head instead of the final multipart head
+  writer.set_head_obj(head_obj);
+
   r = writer.set_stripe_obj(stripe_obj);
   if (r < 0) {
     return r;
@@ -480,7 +490,8 @@ int MultipartObjectProcessor::complete(size_t accounted_size,
                                        const char *if_nomatch,
                                        const std::string *user_data,
                                        rgw_zone_set *zones_trace,
-                                       bool *pcanceled, optional_yield y)
+                                       bool *pcanceled, optional_yield y,
+                                       uint32_t flags)
 {
   int r = writer.drain();
   if (r < 0) {
@@ -504,7 +515,8 @@ int MultipartObjectProcessor::complete(size_t accounted_size,
   obj_op.meta.zones_trace = zones_trace;
   obj_op.meta.modify_tail = true;
 
-  r = obj_op.write_meta(dpp, actual_size, accounted_size, attrs, y);
+  r = obj_op.write_meta(dpp, actual_size, accounted_size, attrs, y,
+                        flags & rgw::sal::FLAG_LOG_OP);
   if (r < 0)
     return r;
 
@@ -684,7 +696,7 @@ int AppendObjectProcessor::complete(size_t accounted_size, const string &etag, c
                                     ceph::real_time set_mtime, rgw::sal::Attrs& attrs,
                                     ceph::real_time delete_at, const char *if_match, const char *if_nomatch,
                                     const string *user_data, rgw_zone_set *zones_trace, bool *pcanceled,
-                                    optional_yield y)
+                                    optional_yield y, uint32_t flags)
 {
   int r = writer.drain();
   if (r < 0)
@@ -742,7 +754,7 @@ int AppendObjectProcessor::complete(size_t accounted_size, const string &etag, c
   }
   r = obj_op.write_meta(dpp, actual_size + cur_size,
 			accounted_size + *cur_accounted_size,
-			attrs, y);
+			attrs, y, flags & rgw::sal::FLAG_LOG_OP);
   if (r < 0) {
     return r;
   }
diff --git a/src/rgw/driver/rados/rgw_putobj_processor.h b/src/rgw/driver/rados/rgw_putobj_processor.h
index fa9200f32..b1946c049 100644
--- a/src/rgw/driver/rados/rgw_putobj_processor.h
+++ b/src/rgw/driver/rados/rgw_putobj_processor.h
@@ -69,7 +69,7 @@ class RadosWriter : public rgw::sal::DataProcessor {
   RGWRados *const store;
   const RGWBucketInfo& bucket_info;
   RGWObjectCtx& obj_ctx;
-  const rgw_obj head_obj;
+  rgw_obj head_obj;
   RGWSI_RADOS::Obj stripe_obj; // current stripe object
   RawObjSet written; // set of written objects for deletion
   const DoutPrefixProvider *dpp;
@@ -88,6 +88,9 @@ class RadosWriter : public rgw::sal::DataProcessor {
   // add alloc hint to osd
   void add_write_hint(librados::ObjectWriteOperation& op);
 
+  // change the head object
+  void set_head_obj(const rgw_obj& head);
+
   // change the current stripe object
   int set_stripe_obj(const rgw_raw_obj& obj);
 
@@ -191,7 +194,8 @@ class AtomicObjectProcessor : public ManifestObjectProcessor {
                const char *if_match, const char *if_nomatch,
                const std::string *user_data,
                rgw_zone_set *zones_trace, bool *canceled,
-               optional_yield y) override;
+               optional_yield y,
+               uint32_t flags) override;
 
 };
 
@@ -238,7 +242,8 @@ class MultipartObjectProcessor : public ManifestObjectProcessor {
                const char *if_match, const char *if_nomatch,
                const std::string *user_data,
                rgw_zone_set *zones_trace, bool *canceled,
-               optional_yield y) override;
+               optional_yield y,
+               uint32_t flags) override;
 
 };
 
@@ -274,7 +279,8 @@ class MultipartObjectProcessor : public ManifestObjectProcessor {
                  std::map<std::string, bufferlist>& attrs, ceph::real_time delete_at,
                  const char *if_match, const char *if_nomatch, const std::string *user_data,
                  rgw_zone_set *zones_trace, bool *canceled,
-                 optional_yield y) override;
+                 optional_yield y,
+                 uint32_t flags) override;
   };
 
 } // namespace putobj
diff --git a/src/rgw/driver/rados/rgw_rados.cc b/src/rgw/driver/rados/rgw_rados.cc
index 10018d4a6..566b7d6c2 100644
--- a/src/rgw/driver/rados/rgw_rados.cc
+++ b/src/rgw/driver/rados/rgw_rados.cc
@@ -900,9 +900,11 @@ void RGWIndexCompletionManager::process()
         continue;
       }
 
-      // This null_yield can stay, for now, since we're in our own thread
-      add_datalog_entry(&dpp, store->svc.datalog_rados, bucket_info,
-			bs.shard_id, null_yield);
+      if (c->log_op) {
+        // This null_yield can stay, for now, since we're in our own thread
+        add_datalog_entry(&dpp, store->svc.datalog_rados, bucket_info,
+                          bs.shard_id, null_yield);
+      }
     }
   }
 }
@@ -3014,7 +3016,8 @@ int RGWRados::Object::Write::_do_write_meta(const DoutPrefixProvider *dpp,
                                            uint64_t size, uint64_t accounted_size,
                                            map<string, bufferlist>& attrs,
                                            bool assume_noent, bool modify_tail,
-                                           void *_index_op, optional_yield y)
+                                           void *_index_op, optional_yield y,
+                                           bool log_op)
 {
   RGWRados::Bucket::UpdateIndex *index_op = static_cast<RGWRados::Bucket::UpdateIndex *>(_index_op);
   RGWRados *store = target->get_store();
@@ -3180,7 +3183,7 @@ int RGWRados::Object::Write::_do_write_meta(const DoutPrefixProvider *dpp,
 
   if (!index_op->is_prepared()) {
     tracepoint(rgw_rados, prepare_enter, req_id.c_str());
-    r = index_op->prepare(dpp, CLS_RGW_OP_ADD, &state->write_tag, y);
+    r = index_op->prepare(dpp, CLS_RGW_OP_ADD, &state->write_tag, y, log_op);
     tracepoint(rgw_rados, prepare_exit, req_id.c_str());
     if (r < 0)
       return r;
@@ -3214,7 +3217,7 @@ int RGWRados::Object::Write::_do_write_meta(const DoutPrefixProvider *dpp,
                         meta.set_mtime, etag, content_type,
                         storage_class, &acl_bl,
 			 meta.category, meta.remove_objs, y,
-			 meta.user_data, meta.appendable);
+			 meta.user_data, meta.appendable, log_op);
   tracepoint(rgw_rados, complete_exit, req_id.c_str());
   if (r < 0)
     goto done_cancel;
@@ -3228,7 +3231,7 @@ int RGWRados::Object::Write::_do_write_meta(const DoutPrefixProvider *dpp,
   state = NULL;
 
   if (versioned_op && meta.olh_epoch) {
-    r = store->set_olh(dpp, target->get_ctx(), target->get_bucket_info(), obj, false, NULL, *meta.olh_epoch, real_time(), false, y, meta.zones_trace);
+    r = store->set_olh(dpp, target->get_ctx(), target->get_bucket_info(), obj, false, NULL, *meta.olh_epoch, real_time(), false, y, meta.zones_trace, log_op);
     if (r < 0) {
       return r;
     }
@@ -3259,7 +3262,7 @@ int RGWRados::Object::Write::_do_write_meta(const DoutPrefixProvider *dpp,
   return 0;
 
 done_cancel:
-  int ret = index_op->cancel(dpp, meta.remove_objs, y);
+  int ret = index_op->cancel(dpp, meta.remove_objs, y, log_op);
   if (ret < 0) {
     ldpp_dout(dpp, 0) << "ERROR: index_op.cancel() returned ret=" << ret << dendl;
   }
@@ -3304,7 +3307,7 @@ done_cancel:
 }
 
 int RGWRados::Object::Write::write_meta(const DoutPrefixProvider *dpp, uint64_t size, uint64_t accounted_size,
-                                           map<string, bufferlist>& attrs, optional_yield y)
+                                           map<string, bufferlist>& attrs, optional_yield y, bool log_op)
 {
   RGWBucketInfo& bucket_info = target->get_bucket_info();
 
@@ -3315,13 +3318,13 @@ int RGWRados::Object::Write::write_meta(const DoutPrefixProvider *dpp, uint64_t
   bool assume_noent = (meta.if_match == NULL && meta.if_nomatch == NULL);
   int r;
   if (assume_noent) {
-    r = _do_write_meta(dpp, size, accounted_size, attrs, assume_noent, meta.modify_tail, (void *)&index_op, y);
+    r = _do_write_meta(dpp, size, accounted_size, attrs, assume_noent, meta.modify_tail, (void *)&index_op, y, log_op);
     if (r == -EEXIST) {
       assume_noent = false;
     }
   }
   if (!assume_noent) {
-    r = _do_write_meta(dpp, size, accounted_size, attrs, assume_noent, meta.modify_tail, (void *)&index_op, y);
+    r = _do_write_meta(dpp, size, accounted_size, attrs, assume_noent, meta.modify_tail, (void *)&index_op, y, log_op);
   }
   return r;
 }
@@ -4198,7 +4201,7 @@ int RGWRados::fetch_remote_obj(RGWObjectCtx& obj_ctx,
     bool canceled = false;
     ret = processor.complete(accounted_size, etag, mtime, set_mtime,
                              attrs, delete_at, nullptr, nullptr, nullptr,
-                             zones_trace, &canceled, null_yield);
+                             zones_trace, &canceled, null_yield, rgw::sal::FLAG_LOG_OP);
     if (ret < 0) {
       goto set_err_state;
     }
@@ -4409,6 +4412,12 @@ int RGWRados::copy_obj(RGWObjectCtx& obj_ctx,
   if (lh != attrs.end())
     src_attrs[RGW_ATTR_OBJECT_LEGAL_HOLD] = lh->second;
 
+  if (dest_bucket_info.flags & BUCKET_VERSIONS_SUSPENDED) {
+    src_attrs.erase(RGW_ATTR_OLH_ID_TAG);
+    src_attrs.erase(RGW_ATTR_OLH_INFO);
+    src_attrs.erase(RGW_ATTR_OLH_VER);
+  }
+
   set_copy_attrs(src_attrs, attrs, attrs_mod);
   attrs.erase(RGW_ATTR_ID_TAG);
   attrs.erase(RGW_ATTR_PG_VER);
@@ -4652,7 +4661,8 @@ int RGWRados::copy_obj_data(RGWObjectCtx& obj_ctx,
 	       real_time delete_at,
                string *petag,
                const DoutPrefixProvider *dpp,
-               optional_yield y)
+               optional_yield y,
+               bool log_op)
 {
   string tag;
   append_rand_alpha(cct, tag, tag, 32);
@@ -4718,7 +4728,8 @@ int RGWRados::copy_obj_data(RGWObjectCtx& obj_ctx,
   }
 
   return processor.complete(accounted_size, etag, mtime, set_mtime, attrs, delete_at,
-                            nullptr, nullptr, nullptr, nullptr, nullptr, y);
+                            nullptr, nullptr, nullptr, nullptr, nullptr, y,
+                            log_op ? rgw::sal::FLAG_LOG_OP : 0);
 }
 
 int RGWRados::transition_obj(RGWObjectCtx& obj_ctx,
@@ -4728,7 +4739,8 @@ int RGWRados::transition_obj(RGWObjectCtx& obj_ctx,
                              const real_time& mtime,
                              uint64_t olh_epoch,
                              const DoutPrefixProvider *dpp,
-                             optional_yield y)
+                             optional_yield y,
+                             bool log_op)
 {
   rgw::sal::Attrs attrs;
   real_time read_mtime;
@@ -4769,7 +4781,8 @@ int RGWRados::transition_obj(RGWObjectCtx& obj_ctx,
                       real_time(),
                       nullptr /* petag */,
                       dpp,
-                      y);
+                      y,
+                      log_op);
   if (ret < 0) {
     return ret;
   }
@@ -5340,7 +5353,7 @@ struct tombstone_entry {
  * obj: name of the object to delete
  * Returns: 0 on success, -ERR# otherwise.
  */
-int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvider *dpp)
+int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvider *dpp, bool log_op)
 {
   RGWRados *store = target->get_store();
   const rgw_obj& src_obj = target->get_obj();
@@ -5354,6 +5367,8 @@ int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvi
   bool explicit_marker_version = (!params.marker_version_id.empty());
 
   if (params.versioning_status & BUCKET_VERSIONED || explicit_marker_version) {
+    bool add_log = log_op && store->svc.zone->need_to_log_data();
+
     if (instance.empty() || explicit_marker_version) {
       rgw_obj marker = obj;
       marker.key.instance.clear();
@@ -5382,7 +5397,9 @@ int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvi
         meta.mtime = params.mtime;
       }
 
-      int r = store->set_olh(dpp, target->get_ctx(), target->get_bucket_info(), marker, true, &meta, params.olh_epoch, params.unmod_since, params.high_precision_time, y, params.zones_trace);
+      int r = store->set_olh(dpp, target->get_ctx(), target->get_bucket_info(), marker, true,
+                             &meta, params.olh_epoch, params.unmod_since, params.high_precision_time,
+                             y, params.zones_trace, add_log);
       if (r < 0) {
         return r;
       }
@@ -5394,7 +5411,8 @@ int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvi
         return r;
       }
       result.delete_marker = dirent.is_delete_marker();
-      r = store->unlink_obj_instance(dpp, target->get_ctx(), target->get_bucket_info(), obj, params.olh_epoch, y, params.zones_trace);
+      r = store->unlink_obj_instance(dpp, target->get_ctx(), target->get_bucket_info(), obj, params.olh_epoch,
+                                     y, params.zones_trace, add_log);
       if (r < 0) {
         return r;
       }
@@ -5408,8 +5426,10 @@ int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvi
       return r;
     }
 
-    add_datalog_entry(dpp, store->svc.datalog_rados,
-                      target->get_bucket_info(), bs->shard_id, y);
+    if (add_log) {
+      add_datalog_entry(dpp, store->svc.datalog_rados,
+                        target->get_bucket_info(), bs->shard_id, y);
+    }
 
     return 0;
   }
@@ -5488,7 +5508,7 @@ int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvi
   index_op.set_zones_trace(params.zones_trace);
   index_op.set_bilog_flags(params.bilog_flags);
 
-  r = index_op.prepare(dpp, CLS_RGW_OP_DEL, &state->write_tag, y);
+  r = index_op.prepare(dpp, CLS_RGW_OP_DEL, &state->write_tag, y, log_op);
   if (r < 0)
     return r;
 
@@ -5507,7 +5527,7 @@ int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvi
       tombstone_entry entry{*state};
       obj_tombstone_cache->add(obj, entry);
     }
-    r = index_op.complete_del(dpp, poolid, ioctx.get_last_version(), state->mtime, params.remove_objs, y);
+    r = index_op.complete_del(dpp, poolid, ioctx.get_last_version(), state->mtime, params.remove_objs, y, log_op);
 
     int ret = target->complete_atomic_modification(dpp);
     if (ret < 0) {
@@ -5515,7 +5535,7 @@ int RGWRados::Object::Delete::delete_obj(optional_yield y, const DoutPrefixProvi
     }
     /* other than that, no need to propagate error */
   } else {
-    int ret = index_op.cancel(dpp, params.remove_objs, y);
+    int ret = index_op.cancel(dpp, params.remove_objs, y, log_op);
     if (ret < 0) {
       ldpp_dout(dpp, 0) << "ERROR: index_op.cancel() returned ret=" << ret << dendl;
     }
@@ -5541,7 +5561,8 @@ int RGWRados::delete_obj(const DoutPrefixProvider *dpp,
                          int versioning_status, // versioning flags defined in enum RGWBucketFlags
                          uint16_t bilog_flags,
                          const real_time& expiration_time,
-                         rgw_zone_set *zones_trace)
+                         rgw_zone_set *zones_trace,
+                         bool log_op)
 {
   RGWRados::Object del_target(this, bucket_info, obj_ctx, obj);
   RGWRados::Object::Delete del_op(&del_target);
@@ -5552,7 +5573,7 @@ int RGWRados::delete_obj(const DoutPrefixProvider *dpp,
   del_op.params.expiration_time = expiration_time;
   del_op.params.zones_trace = zones_trace;
 
-  return del_op.delete_obj(null_yield, dpp);
+  return del_op.delete_obj(null_yield, dpp, log_op ? rgw::sal::FLAG_LOG_OP : 0);
 }
 
 int RGWRados::delete_raw_obj(const DoutPrefixProvider *dpp, const rgw_raw_obj& obj)
@@ -6308,6 +6329,10 @@ int RGWRados::Object::Read::prepare(optional_yield y, const DoutPrefixProvider *
     }
   }
 
+  if (params.lastmod) {
+    *params.lastmod = astate->mtime;
+  }
+
   /* Convert all times go GMT to make them compatible */
   if (conds.mod_ptr || conds.unmod_ptr) {
     obj_time_weight src_weight;
@@ -6357,8 +6382,6 @@ int RGWRados::Object::Read::prepare(optional_yield y, const DoutPrefixProvider *
 
   if (params.obj_size)
     *params.obj_size = astate->size;
-  if (params.lastmod)
-    *params.lastmod = astate->mtime;
 
   return 0;
 }
@@ -6440,7 +6463,8 @@ int RGWRados::Bucket::UpdateIndex::guard_reshard(const DoutPrefixProvider *dpp,
   return 0;
 }
 
-int RGWRados::Bucket::UpdateIndex::prepare(const DoutPrefixProvider *dpp, RGWModifyOp op, const string *write_tag, optional_yield y)
+int RGWRados::Bucket::UpdateIndex::prepare(const DoutPrefixProvider *dpp, RGWModifyOp op, const string *write_tag,
+                                           optional_yield y, bool log_op)
 {
   if (blind) {
     return 0;
@@ -6455,8 +6479,10 @@ int RGWRados::Bucket::UpdateIndex::prepare(const DoutPrefixProvider *dpp, RGWMod
     }
   }
 
+  bool add_log = log_op && store->svc.zone->need_to_log_data();
+
   int r = guard_reshard(dpp, obj, nullptr, [&](BucketShard *bs) -> int {
-				   return store->cls_obj_prepare_op(dpp, *bs, op, optag, obj, bilog_flags, y, zones_trace);
+				   return store->cls_obj_prepare_op(dpp, *bs, op, optag, obj, bilog_flags, y, zones_trace, add_log);
 				 });
 
   if (r < 0) {
@@ -6476,7 +6502,8 @@ int RGWRados::Bucket::UpdateIndex::complete(const DoutPrefixProvider *dpp, int64
                                             list<rgw_obj_index_key> *remove_objs,
 					    optional_yield y,
 					    const string *user_data,
-                                            bool appendable)
+                                            bool appendable,
+                                            bool log_op)
 {
   if (blind) {
     return 0;
@@ -6512,10 +6539,13 @@ int RGWRados::Bucket::UpdateIndex::complete(const DoutPrefixProvider *dpp, int64
   ent.meta.content_type = content_type;
   ent.meta.appendable = appendable;
 
-  ret = store->cls_obj_complete_add(*bs, obj, optag, poolid, epoch, ent, category, remove_objs, bilog_flags, zones_trace);
+  bool add_log = log_op && store->svc.zone->need_to_log_data();
 
-  add_datalog_entry(dpp, store->svc.datalog_rados,
-                    target->bucket_info, bs->shard_id, y);
+  ret = store->cls_obj_complete_add(*bs, obj, optag, poolid, epoch, ent, category, remove_objs, bilog_flags, zones_trace, add_log);
+  if (add_log) {
+    add_datalog_entry(dpp, store->svc.datalog_rados,
+                      target->bucket_info, bs->shard_id, y);
+  }
 
   return ret;
 }
@@ -6524,7 +6554,8 @@ int RGWRados::Bucket::UpdateIndex::complete_del(const DoutPrefixProvider *dpp,
                                                 int64_t poolid, uint64_t epoch,
                                                 real_time& removed_mtime,
                                                 list<rgw_obj_index_key> *remove_objs,
-						optional_yield y)
+                                                optional_yield y,
+                                                bool log_op)
 {
   if (blind) {
     return 0;
@@ -6538,10 +6569,14 @@ int RGWRados::Bucket::UpdateIndex::complete_del(const DoutPrefixProvider *dpp,
     return ret;
   }
 
-  ret = store->cls_obj_complete_del(*bs, optag, poolid, epoch, obj, removed_mtime, remove_objs, bilog_flags, zones_trace);
+  bool add_log = log_op && store->svc.zone->need_to_log_data();
 
-  add_datalog_entry(dpp, store->svc.datalog_rados,
-                    target->bucket_info, bs->shard_id, y);
+  ret = store->cls_obj_complete_del(*bs, optag, poolid, epoch, obj, removed_mtime, remove_objs, bilog_flags, zones_trace, add_log);
+
+  if (add_log) {
+    add_datalog_entry(dpp, store->svc.datalog_rados,
+                      target->bucket_info, bs->shard_id, y);
+  }
 
   return ret;
 }
@@ -6549,7 +6584,8 @@ int RGWRados::Bucket::UpdateIndex::complete_del(const DoutPrefixProvider *dpp,
 
 int RGWRados::Bucket::UpdateIndex::cancel(const DoutPrefixProvider *dpp,
                                           list<rgw_obj_index_key> *remove_objs,
-					  optional_yield y)
+                                          optional_yield y,
+                                          bool log_op)
 {
     if (blind) {
     return 0;
@@ -6557,17 +6593,21 @@ int RGWRados::Bucket::UpdateIndex::cancel(const DoutPrefixProvider *dpp,
   RGWRados *store = target->get_store();
   BucketShard *bs;
 
+  bool add_log = log_op && store->svc.zone->need_to_log_data();
+
   int ret = guard_reshard(dpp, obj, &bs, [&](BucketShard *bs) -> int {
-				 return store->cls_obj_complete_cancel(*bs, optag, obj, remove_objs, bilog_flags, zones_trace);
+				 return store->cls_obj_complete_cancel(*bs, optag, obj, remove_objs, bilog_flags, zones_trace, add_log);
 			       });
 
-  /*
-   * need to update data log anyhow, so that whoever follows needs to update its internal markers
-   * for following the specific bucket shard log. Otherwise they end up staying behind, and users
-   * have no way to tell that they're all caught up
-   */
-  add_datalog_entry(dpp, store->svc.datalog_rados,
-                    target->bucket_info, bs->shard_id, y);
+  if (add_log) {
+    /*
+     * need to update data log anyhow, so that whoever follows needs to update its internal markers
+     * for following the specific bucket shard log. Otherwise they end up staying behind, and users
+     * have no way to tell that they're all caught up
+     */
+    add_datalog_entry(dpp, store->svc.datalog_rados,
+                      target->bucket_info, bs->shard_id, y);
+  }
 
   return ret;
 }
@@ -7304,7 +7344,7 @@ int RGWRados::bucket_index_link_olh(const DoutPrefixProvider *dpp, RGWBucketInfo
 		      cls_rgw_bucket_link_olh(op, key, olh_state.olh_tag,
                                               delete_marker, op_tag, meta, olh_epoch,
 					      unmod_since, high_precision_time,
-					      svc.zone->need_to_log_data(), zones_trace);
+					      log_data_change, zones_trace);
                       return rgw_rados_operate(dpp, ref.pool.ioctx(), ref.obj.oid, &op, null_yield);
                     });
   if (r < 0) {
@@ -7329,7 +7369,7 @@ int RGWRados::bucket_index_unlink_instance(const DoutPrefixProvider *dpp,
                                            RGWBucketInfo& bucket_info,
                                            const rgw_obj& obj_instance,
                                            const string& op_tag, const string& olh_tag,
-                                           uint64_t olh_epoch, rgw_zone_set *_zones_trace)
+                                           uint64_t olh_epoch, rgw_zone_set *_zones_trace, bool log_op)
 {
   rgw_rados_ref ref;
   int r = get_obj_head_ref(dpp, bucket_info, obj_instance, &ref);
@@ -7353,7 +7393,7 @@ int RGWRados::bucket_index_unlink_instance(const DoutPrefixProvider *dpp,
 		      op.assert_exists(); // bucket index shard must exist
 		      cls_rgw_guard_bucket_resharding(op, -ERR_BUSY_RESHARDING);
 		      cls_rgw_bucket_unlink_instance(op, key, op_tag,
-						     olh_tag, olh_epoch, svc.zone->need_to_log_data(), zones_trace);
+						     olh_tag, olh_epoch, log_op, zones_trace);
                       return rgw_rados_operate(dpp, ref.pool.ioctx(), ref.obj.oid, &op, null_yield);
                     });
   if (r < 0) {
@@ -7556,7 +7596,8 @@ int RGWRados::apply_olh_log(const DoutPrefixProvider *dpp,
 			    bufferlist& olh_tag,
 			    std::map<uint64_t, std::vector<rgw_bucket_olh_log_entry> >& log,
 			    uint64_t *plast_ver,
-			    rgw_zone_set* zones_trace)
+			    rgw_zone_set* zones_trace,
+                            bool log_op)
 {
   if (log.empty()) {
     return 0;
@@ -7669,7 +7710,7 @@ int RGWRados::apply_olh_log(const DoutPrefixProvider *dpp,
        liter != remove_instances.end(); ++liter) {
     cls_rgw_obj_key& key = *liter;
     rgw_obj obj_instance(bucket, key);
-    int ret = delete_obj(dpp, obj_ctx, bucket_info, obj_instance, 0, RGW_BILOG_FLAG_VERSIONED_OP, ceph::real_time(), zones_trace);
+    int ret = delete_obj(dpp, obj_ctx, bucket_info, obj_instance, 0, RGW_BILOG_FLAG_VERSIONED_OP, ceph::real_time(), zones_trace, log_op);
     if (ret < 0 && ret != -ENOENT) {
       ldpp_dout(dpp, 0) << "ERROR: delete_obj() returned " << ret << " obj_instance=" << obj_instance << dendl;
       return ret;
@@ -7773,7 +7814,7 @@ int RGWRados::clear_olh(const DoutPrefixProvider *dpp,
 /*
  * read olh log and apply it
  */
-int RGWRados::update_olh(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx, RGWObjState *state, RGWBucketInfo& bucket_info, const rgw_obj& obj, rgw_zone_set *zones_trace)
+int RGWRados::update_olh(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx, RGWObjState *state, RGWBucketInfo& bucket_info, const rgw_obj& obj, rgw_zone_set *zones_trace, bool log_op)
 {
   map<uint64_t, vector<rgw_bucket_olh_log_entry> > log;
   bool is_truncated;
@@ -7784,7 +7825,7 @@ int RGWRados::update_olh(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx, R
     if (ret < 0) {
       return ret;
     }
-    ret = apply_olh_log(dpp, obj_ctx, *state, bucket_info, obj, state->olh_tag, log, &ver_marker, zones_trace);
+    ret = apply_olh_log(dpp, obj_ctx, *state, bucket_info, obj, state->olh_tag, log, &ver_marker, zones_trace, log_op);
     if (ret < 0) {
       return ret;
     }
@@ -7853,7 +7894,7 @@ int RGWRados::set_olh(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx,
       // it's possible that the pending xattr from this op prevented the olh
       // object from being cleaned by another thread that was deleting the last
       // existing version. We invoke a best-effort update_olh here to handle this case.
-      int r = update_olh(dpp, obj_ctx, state, bucket_info, olh_obj);
+      int r = update_olh(dpp, obj_ctx, state, bucket_info, olh_obj, zones_trace, log_data_change);
       if (r < 0 && r != -ECANCELED) {
         ldpp_dout(dpp, 20) << "update_olh() target_obj=" << olh_obj << " returned " << r << dendl;
       }
@@ -7867,7 +7908,7 @@ int RGWRados::set_olh(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx,
     return -EIO;
   }
 
-  ret = update_olh(dpp, obj_ctx, state, bucket_info, olh_obj);
+  ret = update_olh(dpp, obj_ctx, state, bucket_info, olh_obj, zones_trace, log_data_change);
   if (ret == -ECANCELED) { /* already did what we needed, no need to retry, raced with another user */
     ret = 0;
   }
@@ -7880,7 +7921,7 @@ int RGWRados::set_olh(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx,
 }
 
 int RGWRados::unlink_obj_instance(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx, RGWBucketInfo& bucket_info, const rgw_obj& target_obj,
-                                  uint64_t olh_epoch, optional_yield y, rgw_zone_set *zones_trace)
+                                  uint64_t olh_epoch, optional_yield y, rgw_zone_set *zones_trace, bool log_op)
 {
   string op_tag;
 
@@ -7913,7 +7954,7 @@ int RGWRados::unlink_obj_instance(const DoutPrefixProvider *dpp, RGWObjectCtx& o
 
     string olh_tag(state->olh_tag.c_str(), state->olh_tag.length());
 
-    ret = bucket_index_unlink_instance(dpp, bucket_info, target_obj, op_tag, olh_tag, olh_epoch, zones_trace);
+    ret = bucket_index_unlink_instance(dpp, bucket_info, target_obj, op_tag, olh_tag, olh_epoch, zones_trace, log_op);
     if (ret < 0) {
       olh_cancel_modification(dpp, bucket_info, *state, olh_obj, op_tag, y);
       ldpp_dout(dpp, 20) << "bucket_index_unlink_instance() target_obj=" << target_obj << " returned " << ret << dendl;
@@ -7923,7 +7964,7 @@ int RGWRados::unlink_obj_instance(const DoutPrefixProvider *dpp, RGWObjectCtx& o
       // it's possible that the pending xattr from this op prevented the olh
       // object from being cleaned by another thread that was deleting the last
       // existing version. We invoke a best-effort update_olh here to handle this case.
-      int r = update_olh(dpp, obj_ctx, state, bucket_info, olh_obj, zones_trace);
+      int r = update_olh(dpp, obj_ctx, state, bucket_info, olh_obj, zones_trace, log_op);
       if (r < 0 && r != -ECANCELED) {
         ldpp_dout(dpp, 20) << "update_olh() target_obj=" << olh_obj << " returned " << r << dendl;
       }
@@ -7937,7 +7978,7 @@ int RGWRados::unlink_obj_instance(const DoutPrefixProvider *dpp, RGWObjectCtx& o
     return -EIO;
   }
 
-  ret = update_olh(dpp, obj_ctx, state, bucket_info, olh_obj, zones_trace);
+  ret = update_olh(dpp, obj_ctx, state, bucket_info, olh_obj, zones_trace, log_op);
   if (ret == -ECANCELED) { /* already did what we needed, no need to retry, raced with another user */
     return 0;
   }
@@ -8781,7 +8822,8 @@ bool RGWRados::process_expire_objects(const DoutPrefixProvider *dpp)
 }
 
 int RGWRados::cls_obj_prepare_op(const DoutPrefixProvider *dpp, BucketShard& bs, RGWModifyOp op, string& tag,
-                                 rgw_obj& obj, uint16_t bilog_flags, optional_yield y, rgw_zone_set *_zones_trace)
+                                 rgw_obj& obj, uint16_t bilog_flags, optional_yield y, rgw_zone_set *_zones_trace,
+                                 bool log_op)
 {
   const bool bitx = cct->_conf->rgw_bucket_index_transaction_instrumentation;
   ldout_bitx(bitx, dpp, 10) << "ENTERING " << __func__ << ": bucket-shard=" << bs << " obj=" << obj << " tag=" << tag << " op=" << op << dendl_bitx;
@@ -8798,7 +8840,7 @@ int RGWRados::cls_obj_prepare_op(const DoutPrefixProvider *dpp, BucketShard& bs,
 
   cls_rgw_obj_key key(obj.key.get_index_key_name(), obj.key.instance);
   cls_rgw_guard_bucket_resharding(o, -ERR_BUSY_RESHARDING);
-  cls_rgw_bucket_prepare_op(o, op, tag, key, obj.key.get_loc(), svc.zone->need_to_log_data(), bilog_flags, zones_trace);
+  cls_rgw_bucket_prepare_op(o, op, tag, key, obj.key.get_loc(), log_op, bilog_flags, zones_trace);
   int ret = bs.bucket_obj.operate(dpp, &o, y);
   ldout_bitx(bitx, dpp, 10) << "EXITING " << __func__ << ": ret=" << ret << dendl_bitx;
   return ret;
@@ -8807,12 +8849,14 @@ int RGWRados::cls_obj_prepare_op(const DoutPrefixProvider *dpp, BucketShard& bs,
 int RGWRados::cls_obj_complete_op(BucketShard& bs, const rgw_obj& obj, RGWModifyOp op, string& tag,
                                   int64_t pool, uint64_t epoch,
                                   rgw_bucket_dir_entry& ent, RGWObjCategory category,
-				  list<rgw_obj_index_key> *remove_objs, uint16_t bilog_flags, rgw_zone_set *_zones_trace)
+                                  list<rgw_obj_index_key> *remove_objs, uint16_t bilog_flags,
+                                  rgw_zone_set *_zones_trace, bool log_op)
 {
   const bool bitx = cct->_conf->rgw_bucket_index_transaction_instrumentation;
   ldout_bitx_c(bitx, cct, 10) << "ENTERING " << __func__ << ": bucket-shard=" << bs <<
     " obj=" << obj << " tag=" << tag << " op=" << op <<
-    ", remove_objs=" << (remove_objs ? *remove_objs : std::list<rgw_obj_index_key>()) << dendl_bitx;
+    ", remove_objs=" << (remove_objs ? *remove_objs : std::list<rgw_obj_index_key>()) <<
+    ", log_op=" << log_op << dendl_bitx;
   ldout_bitx_c(bitx, cct, 25) << "BACKTRACE: " << __func__ << ": " << ClibBackTrace(0) << dendl_bitx;
 
   ObjectWriteOperation o;
@@ -8834,10 +8878,10 @@ int RGWRados::cls_obj_complete_op(BucketShard& bs, const rgw_obj& obj, RGWModify
   cls_rgw_obj_key key(ent.key.name, ent.key.instance);
   cls_rgw_guard_bucket_resharding(o, -ERR_BUSY_RESHARDING);
   cls_rgw_bucket_complete_op(o, op, tag, ver, key, dir_meta, remove_objs,
-                             svc.zone->need_to_log_data(), bilog_flags, &zones_trace);
+                             log_op, bilog_flags, &zones_trace);
   complete_op_data *arg;
   index_completion_manager->create_completion(obj, op, tag, ver, key, dir_meta, remove_objs,
-                                              svc.zone->need_to_log_data(), bilog_flags, &zones_trace, &arg);
+                                              log_op, bilog_flags, &zones_trace, &arg);
   librados::AioCompletion *completion = arg->rados_completion;
   int ret = bs.bucket_obj.aio_operate(arg->rados_completion, &o);
   completion->release(); /* can't reference arg here, as it might have already been released */
@@ -8849,9 +8893,12 @@ int RGWRados::cls_obj_complete_op(BucketShard& bs, const rgw_obj& obj, RGWModify
 int RGWRados::cls_obj_complete_add(BucketShard& bs, const rgw_obj& obj, string& tag,
                                    int64_t pool, uint64_t epoch,
                                    rgw_bucket_dir_entry& ent, RGWObjCategory category,
-                                   list<rgw_obj_index_key> *remove_objs, uint16_t bilog_flags, rgw_zone_set *zones_trace)
+                                   list<rgw_obj_index_key> *remove_objs, uint16_t bilog_flags,
+                                   rgw_zone_set *zones_trace, bool log_op)
 {
-  return cls_obj_complete_op(bs, obj, CLS_RGW_OP_ADD, tag, pool, epoch, ent, category, remove_objs, bilog_flags, zones_trace);
+  return cls_obj_complete_op(bs, obj, CLS_RGW_OP_ADD, tag, pool, epoch,
+                             ent, category, remove_objs, bilog_flags,
+                             zones_trace, log_op);
 }
 
 int RGWRados::cls_obj_complete_del(BucketShard& bs, string& tag,
@@ -8860,26 +8907,27 @@ int RGWRados::cls_obj_complete_del(BucketShard& bs, string& tag,
                                    real_time& removed_mtime,
                                    list<rgw_obj_index_key> *remove_objs,
                                    uint16_t bilog_flags,
-                                   rgw_zone_set *zones_trace)
+                                   rgw_zone_set *zones_trace,
+                                   bool log_op)
 {
   rgw_bucket_dir_entry ent;
   ent.meta.mtime = removed_mtime;
   obj.key.get_index_key(&ent.key);
   return cls_obj_complete_op(bs, obj, CLS_RGW_OP_DEL, tag, pool, epoch,
 			     ent, RGWObjCategory::None, remove_objs,
-			     bilog_flags, zones_trace);
+			     bilog_flags, zones_trace, log_op);
 }
 
 int RGWRados::cls_obj_complete_cancel(BucketShard& bs, string& tag, rgw_obj& obj,
                                       list<rgw_obj_index_key> *remove_objs,
-                                      uint16_t bilog_flags, rgw_zone_set *zones_trace)
+                                      uint16_t bilog_flags, rgw_zone_set *zones_trace, bool log_op)
 {
   rgw_bucket_dir_entry ent;
   obj.key.get_index_key(&ent.key);
   return cls_obj_complete_op(bs, obj, CLS_RGW_OP_CANCEL, tag,
 			     -1 /* pool id */, 0, ent,
 			     RGWObjCategory::None, remove_objs, bilog_flags,
-			     zones_trace);
+			     zones_trace, log_op);
 }
 
 int RGWRados::cls_obj_set_bucket_tag_timeout(const DoutPrefixProvider *dpp, RGWBucketInfo& bucket_info, uint64_t timeout)
@@ -9846,6 +9894,10 @@ int RGWRados::check_bucket_shards(const RGWBucketInfo& bucket_info,
       return 0;
   }
 
+  if (! is_layout_reshardable(bucket_info.layout)) {
+    return 0;
+  }
+
   bool need_resharding = false;
   uint32_t num_source_shards = rgw::current_num_shards(bucket_info.layout);
   const uint32_t max_dynamic_shards =
diff --git a/src/rgw/driver/rados/rgw_rados.h b/src/rgw/driver/rados/rgw_rados.h
index 75a5e1b54..d77dd5c5f 100644
--- a/src/rgw/driver/rados/rgw_rados.h
+++ b/src/rgw/driver/rados/rgw_rados.h
@@ -812,9 +812,10 @@ public:
                      uint64_t size, uint64_t accounted_size,
                      std::map<std::string, bufferlist>& attrs,
                      bool modify_tail, bool assume_noent,
-                     void *index_op, optional_yield y);
+                     void *index_op, optional_yield y,
+                     bool log_op = true);
       int write_meta(const DoutPrefixProvider *dpp, uint64_t size, uint64_t accounted_size,
-                     std::map<std::string, bufferlist>& attrs, optional_yield y);
+                     std::map<std::string, bufferlist>& attrs, optional_yield y, bool log_op = true);
       int write_data(const char *data, uint64_t ofs, uint64_t len, bool exclusive);
       const req_state* get_req_state() {
         return nullptr;  /* XXX dang Only used by LTTng, and it handles null anyway */
@@ -852,7 +853,7 @@ public:
 
       explicit Delete(RGWRados::Object *_target) : target(_target) {}
 
-      int delete_obj(optional_yield y, const DoutPrefixProvider *dpp);
+      int delete_obj(optional_yield y, const DoutPrefixProvider *dpp, bool log_op = true);
     };
 
     struct Stat {
@@ -957,7 +958,7 @@ public:
         zones_trace = _zones_trace;
       }
 
-      int prepare(const DoutPrefixProvider *dpp, RGWModifyOp, const std::string *write_tag, optional_yield y);
+      int prepare(const DoutPrefixProvider *dpp, RGWModifyOp, const std::string *write_tag, optional_yield y, bool log_op = true);
       int complete(const DoutPrefixProvider *dpp, int64_t poolid, uint64_t epoch, uint64_t size,
                    uint64_t accounted_size, ceph::real_time& ut,
                    const std::string& etag, const std::string& content_type,
@@ -966,15 +967,18 @@ public:
 		   std::list<rgw_obj_index_key> *remove_objs,
 		   optional_yield y,
 		   const std::string *user_data = nullptr,
-		   bool appendable = false);
+		   bool appendable = false,
+                   bool log_op = true);
       int complete_del(const DoutPrefixProvider *dpp,
                        int64_t poolid, uint64_t epoch,
                        ceph::real_time& removed_mtime, /* mtime of removed object */
                        std::list<rgw_obj_index_key> *remove_objs,
-		       optional_yield y);
+                       optional_yield y,
+                       bool log_op = true);
       int cancel(const DoutPrefixProvider *dpp,
                  std::list<rgw_obj_index_key> *remove_objs,
-		 optional_yield y);
+                 optional_yield y,
+                 bool log_op = true);
 
       const std::string *get_optag() { return &optag; }
 
@@ -1195,7 +1199,8 @@ public:
 	       ceph::real_time delete_at,
                std::string *petag,
                const DoutPrefixProvider *dpp,
-               optional_yield y);
+               optional_yield y,
+               bool log_op = true);
 
   int transition_obj(RGWObjectCtx& obj_ctx,
                      RGWBucketInfo& bucket_info,
@@ -1204,7 +1209,8 @@ public:
                      const real_time& mtime,
                      uint64_t olh_epoch,
                      const DoutPrefixProvider *dpp,
-                     optional_yield y);
+                     optional_yield y,
+                     bool log_op = true);
 
   int check_bucket_empty(const DoutPrefixProvider *dpp, RGWBucketInfo& bucket_info, optional_yield y);
 
@@ -1234,7 +1240,8 @@ public:
 		 int versioning_status,  // versioning flags defined in enum RGWBucketFlags
 		 uint16_t bilog_flags = 0,
 		 const ceph::real_time& expiration_time = ceph::real_time(),
-		 rgw_zone_set *zones_trace = nullptr);
+		 rgw_zone_set *zones_trace = nullptr,
+                 bool log_op = true);
 
   int delete_raw_obj(const DoutPrefixProvider *dpp, const rgw_raw_obj& obj);
 
@@ -1319,7 +1326,8 @@ public:
                                    RGWBucketInfo& bucket_info,
                                    const rgw_obj& obj_instance,
                                    const std::string& op_tag, const std::string& olh_tag,
-                                   uint64_t olh_epoch, rgw_zone_set *zones_trace = nullptr);
+                                   uint64_t olh_epoch, rgw_zone_set *zones_trace = nullptr,
+                                   bool log_op = true);
   int bucket_index_read_olh_log(const DoutPrefixProvider *dpp,
                                 RGWBucketInfo& bucket_info, RGWObjState& state,
                                 const rgw_obj& obj_instance, uint64_t ver_marker,
@@ -1328,8 +1336,8 @@ public:
   int bucket_index_clear_olh(const DoutPrefixProvider *dpp, RGWBucketInfo& bucket_info, const std::string& olh_tag, const rgw_obj& obj_instance);
   int apply_olh_log(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx, RGWObjState& obj_state, RGWBucketInfo& bucket_info, const rgw_obj& obj,
                     bufferlist& obj_tag, std::map<uint64_t, std::vector<rgw_bucket_olh_log_entry> >& log,
-                    uint64_t *plast_ver, rgw_zone_set *zones_trace = nullptr);
-  int update_olh(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx, RGWObjState *state, RGWBucketInfo& bucket_info, const rgw_obj& obj, rgw_zone_set *zones_trace = nullptr);
+                    uint64_t *plast_ver, rgw_zone_set *zones_trace = nullptr, bool log_op = true);
+  int update_olh(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx, RGWObjState *state, RGWBucketInfo& bucket_info, const rgw_obj& obj, rgw_zone_set *zones_trace = nullptr, bool log_op = true);
   int clear_olh(const DoutPrefixProvider *dpp,
                 RGWObjectCtx& obj_ctx,
                 const rgw_obj& obj,
@@ -1343,7 +1351,7 @@ public:
   int repair_olh(const DoutPrefixProvider *dpp, RGWObjState* state, const RGWBucketInfo& bucket_info,
                  const rgw_obj& obj);
   int unlink_obj_instance(const DoutPrefixProvider *dpp, RGWObjectCtx& obj_ctx, RGWBucketInfo& bucket_info, const rgw_obj& target_obj,
-                          uint64_t olh_epoch, optional_yield y, rgw_zone_set *zones_trace = nullptr);
+                          uint64_t olh_epoch, optional_yield y, rgw_zone_set *zones_trace = nullptr, bool log_op = true);
 
   void check_pending_olh_entries(const DoutPrefixProvider *dpp, std::map<std::string, bufferlist>& pending_entries, std::map<std::string, bufferlist> *rm_pending_entries);
   int remove_olh_pending_entries(const DoutPrefixProvider *dpp, const RGWBucketInfo& bucket_info, RGWObjState& state, const rgw_obj& olh_obj, std::map<std::string, bufferlist>& pending_attrs);
@@ -1400,16 +1408,20 @@ public:
 			     std::map<std::string, bufferlist> *pattrs, bool create_entry_point,
                              const DoutPrefixProvider *dpp, optional_yield y);
 
-  int cls_obj_prepare_op(const DoutPrefixProvider *dpp, BucketShard& bs, RGWModifyOp op, std::string& tag, rgw_obj& obj, uint16_t bilog_flags, optional_yield y, rgw_zone_set *zones_trace = nullptr);
+  int cls_obj_prepare_op(const DoutPrefixProvider *dpp, BucketShard& bs, RGWModifyOp op, std::string& tag, rgw_obj& obj,
+                         uint16_t bilog_flags, optional_yield y, rgw_zone_set *zones_trace = nullptr, bool log_op = true);
   int cls_obj_complete_op(BucketShard& bs, const rgw_obj& obj, RGWModifyOp op, std::string& tag, int64_t pool, uint64_t epoch,
-                          rgw_bucket_dir_entry& ent, RGWObjCategory category, std::list<rgw_obj_index_key> *remove_objs, uint16_t bilog_flags, rgw_zone_set *zones_trace = nullptr);
+                          rgw_bucket_dir_entry& ent, RGWObjCategory category, std::list<rgw_obj_index_key> *remove_objs,
+                          uint16_t bilog_flags, rgw_zone_set *zones_trace = nullptr, bool log_op = true);
   int cls_obj_complete_add(BucketShard& bs, const rgw_obj& obj, std::string& tag, int64_t pool, uint64_t epoch, rgw_bucket_dir_entry& ent,
-                           RGWObjCategory category, std::list<rgw_obj_index_key> *remove_objs, uint16_t bilog_flags, rgw_zone_set *zones_trace = nullptr);
+                           RGWObjCategory category, std::list<rgw_obj_index_key> *remove_objs, uint16_t bilog_flags,
+                           rgw_zone_set *zones_trace = nullptr, bool log_op = true);
   int cls_obj_complete_del(BucketShard& bs, std::string& tag, int64_t pool, uint64_t epoch, rgw_obj& obj,
-                           ceph::real_time& removed_mtime, std::list<rgw_obj_index_key> *remove_objs, uint16_t bilog_flags, rgw_zone_set *zones_trace = nullptr);
+                           ceph::real_time& removed_mtime, std::list<rgw_obj_index_key> *remove_objs,
+                           uint16_t bilog_flags, rgw_zone_set *zones_trace = nullptr, bool log_op = true);
   int cls_obj_complete_cancel(BucketShard& bs, std::string& tag, rgw_obj& obj,
                               std::list<rgw_obj_index_key> *remove_objs,
-                              uint16_t bilog_flags, rgw_zone_set *zones_trace = nullptr);
+                              uint16_t bilog_flags, rgw_zone_set *zones_trace = nullptr, bool log_op = true);
   int cls_obj_set_bucket_tag_timeout(const DoutPrefixProvider *dpp, RGWBucketInfo& bucket_info, uint64_t timeout);
 
   using ent_map_t =
diff --git a/src/rgw/driver/rados/rgw_reshard.cc b/src/rgw/driver/rados/rgw_reshard.cc
index 2abf02908..25cb4df3c 100644
--- a/src/rgw/driver/rados/rgw_reshard.cc
+++ b/src/rgw/driver/rados/rgw_reshard.cc
@@ -994,11 +994,11 @@ int RGWBucketReshard::execute(int num_shards,
   return 0;
 } // execute
 
-bool RGWBucketReshard::can_reshard(const RGWBucketInfo& bucket,
-                                   const RGWSI_Zone* zone_svc)
+bool RGWBucketReshard::should_zone_reshard_now(const RGWBucketInfo& bucket,
+					       const RGWSI_Zone* zone_svc)
 {
   return !zone_svc->need_to_log_data() ||
-      bucket.layout.logs.size() < max_bilog_history;
+    bucket.layout.logs.size() < max_bilog_history;
 }
 
 
@@ -1240,7 +1240,7 @@ int RGWReshard::process_entry(const cls_rgw_reshard_entry& entry,
     return 0;
   }
 
-  if (!RGWBucketReshard::can_reshard(bucket_info, store->svc()->zone)) {
+  if (!RGWBucketReshard::should_zone_reshard_now(bucket_info, store->svc()->zone)) {
     ldpp_dout(dpp, 1) << "Bucket " << bucket_info.bucket << " is not "
         "eligible for resharding until peer zones finish syncing one "
         "or more of its old log generations" << dendl;
diff --git a/src/rgw/driver/rados/rgw_reshard.h b/src/rgw/driver/rados/rgw_reshard.h
index 59819f3a5..7388c459e 100644
--- a/src/rgw/driver/rados/rgw_reshard.h
+++ b/src/rgw/driver/rados/rgw_reshard.h
@@ -175,8 +175,8 @@ public:
   // too large by refusing to reshard the bucket until the old logs get trimmed
   static constexpr size_t max_bilog_history = 4;
 
-  static bool can_reshard(const RGWBucketInfo& bucket,
-                          const RGWSI_Zone* zone_svc);
+  static bool should_zone_reshard_now(const RGWBucketInfo& bucket,
+				      const RGWSI_Zone* zone_svc);
 }; // RGWBucketReshard
 
 
diff --git a/src/rgw/driver/rados/rgw_sal_rados.cc b/src/rgw/driver/rados/rgw_sal_rados.cc
index 9acdb79d3..414f44bb8 100644
--- a/src/rgw/driver/rados/rgw_sal_rados.cc
+++ b/src/rgw/driver/rados/rgw_sal_rados.cc
@@ -1972,9 +1972,11 @@ int RadosObject::transition(Bucket* bucket,
 			    const real_time& mtime,
 			    uint64_t olh_epoch,
 			    const DoutPrefixProvider* dpp,
-			    optional_yield y)
+			    optional_yield y,
+                            uint32_t flags)
 {
-  return store->getRados()->transition_obj(*rados_ctx, bucket->get_info(), get_obj(), placement_rule, mtime, olh_epoch, dpp, y);
+  return store->getRados()->transition_obj(*rados_ctx, bucket->get_info(), get_obj(), placement_rule,
+                                           mtime, olh_epoch, dpp, y, flags & FLAG_LOG_OP);
 }
 
 int RadosObject::transition_to_cloud(Bucket* bucket,
@@ -2253,7 +2255,7 @@ RadosObject::RadosDeleteOp::RadosDeleteOp(RadosObject *_source) :
 	parent_op(&op_target)
 { }
 
-int RadosObject::RadosDeleteOp::delete_obj(const DoutPrefixProvider* dpp, optional_yield y)
+int RadosObject::RadosDeleteOp::delete_obj(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags)
 {
   parent_op.params.bucket_owner = params.bucket_owner.get_id();
   parent_op.params.versioning_status = params.versioning_status;
@@ -2270,7 +2272,7 @@ int RadosObject::RadosDeleteOp::delete_obj(const DoutPrefixProvider* dpp, option
   parent_op.params.abortmp = params.abortmp;
   parent_op.params.parts_accounted_size = params.parts_accounted_size;
 
-  int ret = parent_op.delete_obj(y, dpp);
+  int ret = parent_op.delete_obj(y, dpp, flags & FLAG_LOG_OP);
   if (ret < 0)
     return ret;
 
@@ -2282,15 +2284,16 @@ int RadosObject::RadosDeleteOp::delete_obj(const DoutPrefixProvider* dpp, option
 
 int RadosObject::delete_object(const DoutPrefixProvider* dpp,
 			       optional_yield y,
-			       bool prevent_versioning)
+			       uint32_t flags)
 {
   RGWRados::Object del_target(store->getRados(), bucket->get_info(), *rados_ctx, get_obj());
   RGWRados::Object::Delete del_op(&del_target);
 
   del_op.params.bucket_owner = bucket->get_info().owner;
-  del_op.params.versioning_status = prevent_versioning ? 0 : bucket->get_info().versioning_status();
+  del_op.params.versioning_status = (flags & FLAG_PREVENT_VERSIONING)
+                                    ? 0 : bucket->get_info().versioning_status();
 
-  return del_op.delete_obj(y, dpp);
+  return del_op.delete_obj(y, dpp, flags & FLAG_LOG_OP);
 }
 
 int RadosObject::delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate,
@@ -2459,7 +2462,7 @@ int RadosMultipartUpload::abort(const DoutPrefixProvider *dpp, CephContext *cct)
 	std::unique_ptr<rgw::sal::Object> obj = bucket->get_object(
 				    rgw_obj_key(obj_part->oid, std::string(), RGW_OBJ_NS_MULTIPART));
 	obj->set_hash_source(mp_obj.get_key());
-	ret = obj->delete_object(dpp, null_yield);
+	ret = obj->delete_object(dpp, null_yield, 0);
         if (ret < 0 && ret != -ENOENT)
           return ret;
       } else {
@@ -2509,7 +2512,7 @@ int RadosMultipartUpload::abort(const DoutPrefixProvider *dpp, CephContext *cct)
   del_op->params.parts_accounted_size = parts_accounted_size;
 
   // and also remove the metadata obj
-  ret = del_op->delete_obj(dpp, null_yield);
+  ret = del_op->delete_obj(dpp, null_yield, 0);
   if (ret < 0) {
     ldpp_dout(dpp, 20) << __func__ << ": del_op.delete_obj returned " <<
       ret << dendl;
@@ -2694,7 +2697,7 @@ int RadosMultipartUpload::complete(const DoutPrefixProvider *dpp,
   int marker = 0;
   uint64_t min_part_size = cct->_conf->rgw_multipart_min_part_size;
   auto etags_iter = part_etags.begin();
-  rgw::sal::Attrs attrs = target_obj->get_attrs();
+  rgw::sal::Attrs& attrs = target_obj->get_attrs();
 
   do {
     ret = list_parts(dpp, cct, max_parts, marker, &marker, &truncated);
@@ -3120,10 +3123,11 @@ int RadosAtomicWriter::complete(size_t accounted_size, const std::string& etag,
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y)
+                       optional_yield y,
+                       uint32_t flags)
 {
   return processor.complete(accounted_size, etag, mtime, set_mtime, attrs, delete_at,
-			    if_match, if_nomatch, user_data, zones_trace, canceled, y);
+			    if_match, if_nomatch, user_data, zones_trace, canceled, y, flags);
 }
 
 int RadosAppendWriter::prepare(optional_yield y)
@@ -3143,10 +3147,11 @@ int RadosAppendWriter::complete(size_t accounted_size, const std::string& etag,
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y)
+                       optional_yield y,
+                       uint32_t flags)
 {
   return processor.complete(accounted_size, etag, mtime, set_mtime, attrs, delete_at,
-			    if_match, if_nomatch, user_data, zones_trace, canceled, y);
+			    if_match, if_nomatch, user_data, zones_trace, canceled, y, flags);
 }
 
 int RadosMultipartWriter::prepare(optional_yield y)
@@ -3166,10 +3171,11 @@ int RadosMultipartWriter::complete(size_t accounted_size, const std::string& eta
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y)
+                       optional_yield y,
+                       uint32_t flags)
 {
   return processor.complete(accounted_size, etag, mtime, set_mtime, attrs, delete_at,
-			    if_match, if_nomatch, user_data, zones_trace, canceled, y);
+			    if_match, if_nomatch, user_data, zones_trace, canceled, y, flags);
 }
 
 const std::string& RadosZoneGroup::get_endpoint() const
@@ -3331,7 +3337,7 @@ const std::string_view RadosZone::get_tier_type()
   if (local_zone)
     return store->svc()->zone->get_zone().tier_type;
 
-  return rgw_zone.id;
+  return rgw_zone.tier_type;
 }
 
 RGWBucketSyncPolicyHandlerRef RadosZone::get_sync_policy_handler()
diff --git a/src/rgw/driver/rados/rgw_sal_rados.h b/src/rgw/driver/rados/rgw_sal_rados.h
index 4d2dc9709..299be16e6 100644
--- a/src/rgw/driver/rados/rgw_sal_rados.h
+++ b/src/rgw/driver/rados/rgw_sal_rados.h
@@ -370,7 +370,7 @@ class RadosObject : public StoreObject {
     public:
       RadosDeleteOp(RadosObject* _source);
 
-      virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y) override;
+      virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags) override;
     };
 
     RadosObject(RadosStore *_st, const rgw_obj_key& _k)
@@ -402,7 +402,7 @@ class RadosObject : public StoreObject {
       rados_ctx->invalidate(get_obj());
     }
     virtual int delete_object(const DoutPrefixProvider* dpp,
-			      optional_yield y, bool prevent_versioning) override;
+			      optional_yield y, uint32_t flags) override;
     virtual int delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate, Completions* aio,
 			       bool keep_index_consistent, optional_yield y) override;
     virtual int copy_object(User* user,
@@ -453,7 +453,8 @@ class RadosObject : public StoreObject {
 			   const real_time& mtime,
 			   uint64_t olh_epoch,
 			   const DoutPrefixProvider* dpp,
-			   optional_yield y) override;
+			   optional_yield y,
+                           uint32_t flags) override;
     virtual int transition_to_cloud(Bucket* bucket,
 			   rgw::sal::PlacementTier* tier,
 			   rgw_bucket_dir_entry& o,
@@ -822,7 +823,8 @@ public:
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y) override;
+                       optional_yield y,
+                       uint32_t flags) override;
 };
 
 class RadosAppendWriter : public StoreWriter {
@@ -869,7 +871,8 @@ public:
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y) override;
+                       optional_yield y,
+                       uint32_t flags) override;
 };
 
 class RadosMultipartWriter : public StoreWriter {
@@ -914,7 +917,8 @@ public:
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y) override;
+                       optional_yield y,
+                       uint32_t flags) override;
 };
 
 class RadosLuaManager : public StoreLuaManager {
diff --git a/src/rgw/driver/rados/rgw_sync_module_aws.cc b/src/rgw/driver/rados/rgw_sync_module_aws.cc
index cefcd9dd1..cdcd831e9 100644
--- a/src/rgw/driver/rados/rgw_sync_module_aws.cc
+++ b/src/rgw/driver/rados/rgw_sync_module_aws.cc
@@ -487,7 +487,7 @@ struct AWSSyncConfig {
   }
 
   bool do_find_profile(const rgw_bucket bucket, std::shared_ptr<AWSSyncConfig_Profile> *result) {
-    const string& name = bucket.name;
+    const string& name = bucket.get_namespaced_name();
     auto iter = explicit_profiles.upper_bound(name);
     if (iter == explicit_profiles.begin()) {
       return false;
diff --git a/src/rgw/driver/rados/rgw_tools.cc b/src/rgw/driver/rados/rgw_tools.cc
index 66651da5c..bc58c71ea 100644
--- a/src/rgw/driver/rados/rgw_tools.cc
+++ b/src/rgw/driver/rados/rgw_tools.cc
@@ -422,7 +422,8 @@ int RGWDataAccess::Object::put(bufferlist& data,
 			    attrs, delete_at,
                             nullptr, nullptr,
                             puser_data,
-                            nullptr, nullptr, y);
+                            nullptr, nullptr, y,
+                            rgw::sal::FLAG_LOG_OP);
 }
 
 void RGWDataAccess::Object::set_policy(const RGWAccessControlPolicy& policy)
diff --git a/src/rgw/driver/rados/rgw_zone.cc b/src/rgw/driver/rados/rgw_zone.cc
index ed09f24f6..5c3f55b3d 100644
--- a/src/rgw/driver/rados/rgw_zone.cc
+++ b/src/rgw/driver/rados/rgw_zone.cc
@@ -1005,10 +1005,12 @@ int create_zone(const DoutPrefixProvider* dpp, optional_yield y,
   }
 
   // add default placement with empty pool name
+  RGWZonePlacementInfo placement;
   rgw_pool pool;
-  auto& placement = info.placement_pools["default-placement"];
   placement.storage_classes.set_storage_class(
       RGW_STORAGE_CLASS_STANDARD, &pool, nullptr);
+  // don't overwrite if it already exists
+  info.placement_pools.emplace("default-placement", std::move(placement));
 
   // build a set of all pool names used by other zones
   std::set<rgw_pool> pools;
diff --git a/src/rgw/rgw_admin.cc b/src/rgw/rgw_admin.cc
index 73b0736b1..5cabd2a9b 100644
--- a/src/rgw/rgw_admin.cc
+++ b/src/rgw/rgw_admin.cc
@@ -3025,6 +3025,14 @@ int check_reshard_bucket_params(rgw::sal::Driver* driver,
     return ret;
   }
 
+  if (! is_layout_reshardable((*bucket)->get_info().layout)) {
+    std::cerr << "Bucket '" << (*bucket)->get_name() <<
+      "' currently has layout '" <<
+      current_layout_desc((*bucket)->get_info().layout) <<
+      "', which does not support resharding." << std::endl;
+    return -EINVAL;
+  }
+
   int num_source_shards = rgw::current_num_shards((*bucket)->get_info().layout);
 
   if (num_shards <= num_source_shards && !yes_i_really_mean_it) {
@@ -8107,7 +8115,8 @@ next:
           "have the resharding feature enabled." << std::endl;
       return ENOTSUP;
     }
-    if (!RGWBucketReshard::can_reshard(bucket->get_info(), zone_svc) &&
+
+    if (!RGWBucketReshard::should_zone_reshard_now(bucket->get_info(), zone_svc) &&
         !yes_i_really_mean_it) {
       std::cerr << "Bucket '" << bucket->get_name() << "' already has too many "
           "log generations (" << bucket->get_info().layout.logs.size() << ") "
diff --git a/src/rgw/rgw_asio_frontend.cc b/src/rgw/rgw_asio_frontend.cc
index 633a29633..d84264a0a 100644
--- a/src/rgw/rgw_asio_frontend.cc
+++ b/src/rgw/rgw_asio_frontend.cc
@@ -1036,10 +1036,10 @@ void AsioFrontend::accept(Listener& l, boost::system::error_code ec)
         handle_connection(context, env, stream, timeout, header_limit,
                           conn->buffer, true, pause_mutex, scheduler.get(),
                           uri_prefix, ec, yield);
-        if (!ec) {
-          // ssl shutdown (ignoring errors)
-          stream.async_shutdown(yield[ec]);
-        }
+
+        // ssl shutdown (ignoring errors)
+        stream.async_shutdown(yield[ec]);
+        
         conn->socket.shutdown(tcp::socket::shutdown_both, ec);
       }, make_stack_allocator());
   } else {
diff --git a/src/rgw/rgw_auth.cc b/src/rgw/rgw_auth.cc
index 2c61b8361..7be651851 100644
--- a/src/rgw/rgw_auth.cc
+++ b/src/rgw/rgw_auth.cc
@@ -172,7 +172,7 @@ strategy_handle_rejected(rgw::auth::Engine::result_t&& engine_result,
 
     case Control::FALLBACK:
       /* Don't try next. */
-      return std::make_pair(false, std::move(strategy_result));
+      return std::make_pair(false, std::move(engine_result));
 
     default:
       /* Huh, memory corruption? */
diff --git a/src/rgw/rgw_auth_s3.cc b/src/rgw/rgw_auth_s3.cc
index 0797f8184..dcd29bfca 100644
--- a/src/rgw/rgw_auth_s3.cc
+++ b/src/rgw/rgw_auth_s3.cc
@@ -574,7 +574,7 @@ std::string get_v4_canonical_qs(const req_info& info, const bool using_qs)
 
   /* Handle case when query string exists. Step 3 described in: http://docs.
    * aws.amazon.com/general/latest/gr/sigv4-create-canonical-request.html */
-  std::map<std::string, std::string> canonical_qs_map;
+  std::multimap<std::string, std::string> canonical_qs_map;
   for (const auto& s : get_str_vec<5>(*params, "&")) {
     std::string_view key, val;
     const auto parsed_pair = parse_key_value(s);
@@ -595,7 +595,7 @@ std::string get_v4_canonical_qs(const req_info& info, const bool using_qs)
     // while awsv4 specs ask for all slashes to be encoded, s3 itself is relaxed
     // in its implementation allowing non-url-encoded slashes to be present in
     // presigned urls for instance
-    canonical_qs_map[aws4_uri_recode(key, true)] = aws4_uri_recode(val, true);
+    canonical_qs_map.insert({{aws4_uri_recode(key, true), aws4_uri_recode(val, true)}});
   }
 
   /* Thanks to the early exist we have the guarantee that canonical_qs_map has
diff --git a/src/rgw/rgw_bucket_layout.h b/src/rgw/rgw_bucket_layout.h
index 40aafd4dd..114f1f1ff 100644
--- a/src/rgw/rgw_bucket_layout.h
+++ b/src/rgw/rgw_bucket_layout.h
@@ -278,5 +278,14 @@ inline uint32_t current_num_shards(const BucketLayout& layout) {
 inline bool is_layout_indexless(const bucket_index_layout_generation& layout) {
   return layout.layout.type == BucketIndexType::Indexless;
 }
+inline bool is_layout_reshardable(const bucket_index_layout_generation& layout) {
+  return layout.layout.type == BucketIndexType::Normal;
+}
+inline bool is_layout_reshardable(const BucketLayout& layout) {
+  return is_layout_reshardable(layout.current_index);
+}
+inline std::string_view current_layout_desc(const BucketLayout& layout) {
+  return rgw::to_string(layout.current_index.layout.type);
+}
 
 } // namespace rgw
diff --git a/src/rgw/rgw_bucket_types.h b/src/rgw/rgw_bucket_types.h
index 61acc58bb..52ac5dc34 100644
--- a/src/rgw/rgw_bucket_types.h
+++ b/src/rgw/rgw_bucket_types.h
@@ -136,6 +136,13 @@ struct rgw_bucket {
     DECODE_FINISH(bl);
   }
 
+  std::string get_namespaced_name() const {
+    if (tenant.empty()) {
+      return name;
+    }
+    return tenant + std::string("/") + name;
+  }
+
   void update_bucket_id(const std::string& new_bucket_id) {
     bucket_id = new_bucket_id;
   }
diff --git a/src/rgw/rgw_cors.cc b/src/rgw/rgw_cors.cc
index 83ba079b2..61e92696d 100644
--- a/src/rgw/rgw_cors.cc
+++ b/src/rgw/rgw_cors.cc
@@ -95,6 +95,8 @@ static bool is_string_in_set(set<string>& s, string h) {
       
       get_str_list((*it), "* \t", ssplit);
       if (off != 0) {
+        if (ssplit.empty())
+          continue;
         string sl = ssplit.front();
         flen = sl.length();
         dout(10) << "Finding " << sl << ", in " << h << ", at offset 0" << dendl;
@@ -103,6 +105,8 @@ static bool is_string_in_set(set<string>& s, string h) {
         ssplit.pop_front();
       }
       if (off != ((*it).length() - 1)) {
+        if (ssplit.empty())
+          continue;
         string sl = ssplit.front();
         dout(10) << "Finding " << sl << ", in " << h 
           << ", at offset not less than " << flen << dendl;
diff --git a/src/rgw/rgw_file.cc b/src/rgw/rgw_file.cc
index ee32170a1..4d2df63c6 100644
--- a/src/rgw/rgw_file.cc
+++ b/src/rgw/rgw_file.cc
@@ -2012,7 +2012,7 @@ namespace rgw {
     op_ret = processor->complete(state->obj_size, etag, &mtime, real_time(), attrs,
                                  (delete_at ? *delete_at : real_time()),
                                 if_match, if_nomatch, nullptr, nullptr, nullptr,
-                                state->yield);
+                                state->yield, rgw::sal::FLAG_LOG_OP);
     if (op_ret != 0) {
       /* revert attr updates */
       rgw_fh->set_mtime(omtime);
diff --git a/src/rgw/rgw_kafka.cc b/src/rgw/rgw_kafka.cc
index 642787a38..5b9d2f27c 100644
--- a/src/rgw/rgw_kafka.cc
+++ b/src/rgw/rgw_kafka.cc
@@ -100,8 +100,9 @@ struct connection_t {
     // fire all remaining callbacks (if not fired by rd_kafka_flush)
     std::for_each(callbacks.begin(), callbacks.end(), [this](auto& cb_tag) {
         cb_tag.cb(status);
-        ldout(cct, 20) << "Kafka destroy: invoking callback with tag=" << cb_tag.tag << 
-          " for: " << broker << dendl;
+        ldout(cct, 20) << "Kafka destroy: invoking callback with tag="
+                       << cb_tag.tag << " for: " << broker
+                       << " with status: " << status << dendl;
       });
     callbacks.clear();
     delivery_tag = 1;
@@ -418,7 +419,9 @@ private:
     if (tag) {
       auto const q_len = conn->callbacks.size();
       if (q_len < max_inflight) {
-        ldout(conn->cct, 20) << "Kafka publish (with callback, tag=" << *tag << "): OK. Queue has: " << q_len << " callbacks" << dendl;
+        ldout(conn->cct, 20)
+            << "Kafka publish (with callback, tag=" << *tag
+            << "): OK. Queue has: " << q_len + 1 << " callbacks" << dendl;
         conn->callbacks.emplace_back(*tag, message->cb);
       } else {
         // immediately invoke callback with error - this is not a connection error
@@ -462,7 +465,7 @@ private:
         if(conn->timestamp.sec() + max_idle_time < ceph_clock_now()) {
           ldout(conn->cct, 20) << "kafka run: deleting a connection due to idle behaviour: " << ceph_clock_now() << dendl;
           std::lock_guard lock(connections_lock);
-          conn->destroy(STATUS_CONNECTION_IDLE);
+          conn->status = STATUS_CONNECTION_IDLE;
           conn_it = connections.erase(conn_it);
           --connection_count; \
           continue;
diff --git a/src/rgw/rgw_lc.cc b/src/rgw/rgw_lc.cc
index 7f4a79501..74bc8b12c 100644
--- a/src/rgw/rgw_lc.cc
+++ b/src/rgw/rgw_lc.cc
@@ -41,6 +41,9 @@
 #define dout_context g_ceph_context
 #define dout_subsys ceph_subsys_rgw
 
+constexpr int32_t hours_in_a_day = 24;
+constexpr int32_t secs_in_a_day = hours_in_a_day * 60 * 60;
+
 using namespace std;
 
 const char* LC_STATUS[] = {
@@ -289,7 +292,7 @@ static bool obj_has_expired(const DoutPrefixProvider *dpp, CephContext *cct, cep
   utime_t base_time;
   if (cct->_conf->rgw_lc_debug_interval <= 0) {
     /* Normal case, run properly */
-    cmp = double(days)*24*60*60;
+    cmp = double(days) * secs_in_a_day;
     base_time = ceph_clock_now().round_to_day();
   } else {
     /* We're in debug mode; Treat each rgw_lc_debug_interval seconds as a day */
@@ -509,6 +512,28 @@ struct lc_op_ctx {
 static std::string lc_id = "rgw lifecycle";
 static std::string lc_req_id = "0";
 
+/* do all zones in the zone group process LC? */
+static bool zonegroup_lc_check(const DoutPrefixProvider *dpp, rgw::sal::Zone* zone)
+{
+  auto& zonegroup = zone->get_zonegroup();
+  std::list<std::string> ids;
+  int ret = zonegroup.list_zones(ids);
+  if (ret < 0) {
+    return false;
+  }
+
+  return std::all_of(ids.begin(), ids.end(), [&](const auto& id) {
+    std::unique_ptr<rgw::sal::Zone> zone;
+    ret = zonegroup.get_zone_by_id(id, &zone);
+    if (ret < 0) {
+      return false;
+    }
+    const auto& tier_type = zone->get_tier_type();
+    ldpp_dout(dpp, 20) << "checking zone tier_type=" << tier_type << dendl;
+    return (tier_type == "rgw" || tier_type == "archive" || tier_type == "");
+  });
+}
+
 static int remove_expired_obj(
   const DoutPrefixProvider *dpp, lc_op_ctx& oc, bool remove_indeed,
   rgw::notify::EventType event_type)
@@ -579,7 +604,10 @@ static int remove_expired_obj(
       << dendl;
     return ret;
   }
-  ret =  del_op->delete_obj(dpp, null_yield);
+
+  uint32_t flags = (!remove_indeed || !zonegroup_lc_check(dpp, oc.driver->get_zone()))
+                   ? rgw::sal::FLAG_LOG_OP : 0;
+  ret =  del_op->delete_obj(dpp, null_yield, flags);
   if (ret < 0) {
     ldpp_dout(dpp, 1) <<
       "ERROR: publishing notification failed, with error: " << ret << dendl;
@@ -843,6 +871,7 @@ int RGWLC::handle_multipart_expiration(rgw::sal::Bucket* target,
     if (obj_has_expired(this, cct, obj.meta.mtime, rule.mp_expiration)) {
       rgw_obj_key key(obj.key);
       std::unique_ptr<rgw::sal::MultipartUpload> mpu = target->get_multipart_upload(key.name);
+
       int ret = mpu->abort(this, cct);
       if (ret == 0) {
         if (perfcounter) {
@@ -1353,8 +1382,10 @@ public:
         return -EINVAL;
       }
 
+      uint32_t flags = !zonegroup_lc_check(oc.dpp, oc.driver->get_zone())
+                       ? rgw::sal::FLAG_LOG_OP : 0;
       int r = oc.obj->transition(oc.bucket, target_placement, o.meta.mtime,
-	  		         o.versioned_epoch, oc.dpp, null_yield);
+                                 o.versioned_epoch, oc.dpp, null_yield, flags);
       if (r < 0) {
         ldpp_dout(oc.dpp, 0) << "ERROR: failed to transition obj " 
 			     << oc.bucket << ":" << o.key 
@@ -1874,8 +1905,7 @@ bool RGWLC::expired_session(time_t started)
   }
 
   time_t interval = (cct->_conf->rgw_lc_debug_interval > 0)
-    ? cct->_conf->rgw_lc_debug_interval
-    : 24*60*60;
+    ? cct->_conf->rgw_lc_debug_interval : secs_in_a_day;
 
   auto now = time(nullptr);
 
@@ -1891,8 +1921,7 @@ bool RGWLC::expired_session(time_t started)
 time_t RGWLC::thread_stop_at()
 {
   uint64_t interval = (cct->_conf->rgw_lc_debug_interval > 0)
-    ? cct->_conf->rgw_lc_debug_interval
-    : 24*60*60;
+    ? cct->_conf->rgw_lc_debug_interval : secs_in_a_day;
 
   return time(nullptr) + interval;
 }
@@ -1983,7 +2012,7 @@ static inline bool allow_shard_rollover(CephContext* cct, time_t now, time_t sha
    *    - the current shard has not rolled over in the last 24 hours
    */
   if (((shard_rollover_date < now) &&
-       (now - shard_rollover_date > 24*60*60)) ||
+       (now - shard_rollover_date > secs_in_a_day)) ||
       (! shard_rollover_date /* no rollover date stored */) ||
       (cct->_conf->rgw_lc_debug_interval > 0 /* defaults to -1 == disabled */)) {
     return true;
@@ -2009,7 +2038,7 @@ static inline bool already_run_today(CephContext* cct, time_t start_date)
   bdt.tm_min = 0;
   bdt.tm_sec = 0;
   begin_of_day = mktime(&bdt);
-  if (now - begin_of_day < 24*60*60)
+  if (now - begin_of_day < secs_in_a_day)
     return true;
   else
     return false;
@@ -2346,6 +2375,12 @@ bool RGWLC::LCWorker::should_work(utime_t& now)
   time_t tt = now.sec();
   localtime_r(&tt, &bdt);
 
+  // next-day adjustment if the configured end_hour is less than start_hour
+  if (end_hour < start_hour) {
+    bdt.tm_hour = bdt.tm_hour > end_hour ? bdt.tm_hour : bdt.tm_hour + hours_in_a_day;
+    end_hour += hours_in_a_day;
+  }
+
   if (cct->_conf->rgw_lc_debug_interval > 0) {
 	  /* We're debugging, so say we can run */
 	  return true;
@@ -2386,7 +2421,7 @@ int RGWLC::LCWorker::schedule_next_start_time(utime_t &start, utime_t& now)
   nt = mktime(&bdt);
   secs = nt - tt;
 
-  return secs>0 ? secs : secs+24*60*60;
+  return secs > 0 ? secs : secs + secs_in_a_day;
 }
 
 RGWLC::LCWorker::~LCWorker()
@@ -2677,7 +2712,7 @@ std::string s3_expiration_header(
       if (rule_expiration.has_days()) {
 	rule_expiration_date =
 	  boost::optional<ceph::real_time>(
-	    mtime + make_timespan(double(rule_expiration.get_days())*24*60*60 - ceph::real_clock::to_time_t(mtime)%(24*60*60) + 24*60*60));
+	    mtime + make_timespan(double(rule_expiration.get_days()) * secs_in_a_day - ceph::real_clock::to_time_t(mtime)%(secs_in_a_day) + secs_in_a_day));
       }
     }
 
@@ -2756,7 +2791,7 @@ bool s3_multipart_abort_header(
     std::optional<ceph::real_time> rule_abort_date;
     if (mp_expiration.has_days()) {
       rule_abort_date = std::optional<ceph::real_time>(
-              mtime + make_timespan(mp_expiration.get_days()*24*60*60 - ceph::real_clock::to_time_t(mtime)%(24*60*60) + 24*60*60));
+              mtime + make_timespan(mp_expiration.get_days() * secs_in_a_day - ceph::real_clock::to_time_t(mtime)%(secs_in_a_day) + secs_in_a_day));
     }
 
     // update earliest abort date
diff --git a/src/rgw/rgw_lib.cc b/src/rgw/rgw_lib.cc
index f449cce21..b105e1dad 100644
--- a/src/rgw/rgw_lib.cc
+++ b/src/rgw/rgw_lib.cc
@@ -468,6 +468,7 @@ namespace rgw {
 
   int RGWLib::init(vector<const char*>& args)
   {
+    int r{0};
     /* alternative default for module */
     map<std::string,std::string> defaults = {
       { "debug_rgw", "1/5" },
@@ -524,7 +525,13 @@ namespace rgw {
     register_async_signal_handler(SIGUSR1, rgw::signal::handle_sigterm);
 
     main.init_tracepoints();
-    main.init_frontends2(this /* rgwlib */);
+    r = main.init_frontends2(this /* rgwlib */);
+    if (r != 0) {
+      derr << "ERROR: unable to initialize frontend, r = " << r << dendl;
+      main.shutdown();
+      return r;
+    }
+
     main.init_notification_endpoints();
     main.init_lua();
 
diff --git a/src/rgw/rgw_lua_request.cc b/src/rgw/rgw_lua_request.cc
index 6d324d4fc..551f5fd72 100644
--- a/src/rgw/rgw_lua_request.cc
+++ b/src/rgw/rgw_lua_request.cc
@@ -786,7 +786,7 @@ struct RequestMetaTable : public EmptyMetaTable {
       create_metatable<ObjectMetaTable>(L, false, s->object);
     } else if (strcasecmp(index, "CopyFrom") == 0) {
       if (s->op_type == RGW_OP_COPY_OBJ) {
-        create_metatable<CopyFromMetaTable>(L, s);
+        create_metatable<CopyFromMetaTable>(L, false, s);
       } else {
         lua_pushnil(L);
       }
diff --git a/src/rgw/rgw_main.cc b/src/rgw/rgw_main.cc
index 6d2630251..61cea7cb7 100644
--- a/src/rgw/rgw_main.cc
+++ b/src/rgw/rgw_main.cc
@@ -157,7 +157,12 @@ int main(int argc, char *argv[])
   main.init_opslog();
   main.init_tracepoints();
   main.init_lua();
-  main.init_frontends2(nullptr /* RGWLib */);
+  r = main.init_frontends2(nullptr /* RGWLib */);
+  if (r != 0) {
+    derr << "ERROR:  initialize frontend fail, r = " << r << dendl;
+    main.shutdown();
+    return r;
+  }
   main.init_notification_endpoints();
 
 #if defined(HAVE_SYS_PRCTL_H)
diff --git a/src/rgw/rgw_object_lock.h b/src/rgw/rgw_object_lock.h
index 27c73feae..9392c7ead 100644
--- a/src/rgw/rgw_object_lock.h
+++ b/src/rgw/rgw_object_lock.h
@@ -170,16 +170,20 @@ public:
   }
 
   void encode(bufferlist& bl) const {
-    ENCODE_START(1, 1, bl);
+    ENCODE_START(2, 1, bl);
     encode(mode, bl);
     encode(retain_until_date, bl);
+    ceph::round_trip_encode(retain_until_date, bl);
     ENCODE_FINISH(bl);
   }
 
   void decode(bufferlist::const_iterator& bl) {
-    DECODE_START(1, bl);
+    DECODE_START(2, bl);
     decode(mode, bl);
     decode(retain_until_date, bl);
+    if (struct_v >= 2) {
+      ceph::round_trip_decode(retain_until_date, bl);
+    }
     DECODE_FINISH(bl);
   }
 
diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index 71fb198f3..bb0bfdbfa 100644
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -621,18 +621,29 @@ int rgw_build_bucket_policies(const DoutPrefixProvider *dpp, rgw::sal::Driver* d
       }
     } catch (const std::exception& e) {
       ldpp_dout(dpp, -1) << "Error reading IAM User Policy: " << e.what() << dendl;
-      ret = -EACCES;
+      if (!s->system_request) {
+        ret = -EACCES;
+      }
     }
   }
 
   try {
     s->iam_policy = get_iam_policy_from_attr(s->cct, s->bucket_attrs, s->bucket_tenant);
   } catch (const std::exception& e) {
-    // Really this is a can't happen condition. We parse the policy
-    // when it's given to us, so perhaps we should abort or otherwise
-    // raise bloody murder.
     ldpp_dout(dpp, 0) << "Error reading IAM Policy: " << e.what() << dendl;
-    ret = -EACCES;
+
+    // This really shouldn't happen. We parse the policy when it's given to us,
+    // so a parsing failure here means we broke backward compatibility. The only
+    // sensible thing to do in this case is to deny access, because the policy
+    // may have.
+    //
+    // However, the only way for an administrator to repair such a bucket is to
+    // send a PutBucketPolicy or DeleteBucketPolicy request as an admin/system
+    // user. We can allow such requests, because even if the policy denied
+    // access, admin/system users override that error from verify_permission().
+    if (!s->system_request) {
+      ret = -EACCES;
+    }
   }
 
   bool success = driver->get_zone()->get_redirect_endpoint(&s->redirect_zone_endpoint);
@@ -2229,11 +2240,11 @@ void RGWGetObj::execute(optional_yield y)
   read_op->params.lastmod = &lastmod;
 
   op_ret = read_op->prepare(s->yield, this);
-  if (op_ret < 0)
-    goto done_err;
   version_id = s->object->get_instance();
   s->obj_size = s->object->get_obj_size();
   attrs = s->object->get_attrs();
+  if (op_ret < 0)
+    goto done_err;
 
   /* STAT ops don't need data, and do no i/o */
   if (get_type() == RGW_OP_STAT_OBJ) {
@@ -4350,7 +4361,7 @@ void RGWPutObj::execute(optional_yield y)
   op_ret = processor->complete(s->obj_size, etag, &mtime, real_time(), attrs,
                                (delete_at ? *delete_at : real_time()), if_match, if_nomatch,
                                (user_data.empty() ? nullptr : &user_data), nullptr, nullptr,
-                               s->yield);
+                               s->yield, rgw::sal::FLAG_LOG_OP);
   tracepoint(rgw_op, processor_complete_exit, s->req_id.c_str());
 
   /* produce torrent */
@@ -4628,7 +4639,7 @@ void RGWPostObj::execute(optional_yield y)
     op_ret = processor->complete(s->obj_size, etag, nullptr, real_time(), attrs,
                                 (delete_at ? *delete_at : real_time()),
                                 nullptr, nullptr, nullptr, nullptr, nullptr,
-                                s->yield);
+                                s->yield, rgw::sal::FLAG_LOG_OP);
     if (op_ret < 0) {
       return;
     }
@@ -5191,7 +5202,7 @@ void RGWDeleteObj::execute(optional_yield y)
       del_op->params.olh_epoch = epoch;
       del_op->params.marker_version_id = version_id;
 
-      op_ret = del_op->delete_obj(this, y);
+      op_ret = del_op->delete_obj(this, y, rgw::sal::FLAG_LOG_OP);
       if (op_ret >= 0) {
 	delete_marker = del_op->result.delete_marker;
 	version_id = del_op->result.version_id;
@@ -6444,9 +6455,6 @@ void RGWCompleteMultipart::execute(optional_yield y)
   RGWMultiCompleteUpload *parts;
   RGWMultiXMLParser parser;
   std::unique_ptr<rgw::sal::MultipartUpload> upload;
-  off_t ofs = 0;
-  std::unique_ptr<rgw::sal::Object> meta_obj;
-  std::unique_ptr<rgw::sal::Object> target_obj;
   uint64_t olh_epoch = 0;
 
   op_ret = get_params(y);
@@ -6535,8 +6543,8 @@ void RGWCompleteMultipart::execute(optional_yield y)
   
 
   // make reservation for notification if needed
-  std::unique_ptr<rgw::sal::Notification> res
-    = driver->get_notification(meta_obj.get(), nullptr, s, rgw::notify::ObjectCreatedCompleteMultipartUpload, y, &s->object->get_name());
+  res = driver->get_notification(meta_obj.get(), nullptr, s, rgw::notify::ObjectCreatedCompleteMultipartUpload, y,
+                                 &s->object->get_name());
   op_ret = res->publish_reserve(this);
   if (op_ret < 0) {
     return;
@@ -6559,21 +6567,10 @@ void RGWCompleteMultipart::execute(optional_yield y)
     return;
   }
 
-  // remove the upload meta object ; the meta object is not versioned
-  // when the bucket is, as that would add an unneeded delete marker
-  int r = meta_obj->delete_object(this, y, true /* prevent versioning */);
-  if (r >= 0)  {
-    /* serializer's exclusive lock is released */
-    serializer->clear_locked();
-  } else {
-    ldpp_dout(this, 0) << "WARNING: failed to remove object " << meta_obj << dendl;
-  }
-
-  // send request to notification manager
-  int ret = res->publish_commit(this, ofs, upload->get_mtime(), etag, target_obj->get_instance());
-  if (ret < 0) {
-    ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
-    // too late to rollback operation, hence op_ret is not set here
+  upload_time = upload->get_mtime();
+  int r = serializer->unlock();
+  if (r < 0) {
+    ldpp_dout(this, 0) << "WARNING: failed to unlock " << *serializer.get() << dendl;
   }
 } // RGWCompleteMultipart::execute
 
@@ -6626,7 +6623,42 @@ void RGWCompleteMultipart::complete()
     }
   }
 
-  etag = s->object->get_attrs()[RGW_ATTR_ETAG].to_str();
+  if (op_ret >= 0 && target_obj.get() != nullptr) {
+    s->object->set_attrs(target_obj->get_attrs());
+    etag = s->object->get_attrs()[RGW_ATTR_ETAG].to_str();
+    // send request to notification manager
+    if (res.get() != nullptr) {
+      int ret = res->publish_commit(this, ofs, upload_time, etag, target_obj->get_instance());
+      if (ret < 0) {
+        ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
+        // too late to rollback operation, hence op_ret is not set here
+      }
+    } else {
+      ldpp_dout(this, 1) << "ERROR: reservation is null" << dendl;
+    }
+  } else {
+    ldpp_dout(this, 1) << "ERROR: either op_ret is negative (execute failed) or target_obj is null, op_ret: "
+                       << op_ret << dendl;
+  }
+
+  // remove the upload meta object ; the meta object is not versioned
+  // when the bucket is, as that would add an unneeded delete marker
+  // moved to complete to prevent segmentation fault in publish commit
+  if (meta_obj.get() != nullptr) {
+    int ret = meta_obj->delete_object(this, null_yield, rgw::sal::FLAG_PREVENT_VERSIONING | rgw::sal::FLAG_LOG_OP);
+    if (ret >= 0) {
+      /* serializer's exclusive lock is released */
+      serializer->clear_locked();
+    } else {
+      ldpp_dout(this, 0) << "WARNING: failed to remove object " << meta_obj << ", ret: " << ret << dendl;
+    }
+  } else {
+    ldpp_dout(this, 0) << "WARNING: meta_obj is null" << dendl;
+  }
+
+  res.reset();
+  meta_obj.reset();
+  target_obj.reset();
 
   send_response();
 }
@@ -7088,7 +7120,7 @@ void RGWDeleteMultiObj::handle_individual_object(const rgw_obj_key& o, optional_
   del_op->params.bucket_owner = s->bucket_owner;
   del_op->params.marker_version_id = version_id;
 
-  op_ret = del_op->delete_obj(this, y);
+  op_ret = del_op->delete_obj(this, y, rgw::sal::FLAG_LOG_OP);
   if (op_ret == -ENOENT) {
     op_ret = 0;
   }
@@ -7264,7 +7296,7 @@ bool RGWBulkDelete::Deleter::delete_single(const acct_path_t& path, optional_yie
     del_op->params.obj_owner = bowner;
     del_op->params.bucket_owner = bucket_owner;
 
-    ret = del_op->delete_obj(dpp, y);
+    ret = del_op->delete_obj(dpp, y, rgw::sal::FLAG_LOG_OP);
     if (ret < 0) {
       goto delop_fail;
     }
@@ -7754,7 +7786,7 @@ int RGWBulkUploadOp::handle_file(const std::string_view path,
   op_ret = processor->complete(size, etag, nullptr, ceph::real_time(),
                               attrs, ceph::real_time() /* delete_at */,
                               nullptr, nullptr, nullptr, nullptr, nullptr,
-                              s->yield);
+                              s->yield, rgw::sal::FLAG_LOG_OP);
   if (op_ret < 0) {
     ldpp_dout(this, 20) << "processor::complete returned op_ret=" << op_ret << dendl;
   }
diff --git a/src/rgw/rgw_op.h b/src/rgw/rgw_op.h
index f398b5b15..2bffcddcc 100644
--- a/src/rgw/rgw_op.h
+++ b/src/rgw/rgw_op.h
@@ -1876,6 +1876,11 @@ protected:
   bufferlist data;
   std::unique_ptr<rgw::sal::MPSerializer> serializer;
   jspan multipart_trace;
+  ceph::real_time upload_time;
+  std::unique_ptr<rgw::sal::Object> target_obj;
+  std::unique_ptr<rgw::sal::Notification> res;
+  std::unique_ptr<rgw::sal::Object> meta_obj;
+  off_t ofs = 0;
 
 public:
   RGWCompleteMultipart() {}
diff --git a/src/rgw/rgw_rest_s3.cc b/src/rgw/rgw_rest_s3.cc
index 0b997f30b..b28a563bc 100644
--- a/src/rgw/rgw_rest_s3.cc
+++ b/src/rgw/rgw_rest_s3.cc
@@ -563,6 +563,20 @@ done:
   }
 
   if (op_ret == -ERR_NOT_MODIFIED) {
+      dump_last_modified(s, lastmod);
+
+      auto iter = attrs.find(RGW_ATTR_ETAG);
+      if (iter != attrs.end())
+        dump_etag(s, iter->second.to_str());
+
+      iter = attrs.find(RGW_ATTR_CACHE_CONTROL);
+      if (iter != attrs.end())
+        dump_header(s, rgw_to_http_attrs[RGW_ATTR_CACHE_CONTROL], iter->second);
+
+      iter = attrs.find(RGW_ATTR_EXPIRES);
+      if (iter != attrs.end())
+        dump_header(s, rgw_to_http_attrs[RGW_ATTR_EXPIRES], iter->second);
+
       end_header(s, this);
   } else {
       if (!content_type)
@@ -5192,8 +5206,9 @@ void parse_post_action(const std::string& post_body, req_state* s)
           if (boost::starts_with(key, "Attributes.")) {
             update_attribute_map(t, map);
           } else {
+            constexpr bool in_query = true; // replace '+' with ' '
             s->info.args.append(t.substr(0, pos),
-                              url_decode(t.substr(pos+1, t.size() -1)));
+                              url_decode(t.substr(pos+1, t.size() -1), in_query));
           }
         }
       }
@@ -6239,7 +6254,7 @@ rgw::auth::s3::LocalEngine::authenticate(
   if (driver->get_user_by_access_key(dpp, access_key_id, y, &user) < 0) {
       ldpp_dout(dpp, 5) << "error reading user info, uid=" << access_key_id
               << " can't authenticate" << dendl;
-      return result_t::deny(-ERR_INVALID_ACCESS_KEY);
+      return result_t::reject(-ERR_INVALID_ACCESS_KEY);
   }
   //TODO: Uncomment, when we have a migration plan in place.
   /*else {
@@ -6253,7 +6268,7 @@ rgw::auth::s3::LocalEngine::authenticate(
   const auto iter = user->get_info().access_keys.find(access_key_id);
   if (iter == std::end(user->get_info().access_keys)) {
     ldpp_dout(dpp, 0) << "ERROR: access key not encoded in user info" << dendl;
-    return result_t::deny(-EPERM);
+    return result_t::reject(-EPERM);
   }
   const RGWAccessKey& k = iter->second;
 
@@ -6269,7 +6284,7 @@ rgw::auth::s3::LocalEngine::authenticate(
   ldpp_dout(dpp, 15) << "compare=" << compare << dendl;
 
   if (compare != 0) {
-    return result_t::deny(-ERR_SIGNATURE_NO_MATCH);
+    return result_t::reject(-ERR_SIGNATURE_NO_MATCH);
   }
 
   auto apl = apl_factory->create_apl_local(cct, s, user->get_info(),
diff --git a/src/rgw/rgw_rest_swift.cc b/src/rgw/rgw_rest_swift.cc
index ee943ea44..f5deaa92b 100644
--- a/src/rgw/rgw_rest_swift.cc
+++ b/src/rgw/rgw_rest_swift.cc
@@ -2473,6 +2473,7 @@ RGWOp* RGWSwiftWebsiteHandler::get_ws_index_op()
   } else {
     s->object->set_name(s->bucket->get_info().website_conf.get_index_doc());
   }
+  s->object->set_bucket(s->bucket.get());
 
   auto getop = new RGWGetObj_ObjStore_SWIFT;
   getop->set_get_data(boost::algorithm::equals("GET", s->info.method));
diff --git a/src/rgw/rgw_rest_user_policy.cc b/src/rgw/rgw_rest_user_policy.cc
index 2e300468b..2995a43ad 100644
--- a/src/rgw/rgw_rest_user_policy.cc
+++ b/src/rgw/rgw_rest_user_policy.cc
@@ -91,9 +91,9 @@ uint64_t RGWPutUserPolicy::get_op()
 
 int RGWPutUserPolicy::get_params()
 {
-  policy_name = url_decode(s->info.args.get("PolicyName"), true);
-  user_name = url_decode(s->info.args.get("UserName"), true);
-  policy = url_decode(s->info.args.get("PolicyDocument"), true);
+  policy_name = s->info.args.get("PolicyName");
+  user_name = s->info.args.get("UserName");
+  policy = s->info.args.get("PolicyDocument");
 
   if (policy_name.empty() || user_name.empty() || policy.empty()) {
     ldpp_dout(this, 20) << "ERROR: one of policy name, user name or policy document is empty"
diff --git a/src/rgw/rgw_s3select.cc b/src/rgw/rgw_s3select.cc
index c7eaa6984..4080d0938 100644
--- a/src/rgw/rgw_s3select.cc
+++ b/src/rgw/rgw_s3select.cc
@@ -408,7 +408,7 @@ int RGWSelectObj_ObjStore_S3::run_s3select_on_csv(const char* query, const char*
   } else if(m_header_info.compare("USE")==0) {
     csv.use_header_info=true;
   }
-  //m_s3_csv_object.set_external_debug_system(fp_debug_mesg);
+  m_s3_csv_object.set_external_debug_system(fp_debug_mesg);
   m_s3_csv_object.set_result_formatters(fp_s3select_result_format,fp_result_header_format);
   m_s3_csv_object.set_csv_query(&s3select_syntax, csv);
   if (s3select_syntax.get_error_description().empty() == false) {
@@ -527,7 +527,8 @@ int RGWSelectObj_ObjStore_S3::run_s3select_on_json(const char* query, const char
   }
     
   //initializing json processor
-  m_s3_json_object.set_json_query(&s3select_syntax);
+  json_object::csv_definitions output_definition;
+  m_s3_json_object.set_json_query(&s3select_syntax,output_definition);
 
   if (input == nullptr) {
     input = "";
@@ -862,20 +863,23 @@ int RGWSelectObj_ObjStore_S3::csv_processing(bufferlist& bl, off_t ofs, off_t le
         continue;
       }
 
-      if((ofs + len) > it.length()){
+
+      if(ofs > it.length()){
+      //safety check
 	ldpp_dout(this, 10) << "offset and length may cause invalid read: ofs = " << ofs << " len = " << len << " it.length() = " << it.length() << dendl;
 	ofs = 0;
-	len = it.length();
       }
 
     if(m_is_trino_request){
+      //TODO replace len with it.length() ? ; test Trino flow with compressed objects.
+      //is it possible to send get-by-ranges? in parallel?
       shape_chunk_per_trino_requests(&(it)[0], ofs, len); 
     }
 
     ldpp_dout(this, 10) << "s3select: chunk:  ofs = " << ofs << " len = " << len << " it.length() = " << it.length() << " m_object_size_for_processing = " << m_object_size_for_processing << dendl;
     
     m_aws_response_handler.update_processed_size(it.length());//NOTE : to run analysis to validate len is aligned with m_processed_bytes
-    status = run_s3select_on_csv(m_sql_query.c_str(), &(it)[0] + ofs, len);
+    status = run_s3select_on_csv(m_sql_query.c_str(), &(it)[0] + ofs, it.length());
     if (status<0) {
 	  return -EINVAL;
     }
diff --git a/src/rgw/rgw_sal.h b/src/rgw/rgw_sal.h
index 944737dee..50fe30c6b 100644
--- a/src/rgw/rgw_sal.h
+++ b/src/rgw/rgw_sal.h
@@ -210,6 +210,9 @@ enum AttrsMod {
   ATTRSMOD_MERGE   = 2
 };
 
+static constexpr uint32_t FLAG_LOG_OP = 0x0001;
+static constexpr uint32_t FLAG_PREVENT_VERSIONING = 0x0002;
+
 // a simple streaming data processing abstraction
 /**
  * @brief A simple streaming data processing abstraction
@@ -243,7 +246,8 @@ class ObjectProcessor : public DataProcessor {
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y) = 0;
+                       optional_yield y,
+                       uint32_t flags) = 0;
 };
 
 /** Base class for AIO completions */
@@ -947,7 +951,7 @@ class Object {
       virtual ~DeleteOp() = default;
 
       /** Delete the object */
-      virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y) = 0;
+      virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags) = 0;
     };
 
     Object() {}
@@ -956,7 +960,7 @@ class Object {
     /** Shortcut synchronous delete call for common deletes */
     virtual int delete_object(const DoutPrefixProvider* dpp,
 			      optional_yield y,
-			      bool prevent_versioning = false) = 0;
+			      uint32_t flags) = 0;
     /** Asynchronous delete call */
     virtual int delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate, Completions* aio,
 			       bool keep_index_consistent, optional_yield y) = 0;
@@ -1025,7 +1029,8 @@ class Object {
 			   const real_time& mtime,
 			   uint64_t olh_epoch,
 			   const DoutPrefixProvider* dpp,
-			   optional_yield y) = 0;
+			   optional_yield y,
+                           uint32_t flags) = 0;
     /** Move an object to the cloud */
     virtual int transition_to_cloud(Bucket* bucket,
 			   rgw::sal::PlacementTier* tier,
@@ -1436,7 +1441,8 @@ public:
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y) = 0;
+                       optional_yield y,
+                       uint32_t flags) = 0;
 };
 
 
diff --git a/src/rgw/rgw_sal_daos.cc b/src/rgw/rgw_sal_daos.cc
index 4b0234b1f..00e9157ac 100644
--- a/src/rgw/rgw_sal_daos.cc
+++ b/src/rgw/rgw_sal_daos.cc
@@ -1052,7 +1052,8 @@ std::unique_ptr<MPSerializer> DaosObject::get_serializer(
 int DaosObject::transition(Bucket* bucket,
                            const rgw_placement_rule& placement_rule,
                            const real_time& mtime, uint64_t olh_epoch,
-                           const DoutPrefixProvider* dpp, optional_yield y) {
+                           const DoutPrefixProvider* dpp, optional_yield y,
+                           uint32_t flags) {
   return DAOS_NOT_IMPLEMENTED_LOG(dpp);
 }
 
@@ -1202,7 +1203,7 @@ DaosObject::DaosDeleteOp::DaosDeleteOp(DaosObject* _source) : source(_source) {}
 // 3. Handle empty directories
 // 4. Fail when file doesn't exist
 int DaosObject::DaosDeleteOp::delete_obj(const DoutPrefixProvider* dpp,
-                                         optional_yield y) {
+                                         optional_yield y, uint32_t flags) {
   ldpp_dout(dpp, 20) << "DaosDeleteOp::delete_obj "
                      << source->get_key().get_oid() << " from "
                      << source->get_bucket()->get_name() << dendl;
@@ -1231,13 +1232,13 @@ int DaosObject::DaosDeleteOp::delete_obj(const DoutPrefixProvider* dpp,
 }
 
 int DaosObject::delete_object(const DoutPrefixProvider* dpp, optional_yield y,
-                              bool prevent_versioning) {
+                              uint32_t flags) {
   ldpp_dout(dpp, 20) << "DEBUG: delete_object" << dendl;
   DaosObject::DaosDeleteOp del_op(this);
   del_op.params.bucket_owner = bucket->get_info().owner;
   del_op.params.versioning_status = bucket->get_info().versioning_status();
 
-  return del_op.delete_obj(dpp, y);
+  return del_op.delete_obj(dpp, y, flags);
 }
 
 int DaosObject::delete_obj_aio(const DoutPrefixProvider* dpp,
@@ -1548,7 +1549,7 @@ int DaosAtomicWriter::complete(
     ceph::real_time set_mtime, std::map<std::string, bufferlist>& attrs,
     ceph::real_time delete_at, const char* if_match, const char* if_nomatch,
     const std::string* user_data, rgw_zone_set* zones_trace, bool* canceled,
-    optional_yield y) {
+    optional_yield y, uint32_t flags) {
   ldpp_dout(dpp, 20) << "DEBUG: complete" << dendl;
   bufferlist bl;
   rgw_bucket_dir_entry ent;
@@ -1736,7 +1737,7 @@ int DaosMultipartUpload::complete(
   int marker = 0;
   uint64_t min_part_size = cct->_conf->rgw_multipart_min_part_size;
   auto etags_iter = part_etags.begin();
-  rgw::sal::Attrs attrs = target_obj->get_attrs();
+  rgw::sal::Attrs& attrs = target_obj->get_attrs();
 
   do {
     ldpp_dout(dpp, 20) << "DaosMultipartUpload::complete(): list_parts()"
@@ -2072,7 +2073,7 @@ int DaosMultipartWriter::complete(
     ceph::real_time set_mtime, std::map<std::string, bufferlist>& attrs,
     ceph::real_time delete_at, const char* if_match, const char* if_nomatch,
     const std::string* user_data, rgw_zone_set* zones_trace, bool* canceled,
-    optional_yield y) {
+    optional_yield y, uint32_t flags) {
   ldpp_dout(dpp, 20) << "DaosMultipartWriter::complete(): enter part="
                      << part_num_str << dendl;
 
diff --git a/src/rgw/rgw_sal_daos.h b/src/rgw/rgw_sal_daos.h
index 64bf49c7c..af3063a60 100644
--- a/src/rgw/rgw_sal_daos.h
+++ b/src/rgw/rgw_sal_daos.h
@@ -589,7 +589,7 @@ class DaosObject : public StoreObject {
     DaosDeleteOp(DaosObject* _source);
 
     virtual int delete_obj(const DoutPrefixProvider* dpp,
-                           optional_yield y) override;
+                           optional_yield y, uint32_t flags) override;
   };
 
   ds3_obj_t* ds3o = nullptr;
@@ -606,7 +606,7 @@ class DaosObject : public StoreObject {
   virtual ~DaosObject();
 
   virtual int delete_object(const DoutPrefixProvider* dpp, optional_yield y,
-                            bool prevent_versioning = false) override;
+                            uint32_t flags) override;
   virtual int delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate,
                              Completions* aio, bool keep_index_consistent,
                              optional_yield y) override;
@@ -652,7 +652,8 @@ class DaosObject : public StoreObject {
                          const rgw_placement_rule& placement_rule,
                          const real_time& mtime, uint64_t olh_epoch,
                          const DoutPrefixProvider* dpp,
-                         optional_yield y) override;
+                         optional_yield y,
+                         uint32_t flags) override;
   virtual int transition_to_cloud(Bucket* bucket, rgw::sal::PlacementTier* tier,
                                   rgw_bucket_dir_entry& o,
                                   std::set<std::string>& cloud_targets,
@@ -763,7 +764,8 @@ class DaosAtomicWriter : public StoreWriter {
                        ceph::real_time delete_at, const char* if_match,
                        const char* if_nomatch, const std::string* user_data,
                        rgw_zone_set* zones_trace, bool* canceled,
-                       optional_yield y) override;
+                       optional_yield y,
+                       uint32_t flags) override;
 };
 
 class DaosMultipartWriter : public StoreWriter {
@@ -808,7 +810,8 @@ class DaosMultipartWriter : public StoreWriter {
                        ceph::real_time delete_at, const char* if_match,
                        const char* if_nomatch, const std::string* user_data,
                        rgw_zone_set* zones_trace, bool* canceled,
-                       optional_yield y) override;
+                       optional_yield y,
+                       uint32_t flags) override;
 
   const std::string& get_bucket_name();
 };
diff --git a/src/rgw/rgw_sal_dbstore.cc b/src/rgw/rgw_sal_dbstore.cc
index 5100dc41e..ca1b3f333 100644
--- a/src/rgw/rgw_sal_dbstore.cc
+++ b/src/rgw/rgw_sal_dbstore.cc
@@ -734,7 +734,8 @@ namespace rgw::sal {
       const real_time& mtime,
       uint64_t olh_epoch,
       const DoutPrefixProvider* dpp,
-      optional_yield y)
+      optional_yield y,
+      uint32_t flags)
   {
     DB::Object op_target(store->getDB(),
         get_bucket()->get_info(), get_obj());
@@ -815,7 +816,7 @@ namespace rgw::sal {
     parent_op(&op_target)
   { }
 
-  int DBObject::DBDeleteOp::delete_obj(const DoutPrefixProvider* dpp, optional_yield y)
+  int DBObject::DBDeleteOp::delete_obj(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags)
   {
     parent_op.params.bucket_owner = params.bucket_owner.get_id();
     parent_op.params.versioning_status = params.versioning_status;
@@ -842,7 +843,7 @@ namespace rgw::sal {
     return ret;
   }
 
-  int DBObject::delete_object(const DoutPrefixProvider* dpp, optional_yield y, bool prevent_versioning)
+  int DBObject::delete_object(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags)
   {
     DB::Object del_target(store->getDB(), bucket->get_info(), get_obj());
     DB::Object::Delete del_op(&del_target);
@@ -923,7 +924,7 @@ namespace rgw::sal {
     // Since the data objects are associated with meta obj till
     // MultipartUpload::Complete() is done, removing the metadata obj
     // should remove all the uploads so far.
-    ret = del_op->delete_obj(dpp, null_yield);
+    ret = del_op->delete_obj(dpp, null_yield, 0);
     if (ret < 0) {
       ldpp_dout(dpp, 20) << __func__ << ": del_op.delete_obj returned " <<
         ret << dendl;
@@ -1056,7 +1057,7 @@ namespace rgw::sal {
     int marker = 0;
     uint64_t min_part_size = cct->_conf->rgw_multipart_min_part_size;
     auto etags_iter = part_etags.begin();
-    rgw::sal::Attrs attrs = target_obj->get_attrs();
+    rgw::sal::Attrs& attrs = target_obj->get_attrs();
 
     ofs = 0;
     accounted_size = 0;
@@ -1350,7 +1351,8 @@ namespace rgw::sal {
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y)
+                       optional_yield y,
+                       uint32_t flags)
   {
     int ret = 0;
     /* XXX: same as AtomicWriter..consolidate code */
@@ -1508,7 +1510,8 @@ namespace rgw::sal {
                          const char *if_match, const char *if_nomatch,
                          const std::string *user_data,
                          rgw_zone_set *zones_trace, bool *canceled,
-                         optional_yield y)
+                         optional_yield y,
+                         uint32_t flags)
   {
     parent_op.meta.mtime = mtime;
     parent_op.meta.delete_at = delete_at;
diff --git a/src/rgw/rgw_sal_dbstore.h b/src/rgw/rgw_sal_dbstore.h
index 3acdb4ba3..2d65a7f3f 100644
--- a/src/rgw/rgw_sal_dbstore.h
+++ b/src/rgw/rgw_sal_dbstore.h
@@ -567,7 +567,7 @@ protected:
         public:
           DBDeleteOp(DBObject* _source);
 
-          virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y) override;
+          virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags) override;
       };
 
       DBObject() = default;
@@ -586,7 +586,7 @@ protected:
 
       virtual int delete_object(const DoutPrefixProvider* dpp,
           optional_yield y,
-          bool prevent_versioning = false) override;
+          uint32_t flags) override;
       virtual int delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate, Completions* aio,
           bool keep_index_consistent, optional_yield y) override;
       virtual int copy_object(User* user,
@@ -624,7 +624,8 @@ protected:
           const real_time& mtime,
           uint64_t olh_epoch,
           const DoutPrefixProvider* dpp,
-          optional_yield y) override;
+          optional_yield y,
+          uint32_t flags) override;
       virtual bool placement_rules_match(rgw_placement_rule& r1, rgw_placement_rule& r2) override;
       virtual int dump_obj_layout(const DoutPrefixProvider *dpp, optional_yield y, Formatter* f) override;
 
@@ -705,7 +706,8 @@ protected:
                          const char *if_match, const char *if_nomatch,
                          const std::string *user_data,
                          rgw_zone_set *zones_trace, bool *canceled,
-                         optional_yield y) override;
+                         optional_yield y,
+                         uint32_t flags) override;
   };
 
   class DBMultipartWriter : public StoreWriter {
@@ -753,7 +755,8 @@ public:
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y) override;
+                       optional_yield y,
+                       uint32_t flags) override;
   };
 
   class DBStore : public StoreDriver {
diff --git a/src/rgw/rgw_sal_filter.cc b/src/rgw/rgw_sal_filter.cc
index 2a48cec9c..8cf9db111 100644
--- a/src/rgw/rgw_sal_filter.cc
+++ b/src/rgw/rgw_sal_filter.cc
@@ -852,9 +852,9 @@ int FilterBucket::abort_multiparts(const DoutPrefixProvider* dpp, CephContext* c
 
 int FilterObject::delete_object(const DoutPrefixProvider* dpp,
 				optional_yield y,
-				bool prevent_versioning)
+				uint32_t flags)
 {
-  return next->delete_object(dpp, y, prevent_versioning);
+  return next->delete_object(dpp, y, flags);
 }
 
 int FilterObject::delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate,
@@ -960,10 +960,11 @@ int FilterObject::transition(Bucket* bucket,
 			     const real_time& mtime,
 			     uint64_t olh_epoch,
 			     const DoutPrefixProvider* dpp,
-			     optional_yield y)
+			     optional_yield y,
+                             uint32_t flags)
 {
   return next->transition(nextBucket(bucket), placement_rule, mtime, olh_epoch,
-			  dpp, y);
+			  dpp, y, flags);
 }
 
 int FilterObject::transition_to_cloud(Bucket* bucket,
@@ -1091,11 +1092,11 @@ int FilterObject::FilterReadOp::iterate(const DoutPrefixProvider* dpp, int64_t o
 }
 
 int FilterObject::FilterDeleteOp::delete_obj(const DoutPrefixProvider* dpp,
-					   optional_yield y)
+					   optional_yield y, uint32_t flags)
 {
   /* Copy params into next */
   next->params = params;
-  int ret = next->delete_obj(dpp, y);
+  int ret = next->delete_obj(dpp, y, flags);
   /* Copy result back */
   result = next->result;
   return ret;
@@ -1313,11 +1314,12 @@ int FilterWriter::complete(size_t accounted_size, const std::string& etag,
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y)
+                       optional_yield y,
+                       uint32_t flags)
 {
   return next->complete(accounted_size, etag, mtime, set_mtime, attrs,
 			delete_at, if_match, if_nomatch, user_data, zones_trace,
-			canceled, y);
+			canceled, y, flags);
 }
 
 int FilterLuaManager::get_script(const DoutPrefixProvider* dpp, optional_yield y,
diff --git a/src/rgw/rgw_sal_filter.h b/src/rgw/rgw_sal_filter.h
index 951a1de5f..fdcc9c6f1 100644
--- a/src/rgw/rgw_sal_filter.h
+++ b/src/rgw/rgw_sal_filter.h
@@ -575,7 +575,7 @@ public:
     FilterDeleteOp(std::unique_ptr<DeleteOp> _next) : next(std::move(_next)) {}
     virtual ~FilterDeleteOp() = default;
 
-    virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y) override;
+    virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags) override;
   };
 
   FilterObject(std::unique_ptr<Object> _next) : next(std::move(_next)) {}
@@ -589,7 +589,7 @@ public:
 
   virtual int delete_object(const DoutPrefixProvider* dpp,
 			    optional_yield y,
-			    bool prevent_versioning = false) override;
+			    uint32_t flags) override;
   virtual int delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate,
 			     Completions* aio,
 			     bool keep_index_consistent, optional_yield y) override;
@@ -639,7 +639,8 @@ public:
 			 const real_time& mtime,
 			 uint64_t olh_epoch,
 			 const DoutPrefixProvider* dpp,
-			 optional_yield y) override;
+			 optional_yield y,
+                         uint32_t flags) override;
   virtual int transition_to_cloud(Bucket* bucket,
 				  rgw::sal::PlacementTier* tier,
 				  rgw_bucket_dir_entry& o,
@@ -899,7 +900,8 @@ public:
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y) override;
+                       optional_yield y,
+                       uint32_t flags) override;
 };
 
 class FilterLuaManager : public LuaManager {
diff --git a/src/rgw/rgw_sal_motr.cc b/src/rgw/rgw_sal_motr.cc
index de18ba944..1ae7565f0 100644
--- a/src/rgw/rgw_sal_motr.cc
+++ b/src/rgw/rgw_sal_motr.cc
@@ -1336,7 +1336,8 @@ int MotrObject::transition(Bucket* bucket,
     const real_time& mtime,
     uint64_t olh_epoch,
     const DoutPrefixProvider* dpp,
-    optional_yield y)
+    optional_yield y,
+    uint32_t flags)
 {
   return 0;
 }
@@ -1498,7 +1499,7 @@ MotrObject::MotrDeleteOp::MotrDeleteOp(MotrObject *_source) :
 // Delete::delete_obj() in rgw_rados.cc shows how rados backend process the
 // params.
 // 2. Delete an object when its versioning is turned on.
-int MotrObject::MotrDeleteOp::delete_obj(const DoutPrefixProvider* dpp, optional_yield y)
+int MotrObject::MotrDeleteOp::delete_obj(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags)
 {
   ldpp_dout(dpp, 20) << "delete " << source->get_key().get_oid() << " from " << source->get_bucket()->get_name() << dendl;
 
@@ -1542,13 +1543,13 @@ int MotrObject::MotrDeleteOp::delete_obj(const DoutPrefixProvider* dpp, optional
   return 0;
 }
 
-int MotrObject::delete_object(const DoutPrefixProvider* dpp, optional_yield y, bool prevent_versioning)
+int MotrObject::delete_object(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags)
 {
   MotrObject::MotrDeleteOp del_op(this);
   del_op.params.bucket_owner = bucket->get_info().owner;
   del_op.params.versioning_status = bucket->get_info().versioning_status();
 
-  return del_op.delete_obj(dpp, y);
+  return del_op.delete_obj(dpp, y, flags);
 }
 
 int MotrObject::delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate,
@@ -2374,7 +2375,8 @@ int MotrAtomicWriter::complete(size_t accounted_size, const std::string& etag,
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y)
+                       optional_yield y,
+                       uint32_t flags)
 {
   int rc = 0;
 
@@ -2733,7 +2735,7 @@ int MotrMultipartUpload::complete(const DoutPrefixProvider *dpp,
   int marker = 0;
   uint64_t min_part_size = cct->_conf->rgw_multipart_min_part_size;
   auto etags_iter = part_etags.begin();
-  rgw::sal::Attrs attrs = target_obj->get_attrs();
+  rgw::sal::Attrs& attrs = target_obj->get_attrs();
 
   do {
     ldpp_dout(dpp, 20) << "MotrMultipartUpload::complete(): list_parts()" << dendl;
@@ -3031,7 +3033,8 @@ int MotrMultipartWriter::complete(size_t accounted_size, const std::string& etag
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y)
+                       optional_yield y,
+                       uint32_t flags)
 {
   // Should the dir entry(object metadata) be updated? For example
   // mtime.
diff --git a/src/rgw/rgw_sal_motr.h b/src/rgw/rgw_sal_motr.h
index b7230f7e1..c09c95ede 100644
--- a/src/rgw/rgw_sal_motr.h
+++ b/src/rgw/rgw_sal_motr.h
@@ -660,7 +660,7 @@ class MotrObject : public StoreObject {
       public:
         MotrDeleteOp(MotrObject* _source);
 
-        virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y) override;
+        virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y, uint32_t flags) override;
     };
 
     MotrObject() = default;
@@ -676,7 +676,7 @@ class MotrObject : public StoreObject {
 
     virtual int delete_object(const DoutPrefixProvider* dpp,
         optional_yield y,
-        bool prevent_versioning = false) override;
+        uint32_t flags) override;
     virtual int delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate, Completions* aio,
         bool keep_index_consistent, optional_yield y) override;
     virtual int copy_object(User* user,
@@ -712,7 +712,8 @@ class MotrObject : public StoreObject {
         const real_time& mtime,
         uint64_t olh_epoch,
         const DoutPrefixProvider* dpp,
-        optional_yield y) override;
+        optional_yield y,
+        uint32_t flags) override;
     virtual bool placement_rules_match(rgw_placement_rule& r1, rgw_placement_rule& r2) override;
     virtual int dump_obj_layout(const DoutPrefixProvider *dpp, optional_yield y, Formatter* f) override;
 
@@ -819,7 +820,8 @@ class MotrAtomicWriter : public StoreWriter {
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y) override;
+                       optional_yield y,
+                       uint32_t flags) override;
 
   unsigned populate_bvec(unsigned len, bufferlist::iterator &bi);
   void cleanup();
@@ -866,7 +868,8 @@ public:
                        const char *if_match, const char *if_nomatch,
                        const std::string *user_data,
                        rgw_zone_set *zones_trace, bool *canceled,
-                       optional_yield y) override;
+                       optional_yield y,
+                       uint32_t flags) override;
 };
 
 // The implementation of multipart upload in POC roughly follows the
diff --git a/src/rgw/rgw_sts.cc b/src/rgw/rgw_sts.cc
index b55283442..2b31d5c5a 100644
--- a/src/rgw/rgw_sts.cc
+++ b/src/rgw/rgw_sts.cc
@@ -54,7 +54,7 @@ int Credentials::generateCredentials(const DoutPrefixProvider *dpp,
                           rgw::auth::Identity* identity)
 {
   uuid_d accessKey, secretKey;
-  char accessKeyId_str[MAX_ACCESS_KEY_LEN], secretAccessKey_str[MAX_SECRET_KEY_LEN];
+  char accessKeyId_str[MAX_ACCESS_KEY_LEN + 1], secretAccessKey_str[MAX_SECRET_KEY_LEN + 1];
 
   //AccessKeyId
   gen_rand_alphanumeric_plain(cct, accessKeyId_str, sizeof(accessKeyId_str));
diff --git a/src/rgw/services/svc_sys_obj_cache.cc b/src/rgw/services/svc_sys_obj_cache.cc
index d1b7a3dbb..4c3603d1c 100644
--- a/src/rgw/services/svc_sys_obj_cache.cc
+++ b/src/rgw/services/svc_sys_obj_cache.cc
@@ -90,6 +90,11 @@ int RGWSI_SysObj_Cache::remove(const DoutPrefixProvider *dpp,
                                optional_yield y)
 
 {
+  int r = RGWSI_SysObj_Core::remove(dpp, objv_tracker, obj, y);
+  if (r < 0) {
+    return r;
+  }
+
   rgw_pool pool;
   string oid;
   normalize_pool_and_obj(obj.pool, obj.oid, pool, oid);
@@ -98,12 +103,12 @@ int RGWSI_SysObj_Cache::remove(const DoutPrefixProvider *dpp,
   cache.invalidate_remove(dpp, name);
 
   ObjectCacheInfo info;
-  int r = distribute_cache(dpp, name, obj, info, INVALIDATE_OBJ, y);
+  r = distribute_cache(dpp, name, obj, info, INVALIDATE_OBJ, y);
   if (r < 0) {
     ldpp_dout(dpp, 0) << "ERROR: " << __func__ << "(): failed to distribute cache: r=" << r << dendl;
-  }
+  } // not fatal
 
-  return RGWSI_SysObj_Core::remove(dpp, objv_tracker, obj, y);
+  return 0;
 }
 
 int RGWSI_SysObj_Cache::read(const DoutPrefixProvider *dpp,