// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab ft=cpp

/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2021 Red Hat, Inc.
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software
 * Foundation. See file COPYING.
 *
 */

#include <errno.h>
#include <stdlib.h>
#include <system_error>
#include <unistd.h>
#include <sstream>

#include "common/Clock.h"
#include "common/errno.h"

#include "rgw_sal.h"
#include "rgw_sal_dbstore.h"
#include "rgw_bucket.h"

#define dout_subsys ceph_subsys_rgw

using namespace std;

namespace rgw::sal {

  int DBUser::list_buckets(const DoutPrefixProvider *dpp, const string& marker,
      const string& end_marker, uint64_t max, bool need_stats,
      BucketList &buckets, optional_yield y)
  {
    RGWUserBuckets ulist;
    bool is_truncated = false;
    int ret;

    buckets.clear();
    ret = store->getDB()->list_buckets(dpp, "", info.user_id, marker, end_marker, max,
        need_stats, &ulist, &is_truncated);
    if (ret < 0)
      return ret;

    buckets.set_truncated(is_truncated);
    for (const auto& ent : ulist.get_buckets()) {
      buckets.add(std::make_unique<DBBucket>(this->store, ent.second, this));
    }

    return 0;
  }

  int DBUser::create_bucket(const DoutPrefixProvider *dpp,
      const rgw_bucket& b,
      const string& zonegroup_id,
      rgw_placement_rule& placement_rule,
      string& swift_ver_location,
      const RGWQuotaInfo * pquota_info,
      const RGWAccessControlPolicy& policy,
      Attrs& attrs,
      RGWBucketInfo& info,
      obj_version& ep_objv,
      bool exclusive,
      bool obj_lock_enabled,
      bool *existed,
      req_info& req_info,
      std::unique_ptr<Bucket>* bucket_out,
      optional_yield y)
  {
    int ret;
    bufferlist in_data;
    RGWBucketInfo master_info;
    rgw_bucket *pmaster_bucket = nullptr;
    uint32_t *pmaster_num_shards = nullptr;
    real_time creation_time;
    std::unique_ptr<Bucket> bucket;
    obj_version objv, *pobjv = NULL;

    /* If it exists, look it up; otherwise create it */
    ret = store->get_bucket(dpp, this, b, &bucket, y);
    if (ret < 0 && ret != -ENOENT)
      return ret;

    if (ret != -ENOENT) {
      RGWAccessControlPolicy old_policy(store->ctx());
      *existed = true;
      if (swift_ver_location.empty()) {
        swift_ver_location = bucket->get_info().swift_ver_location;
      }
      placement_rule.inherit_from(bucket->get_info().placement_rule);

      // don't allow changes to the acl policy
      /*    int r = rgw_op_get_bucket_policy_from_attr(dpp, this, this, bucket->get_attrs(),
            &old_policy, y);
            if (r >= 0 && old_policy != policy) {
            bucket_out->swap(bucket);
            return -EEXIST;
            }*/
    } else {
      bucket = std::make_unique<DBBucket>(store, b, this);
      *existed = false;
      bucket->set_attrs(attrs);
      // XXX: For now single default zone and STANDARD storage class
      // supported.
      placement_rule.name = "default";
      placement_rule.storage_class = "STANDARD";
    }

    /*
     * XXX: If not master zone, fwd the request to master zone.
     * For now DBStore has single zone.
     */
    std::string zid = zonegroup_id;
    /* if (zid.empty()) {
       zid = svc()->zone->get_zonegroup().get_id();
       } */

    if (*existed) {
      rgw_placement_rule selected_placement_rule;
      /* XXX: Handle this when zone is implemented
         ret = svc()->zone->select_bucket_placement(this.get_info(),
         zid, placement_rule,
         &selected_placement_rule, nullptr, y);
         if (selected_placement_rule != info.placement_rule) {
         ret = -EEXIST;
         bucket_out->swap(bucket);
         return ret;
         } */
    } else {

      /* XXX: We may not need to send all these params. Cleanup the unused ones */
      ret = store->getDB()->create_bucket(dpp, this->get_info(), bucket->get_key(),
          zid, placement_rule, swift_ver_location, pquota_info,
          attrs, info, pobjv, &ep_objv, creation_time,
          pmaster_bucket, pmaster_num_shards, y, exclusive);
      if (ret == -EEXIST) {
        *existed = true;
        ret = 0;
      } else if (ret != 0) {
        return ret;
      }
    }

    bucket->set_version(ep_objv);
    bucket->get_info() = info;

    bucket_out->swap(bucket);

    return ret;
  }

  int DBUser::read_attrs(const DoutPrefixProvider* dpp, optional_yield y)
  {
    int ret;
    ret = store->getDB()->get_user(dpp, string("user_id"), get_id().id, info, &attrs,
        &objv_tracker);
    return ret;
  }

  int DBUser::read_stats(const DoutPrefixProvider *dpp,
      optional_yield y, RGWStorageStats* stats,
      ceph::real_time *last_stats_sync,
      ceph::real_time *last_stats_update)
  {
    return 0;
  }

  /* stats - Not for first pass */
  int DBUser::read_stats_async(const DoutPrefixProvider *dpp, RGWGetUserStats_CB *cb)
  {
    return 0;
  }

  int DBUser::complete_flush_stats(const DoutPrefixProvider *dpp, optional_yield y)
  {
    return 0;
  }

  int DBUser::read_usage(const DoutPrefixProvider *dpp, uint64_t start_epoch, uint64_t end_epoch, uint32_t max_entries,
      bool *is_truncated, RGWUsageIter& usage_iter,
      map<rgw_user_bucket, rgw_usage_log_entry>& usage)
  {
    return 0;
  }

  int DBUser::trim_usage(const DoutPrefixProvider *dpp, uint64_t start_epoch, uint64_t end_epoch)
  {
    return 0;
  }

  int DBUser::load_user(const DoutPrefixProvider *dpp, optional_yield y)
  {
    int ret = 0;

    ret = store->getDB()->get_user(dpp, string("user_id"), get_id().id, info, &attrs,
        &objv_tracker);

    return ret;
  }
  int DBUser::merge_and_store_attrs(const DoutPrefixProvider* dpp, Attrs& new_attrs, optional_yield y)
  {
    for(auto& it : new_attrs) {
  	  attrs[it.first] = it.second;
    }
    return store_user(dpp, y, false);
  }
  int DBUser::store_user(const DoutPrefixProvider* dpp, optional_yield y, bool exclusive, RGWUserInfo* old_info)
  {
    int ret = 0;

    ret = store->getDB()->store_user(dpp, info, exclusive, &attrs, &objv_tracker, old_info);

    return ret;
  }

  int DBUser::remove_user(const DoutPrefixProvider* dpp, optional_yield y)
  {
    int ret = 0;

    ret = store->getDB()->remove_user(dpp, info, &objv_tracker);

    return ret;
  }

  int DBUser::verify_mfa(const std::string& mfa_str, bool* verified, const DoutPrefixProvider *dpp, optional_yield y)
  {
    *verified = false;
    return 0;
  }

  int DBBucket::remove_bucket(const DoutPrefixProvider *dpp, bool delete_children, bool forward_to_master, req_info* req_info, optional_yield y)
  {
    int ret;

    ret = load_bucket(dpp, y);
    if (ret < 0)
      return ret;

    /* XXX: handle delete_children */

    if (!delete_children) {
      /* Check if there are any objects */
      rgw::sal::Bucket::ListParams params;
      params.list_versions = true;
      params.allow_unordered = true;

      rgw::sal::Bucket::ListResults results;

      results.objs.clear();

      ret = list(dpp, params, 2, results, null_yield);

      if (ret < 0) {
        ldpp_dout(dpp, 20) << __func__ << ": Bucket list objects returned " <<
        ret << dendl;
        return ret;
      }

      if (!results.objs.empty()) {
        ret = -ENOTEMPTY;
        ldpp_dout(dpp, -1) << __func__ << ": Bucket Not Empty.. returning " <<
        ret << dendl;
        return ret;
      }
    }

    ret = store->getDB()->remove_bucket(dpp, info);

    return ret;
  }

  int DBBucket::remove_bucket_bypass_gc(int concurrent_max, bool
					keep_index_consistent,
					optional_yield y, const
					DoutPrefixProvider *dpp) {
    return 0;
  }

  int DBBucket::load_bucket(const DoutPrefixProvider *dpp, optional_yield y, bool get_stats)
  {
    int ret = 0;

    ret = store->getDB()->get_bucket_info(dpp, string("name"), "", info, &attrs,
        &mtime, &bucket_version);

    return ret;
  }

  /* stats - Not for first pass */
  int DBBucket::read_stats(const DoutPrefixProvider *dpp,
      const bucket_index_layout_generation& idx_layout,
      int shard_id,
      std::string *bucket_ver, std::string *master_ver,
      std::map<RGWObjCategory, RGWStorageStats>& stats,
      std::string *max_marker, bool *syncstopped)
  {
    return 0;
  }

  int DBBucket::read_stats_async(const DoutPrefixProvider *dpp, const bucket_index_layout_generation& idx_layout, int shard_id, RGWGetBucketStats_CB *ctx)
  {
    return 0;
  }

  int DBBucket::sync_user_stats(const DoutPrefixProvider *dpp, optional_yield y)
  {
    return 0;
  }

  int DBBucket::update_container_stats(const DoutPrefixProvider *dpp)
  {
    return 0;
  }

  int DBBucket::check_bucket_shards(const DoutPrefixProvider *dpp)
  {
    return 0;
  }

  int DBBucket::chown(const DoutPrefixProvider *dpp, User& new_user, optional_yield y)
  {
    int ret;

    ret = store->getDB()->update_bucket(dpp, "owner", info, false, &(new_user.get_id()), nullptr, nullptr, nullptr);
    return ret;
  }

  int DBBucket::put_info(const DoutPrefixProvider *dpp, bool exclusive, ceph::real_time _mtime)
  {
    int ret;

    ret = store->getDB()->update_bucket(dpp, "info", info, exclusive, nullptr, nullptr, &_mtime, &info.objv_tracker);

    return ret;

  }

  /* Make sure to call get_bucket_info() if you need it first */
  bool DBBucket::is_owner(User* user)
  {
    return (info.owner.compare(user->get_id()) == 0);
  }

  int DBBucket::check_empty(const DoutPrefixProvider *dpp, optional_yield y)
  {
    /* XXX: Check if bucket contains any objects */
    return 0;
  }

  int DBBucket::check_quota(const DoutPrefixProvider *dpp, RGWQuota& quota, uint64_t obj_size,
      optional_yield y, bool check_size_only)
  {
    /* Not Handled in the first pass as stats are also needed */
    return 0;
  }

  int DBBucket::merge_and_store_attrs(const DoutPrefixProvider *dpp, Attrs& new_attrs, optional_yield y)
  {
    int ret = 0;

    for(auto& it : new_attrs) {
	    attrs[it.first] = it.second;
    }

    /* XXX: handle has_instance_obj like in set_bucket_instance_attrs() */

    ret = store->getDB()->update_bucket(dpp, "attrs", info, false, nullptr, &new_attrs, nullptr, &get_info().objv_tracker);

    return ret;
  }

  int DBBucket::try_refresh_info(const DoutPrefixProvider *dpp, ceph::real_time *pmtime)
  {
    int ret = 0;

    ret = store->getDB()->get_bucket_info(dpp, string("name"), "", info, &attrs,
        pmtime, &bucket_version);

    return ret;
  }

  /* XXX: usage and stats not supported in the first pass */
  int DBBucket::read_usage(const DoutPrefixProvider *dpp, uint64_t start_epoch, uint64_t end_epoch,
      uint32_t max_entries, bool *is_truncated,
      RGWUsageIter& usage_iter,
      map<rgw_user_bucket, rgw_usage_log_entry>& usage)
  {
    return 0;
  }

  int DBBucket::trim_usage(const DoutPrefixProvider *dpp, uint64_t start_epoch, uint64_t end_epoch)
  {
    return 0;
  }

  int DBBucket::remove_objs_from_index(const DoutPrefixProvider *dpp, std::list<rgw_obj_index_key>& objs_to_unlink)
  {
    /* XXX: CHECK: Unlike RadosStore, there is no seperate bucket index table.
     * Delete all the object in the list from the object table of this
     * bucket
     */
    return 0;
  }

  int DBBucket::check_index(const DoutPrefixProvider *dpp, std::map<RGWObjCategory, RGWStorageStats>& existing_stats, std::map<RGWObjCategory, RGWStorageStats>& calculated_stats)
  {
    /* XXX: stats not supported yet */
    return 0;
  }

  int DBBucket::rebuild_index(const DoutPrefixProvider *dpp)
  {
    /* there is no index table in dbstore. Not applicable */
    return 0;
  }

  int DBBucket::set_tag_timeout(const DoutPrefixProvider *dpp, uint64_t timeout)
  {
    /* XXX: CHECK: set tag timeout for all the bucket objects? */
    return 0;
  }

  int DBBucket::purge_instance(const DoutPrefixProvider *dpp)
  {
    /* XXX: CHECK: for dbstore only single instance supported.
     * Remove all the objects for that instance? Anything extra needed?
     */
    return 0;
  }

  int DBBucket::set_acl(const DoutPrefixProvider *dpp, RGWAccessControlPolicy &acl, optional_yield y)
  {
    int ret = 0;
    bufferlist aclbl;

    acls = acl;
    acl.encode(aclbl);

    Attrs attrs = get_attrs();
    attrs[RGW_ATTR_ACL] = aclbl;

    ret = store->getDB()->update_bucket(dpp, "attrs", info, false, &(acl.get_owner().get_id()), &attrs, nullptr, nullptr);

    return ret;
  }

  std::unique_ptr<Object> DBBucket::get_object(const rgw_obj_key& k)
  {
    return std::make_unique<DBObject>(this->store, k, this);
  }

  int DBBucket::list(const DoutPrefixProvider *dpp, ListParams& params, int max, ListResults& results, optional_yield y)
  {
    int ret = 0;

    results.objs.clear();

    DB::Bucket target(store->getDB(), get_info());
    DB::Bucket::List list_op(&target);

    list_op.params.prefix = params.prefix;
    list_op.params.delim = params.delim;
    list_op.params.marker = params.marker;
    list_op.params.ns = params.ns;
    list_op.params.end_marker = params.end_marker;
    list_op.params.ns = params.ns;
    list_op.params.enforce_ns = params.enforce_ns;
    list_op.params.access_list_filter = params.access_list_filter;
    list_op.params.force_check_filter = params.force_check_filter;
    list_op.params.list_versions = params.list_versions;
    list_op.params.allow_unordered = params.allow_unordered;

    results.objs.clear();
    ret = list_op.list_objects(dpp, max, &results.objs, &results.common_prefixes, &results.is_truncated);
    if (ret >= 0) {
      results.next_marker = list_op.get_next_marker();
      params.marker = results.next_marker;
    }

    return ret;
  }

  std::unique_ptr<MultipartUpload> DBBucket::get_multipart_upload(
				const std::string& oid,
				std::optional<std::string> upload_id,
				ACLOwner owner, ceph::real_time mtime) {
    return std::make_unique<DBMultipartUpload>(this->store, this, oid, upload_id,
						std::move(owner), mtime);
  }

  int DBBucket::list_multiparts(const DoutPrefixProvider *dpp,
				const string& prefix,
				string& marker,
				const string& delim,
				const int& max_uploads,
				vector<std::unique_ptr<MultipartUpload>>& uploads,
				map<string, bool> *common_prefixes,
				bool *is_truncated) {
    return 0;
  }

  int DBBucket::abort_multiparts(const DoutPrefixProvider* dpp,
				 CephContext* cct) {
    return 0;
  }

  void DBStore::finalize(void)
  {
    if (dbsm)
      dbsm->destroyAllHandles();
  }

  const std::string&  DBZoneGroup::get_endpoint() const {
    if (!group->endpoints.empty()) {
      return group->endpoints.front();
    } else {
      // use zonegroup's master zone endpoints
      auto z = group->zones.find(group->master_zone);
      if (z != group->zones.end() && !z->second.endpoints.empty()) {
	return z->second.endpoints.front();
      }
    }
    return empty;
  }

  bool DBZoneGroup::placement_target_exists(std::string& target) const {
    return !!group->placement_targets.count(target);
  }

  void DBZoneGroup::get_placement_target_names(std::set<std::string>& names) const {
    for (const auto& target : group->placement_targets) {
      names.emplace(target.second.name);
    }
  }

  ZoneGroup& DBZone::get_zonegroup()
  {
    return *zonegroup;
  }

  const RGWZoneParams& DBZone::get_rgw_params()
  {
    return *zone_params;
  }

  const std::string& DBZone::get_id()
  {
    return zone_params->get_id();
  }


  const std::string& DBZone::get_name() const
  {
    return zone_params->get_name();
  }

  bool DBZone::is_writeable()
  {
    return true;
  }

  bool DBZone::get_redirect_endpoint(std::string* endpoint)
  {
    return false;
  }

  bool DBZone::has_zonegroup_api(const std::string& api) const
  {
    return false;
  }

  const std::string& DBZone::get_current_period_id()
  {
    return current_period->get_id();
  }

  const RGWAccessKey& DBZone::get_system_key()
  {
    return zone_params->system_key;
  }

  const std::string& DBZone::get_realm_name()
  {
    return realm->get_name();
  }

  const std::string& DBZone::get_realm_id()
  {
    return realm->get_id();
  }

  RGWBucketSyncPolicyHandlerRef DBZone::get_sync_policy_handler()
  {
    return nullptr;
  }

  std::unique_ptr<LuaManager> DBStore::get_lua_manager()
  {
    return std::make_unique<DBLuaManager>(this);
  }

  int DBObject::get_obj_state(const DoutPrefixProvider* dpp, RGWObjState **pstate, optional_yield y, bool follow_olh)
  {
    RGWObjState* astate;
    DB::Object op_target(store->getDB(), get_bucket()->get_info(), get_obj());
    int ret = op_target.get_obj_state(dpp, get_bucket()->get_info(), get_obj(), follow_olh, &astate);
    if (ret < 0) {
      return ret;
    }

    /* Don't overwrite obj, atomic, or prefetch */
    rgw_obj obj = get_obj();
    bool is_atomic = state.is_atomic;
    bool prefetch_data = state.prefetch_data;

    state = *astate;
    *pstate = &state;

    state.obj = obj;
    state.is_atomic = is_atomic;
    state.prefetch_data = prefetch_data;
    return ret;
  }

  int DBObject::read_attrs(const DoutPrefixProvider* dpp, DB::Object::Read &read_op, optional_yield y, rgw_obj* target_obj)
  {
    read_op.params.attrs = &state.attrset;
    read_op.params.target_obj = target_obj;
    read_op.params.obj_size = &state.size;
    read_op.params.lastmod = &state.mtime;

    return read_op.prepare(dpp);
  }

  int DBObject::set_obj_attrs(const DoutPrefixProvider* dpp, Attrs* setattrs, Attrs* delattrs, optional_yield y)
  {
    Attrs empty;
    DB::Object op_target(store->getDB(),
        get_bucket()->get_info(), get_obj());
    return op_target.set_attrs(dpp, setattrs ? *setattrs : empty, delattrs);
  }

  int DBObject::get_obj_attrs(optional_yield y, const DoutPrefixProvider* dpp, rgw_obj* target_obj)
  {
    DB::Object op_target(store->getDB(), get_bucket()->get_info(), get_obj());
    DB::Object::Read read_op(&op_target);

    return read_attrs(dpp, read_op, y, target_obj);
  }

  int DBObject::modify_obj_attrs(const char* attr_name, bufferlist& attr_val, optional_yield y, const DoutPrefixProvider* dpp)
  {
    rgw_obj target = get_obj();
    int r = get_obj_attrs(y, dpp, &target);
    if (r < 0) {
      return r;
    }
    set_atomic();
    state.attrset[attr_name] = attr_val;
    return set_obj_attrs(dpp, &state.attrset, nullptr, y);
  }

  int DBObject::delete_obj_attrs(const DoutPrefixProvider* dpp, const char* attr_name, optional_yield y)
  {
    Attrs rmattr;
    bufferlist bl;

    set_atomic();
    rmattr[attr_name] = bl;
    return set_obj_attrs(dpp, nullptr, &rmattr, y);
  }

  bool DBObject::is_expired() {
    return false;
  }

  void DBObject::gen_rand_obj_instance_name()
  {
     store->getDB()->gen_rand_obj_instance_name(&state.obj.key);
  }


  int DBObject::omap_get_vals(const DoutPrefixProvider *dpp, const std::string& marker, uint64_t count,
      std::map<std::string, bufferlist> *m,
      bool* pmore, optional_yield y)
  {
    DB::Object op_target(store->getDB(),
        get_bucket()->get_info(), get_obj());
    return op_target.obj_omap_get_vals(dpp, marker, count, m, pmore);
  }

  int DBObject::omap_get_all(const DoutPrefixProvider *dpp, std::map<std::string, bufferlist> *m,
      optional_yield y)
  {
    DB::Object op_target(store->getDB(),
        get_bucket()->get_info(), get_obj());
    return op_target.obj_omap_get_all(dpp, m);
  }

  int DBObject::omap_get_vals_by_keys(const DoutPrefixProvider *dpp, const std::string& oid,
      const std::set<std::string>& keys,
      Attrs* vals)
  {
    DB::Object op_target(store->getDB(),
        get_bucket()->get_info(), get_obj());
    return op_target.obj_omap_get_vals_by_keys(dpp, oid, keys, vals);
  }

  int DBObject::omap_set_val_by_key(const DoutPrefixProvider *dpp, const std::string& key, bufferlist& val,
      bool must_exist, optional_yield y)
  {
    DB::Object op_target(store->getDB(),
        get_bucket()->get_info(), get_obj());
    return op_target.obj_omap_set_val_by_key(dpp, key, val, must_exist);
  }

  int DBObject::chown(User& new_user, const DoutPrefixProvider* dpp, optional_yield y)
  {
    return 0;
  }

  std::unique_ptr<MPSerializer> DBObject::get_serializer(const DoutPrefixProvider *dpp,
							 const std::string& lock_name)
  {
    return std::make_unique<MPDBSerializer>(dpp, store, this, lock_name);
  }

  int DBObject::transition(Bucket* bucket,
      const rgw_placement_rule& placement_rule,
      const real_time& mtime,
      uint64_t olh_epoch,
      const DoutPrefixProvider* dpp,
      optional_yield y)
  {
    DB::Object op_target(store->getDB(),
        get_bucket()->get_info(), get_obj());
    return op_target.transition(dpp, placement_rule, mtime, olh_epoch);
  }

  bool DBObject::placement_rules_match(rgw_placement_rule& r1, rgw_placement_rule& r2)
  {
    /* XXX: support single default zone and zonegroup for now */
    return true;
  }

  int DBObject::dump_obj_layout(const DoutPrefixProvider *dpp, optional_yield y, Formatter* f)
  {
    return 0;
  }

  std::unique_ptr<Object::ReadOp> DBObject::get_read_op()
  {
    return std::make_unique<DBObject::DBReadOp>(this, nullptr);
  }

  DBObject::DBReadOp::DBReadOp(DBObject *_source, RGWObjectCtx *_rctx) :
    source(_source),
    rctx(_rctx),
    op_target(_source->store->getDB(),
        _source->get_bucket()->get_info(),
        _source->get_obj()),
    parent_op(&op_target)
  { }

  int DBObject::DBReadOp::prepare(optional_yield y, const DoutPrefixProvider* dpp)
  {
    uint64_t obj_size;

    parent_op.conds.mod_ptr = params.mod_ptr;
    parent_op.conds.unmod_ptr = params.unmod_ptr;
    parent_op.conds.high_precision_time = params.high_precision_time;
    parent_op.conds.mod_zone_id = params.mod_zone_id;
    parent_op.conds.mod_pg_ver = params.mod_pg_ver;
    parent_op.conds.if_match = params.if_match;
    parent_op.conds.if_nomatch = params.if_nomatch;
    parent_op.params.lastmod = params.lastmod;
    parent_op.params.target_obj = params.target_obj;
    parent_op.params.obj_size = &obj_size;
    parent_op.params.attrs = &source->get_attrs();

    int ret = parent_op.prepare(dpp);
    if (ret < 0)
      return ret;

    source->set_key(parent_op.state.obj.key);
    source->set_obj_size(obj_size);

    return ret;
  }

  int DBObject::DBReadOp::read(int64_t ofs, int64_t end, bufferlist& bl, optional_yield y, const DoutPrefixProvider* dpp)
  {
    return parent_op.read(ofs, end, bl, dpp);
  }

  int DBObject::DBReadOp::get_attr(const DoutPrefixProvider* dpp, const char* name, bufferlist& dest, optional_yield y)
  {
    return parent_op.get_attr(dpp, name, dest);
  }

  std::unique_ptr<Object::DeleteOp> DBObject::get_delete_op()
  {
    return std::make_unique<DBObject::DBDeleteOp>(this);
  }

  DBObject::DBDeleteOp::DBDeleteOp(DBObject *_source) :
    source(_source),
    op_target(_source->store->getDB(),
        _source->get_bucket()->get_info(),
        _source->get_obj()),
    parent_op(&op_target)
  { }

  int DBObject::DBDeleteOp::delete_obj(const DoutPrefixProvider* dpp, optional_yield y)
  {
    parent_op.params.bucket_owner = params.bucket_owner.get_id();
    parent_op.params.versioning_status = params.versioning_status;
    parent_op.params.obj_owner = params.obj_owner;
    parent_op.params.olh_epoch = params.olh_epoch;
    parent_op.params.marker_version_id = params.marker_version_id;
    parent_op.params.bilog_flags = params.bilog_flags;
    parent_op.params.remove_objs = params.remove_objs;
    parent_op.params.expiration_time = params.expiration_time;
    parent_op.params.unmod_since = params.unmod_since;
    parent_op.params.mtime = params.mtime;
    parent_op.params.high_precision_time = params.high_precision_time;
    parent_op.params.zones_trace = params.zones_trace;
    parent_op.params.abortmp = params.abortmp;
    parent_op.params.parts_accounted_size = params.parts_accounted_size;

    int ret = parent_op.delete_obj(dpp);
    if (ret < 0)
      return ret;

    result.delete_marker = parent_op.result.delete_marker;
    result.version_id = parent_op.result.version_id;

    return ret;
  }

  int DBObject::delete_object(const DoutPrefixProvider* dpp, optional_yield y, bool prevent_versioning)
  {
    DB::Object del_target(store->getDB(), bucket->get_info(), get_obj());
    DB::Object::Delete del_op(&del_target);

    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);
  }

  int DBObject::delete_obj_aio(const DoutPrefixProvider* dpp, RGWObjState* astate,
      Completions* aio, bool keep_index_consistent,
      optional_yield y)
  {
    /* XXX: Make it async */
    return 0;
  }

  int DBObject::copy_object(User* user,
      req_info* info,
      const rgw_zone_id& source_zone,
      rgw::sal::Object* dest_object,
      rgw::sal::Bucket* dest_bucket,
      rgw::sal::Bucket* src_bucket,
      const rgw_placement_rule& dest_placement,
      ceph::real_time* src_mtime,
      ceph::real_time* mtime,
      const ceph::real_time* mod_ptr,
      const ceph::real_time* unmod_ptr,
      bool high_precision_time,
      const char* if_match,
      const char* if_nomatch,
      AttrsMod attrs_mod,
      bool copy_if_newer,
      Attrs& attrs,
      RGWObjCategory category,
      uint64_t olh_epoch,
      boost::optional<ceph::real_time> delete_at,
      std::string* version_id,
      std::string* tag,
      std::string* etag,
      void (*progress_cb)(off_t, void *),
      void* progress_data,
      const DoutPrefixProvider* dpp,
      optional_yield y)
  {
        return 0;
  }

  int DBObject::DBReadOp::iterate(const DoutPrefixProvider* dpp, int64_t ofs, int64_t end, RGWGetDataCB* cb, optional_yield y)
  {
    return parent_op.iterate(dpp, ofs, end, cb);
  }

  int DBObject::swift_versioning_restore(bool& restored,
      const DoutPrefixProvider* dpp)
  {
    return 0;
  }

  int DBObject::swift_versioning_copy(const DoutPrefixProvider* dpp,
      optional_yield y)
  {
    return 0;
  }

  int DBMultipartUpload::abort(const DoutPrefixProvider *dpp, CephContext *cct)
  {
    std::unique_ptr<rgw::sal::Object> meta_obj = get_meta_obj();
    meta_obj->set_in_extra_data(true);
    meta_obj->set_hash_source(mp_obj.get_key());
    int ret;

    std::unique_ptr<rgw::sal::Object::DeleteOp> del_op = meta_obj->get_delete_op();
    del_op->params.bucket_owner = bucket->get_acl_owner();
    del_op->params.versioning_status = 0;

    // 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);
    if (ret < 0) {
      ldpp_dout(dpp, 20) << __func__ << ": del_op.delete_obj returned " <<
        ret << dendl;
    }
    return (ret == -ENOENT) ? -ERR_NO_SUCH_UPLOAD : ret;
  }

  static string mp_ns = RGW_OBJ_NS_MULTIPART;

  std::unique_ptr<rgw::sal::Object> DBMultipartUpload::get_meta_obj()
  {
    return bucket->get_object(rgw_obj_key(get_meta(), string(), mp_ns));
  }

  int DBMultipartUpload::init(const DoutPrefixProvider *dpp, optional_yield y, ACLOwner& owner, rgw_placement_rule& dest_placement, rgw::sal::Attrs& attrs)
  {
    int ret;
    std::string oid = mp_obj.get_key();

    char buf[33];
    std::unique_ptr<rgw::sal::Object> obj; // create meta obj
    gen_rand_alphanumeric(store->ctx(), buf, sizeof(buf) - 1);
    std::string upload_id = MULTIPART_UPLOAD_ID_PREFIX; /* v2 upload id */
    upload_id.append(buf);

    mp_obj.init(oid, upload_id);
    obj = get_meta_obj();

    DB::Object op_target(store->getDB(), obj->get_bucket()->get_info(),
			       obj->get_obj());
    DB::Object::Write obj_op(&op_target);

    /* Create meta object */
    obj_op.meta.owner = owner.get_id();
    obj_op.meta.category = RGWObjCategory::MultiMeta;
    obj_op.meta.flags = PUT_OBJ_CREATE_EXCL;
    obj_op.meta.mtime = &mtime;

    multipart_upload_info upload_info;
    upload_info.dest_placement = dest_placement;

    bufferlist bl;
    encode(upload_info, bl);
    obj_op.meta.data = &bl; 
    ret = obj_op.prepare(dpp);
    if (ret < 0)
      return ret;
    ret = obj_op.write_meta(dpp, bl.length(), bl.length(), attrs);

    return ret;
  }

  int DBMultipartUpload::list_parts(const DoutPrefixProvider *dpp, CephContext *cct,
				     int num_parts, int marker,
				     int *next_marker, bool *truncated,
				     bool assume_unsorted)
  {
    std::list<RGWUploadPartInfo> parts_map;

    std::unique_ptr<rgw::sal::Object> obj = get_meta_obj();

    parts.clear();
    int ret;

    DB::Object op_target(store->getDB(),
        obj->get_bucket()->get_info(), obj->get_obj());
    ret = op_target.get_mp_parts_list(dpp, parts_map);
    if (ret < 0) {
      return ret;
    }

    int last_num = 0;

    while (!parts_map.empty()) {
      std::unique_ptr<DBMultipartPart> part = std::make_unique<DBMultipartPart>();
      RGWUploadPartInfo &pinfo = parts_map.front();
      part->set_info(pinfo);
      if ((int)pinfo.num > marker) {
        last_num = pinfo.num;
        parts[pinfo.num] = std::move(part);
      }
      parts_map.pop_front();
    }

    /* rebuild a map with only num_parts entries */
    std::map<uint32_t, std::unique_ptr<MultipartPart>> new_parts;
    std::map<uint32_t, std::unique_ptr<MultipartPart>>::iterator piter;
    int i;
    for (i = 0, piter = parts.begin();
	 i < num_parts && piter != parts.end();
	 ++i, ++piter) {
      last_num = piter->first;
      new_parts[piter->first] = std::move(piter->second);
    }

    if (truncated) {
      *truncated = (piter != parts.end());
    }

    parts.swap(new_parts);

    if (next_marker) {
      *next_marker = last_num;
    }

    return 0;
  }

  int DBMultipartUpload::complete(const DoutPrefixProvider *dpp,
				   optional_yield y, CephContext* cct,
				   map<int, string>& part_etags,
				   list<rgw_obj_index_key>& remove_objs,
				   uint64_t& accounted_size, bool& compressed,
				   RGWCompressionInfo& cs_info, off_t& ofs,
				   std::string& tag, ACLOwner& owner,
				   uint64_t olh_epoch,
				   rgw::sal::Object* target_obj)
  {
    char final_etag[CEPH_CRYPTO_MD5_DIGESTSIZE];
    char final_etag_str[CEPH_CRYPTO_MD5_DIGESTSIZE * 2 + 16];
    std::string etag;
    bufferlist etag_bl;
    MD5 hash;
    bool truncated;
    int ret;

    int total_parts = 0;
    int handled_parts = 0;
    int max_parts = 1000;
    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();

    ofs = 0;
    accounted_size = 0;
    do {
      ret = list_parts(dpp, cct, max_parts, marker, &marker, &truncated);
      if (ret == -ENOENT) {
        ret = -ERR_NO_SUCH_UPLOAD;
      }
      if (ret < 0)
        return ret;

      total_parts += parts.size();
      if (!truncated && total_parts != (int)part_etags.size()) {
        ldpp_dout(dpp, 0) << "NOTICE: total parts mismatch: have: " << total_parts
		       << " expected: " << part_etags.size() << dendl;
        ret = -ERR_INVALID_PART;
        return ret;
      }

      for (auto obj_iter = parts.begin(); etags_iter != part_etags.end() && obj_iter != parts.end(); ++etags_iter, ++obj_iter, ++handled_parts) {
        DBMultipartPart* part = dynamic_cast<rgw::sal::DBMultipartPart*>(obj_iter->second.get());
        uint64_t part_size = part->get_size();
        if (handled_parts < (int)part_etags.size() - 1 &&
            part_size < min_part_size) {
          ret = -ERR_TOO_SMALL;
          return ret;
        }

        char petag[CEPH_CRYPTO_MD5_DIGESTSIZE];
        if (etags_iter->first != (int)obj_iter->first) {
          ldpp_dout(dpp, 0) << "NOTICE: parts num mismatch: next requested: "
		  	 << etags_iter->first << " next uploaded: "
			 << obj_iter->first << dendl;
          ret = -ERR_INVALID_PART;
          return ret;
        }
        string part_etag = rgw_string_unquote(etags_iter->second);
        if (part_etag.compare(part->get_etag()) != 0) {
          ldpp_dout(dpp, 0) << "NOTICE: etag mismatch: part: " << etags_iter->first
			 << " etag: " << etags_iter->second << dendl;
          ret = -ERR_INVALID_PART;
          return ret;
        }

        hex_to_buf(part->get_etag().c_str(), petag,
		  CEPH_CRYPTO_MD5_DIGESTSIZE);
        hash.Update((const unsigned char *)petag, sizeof(petag));

        RGWUploadPartInfo& obj_part = part->get_info();

        ofs += obj_part.size;
        accounted_size += obj_part.accounted_size;
      }
    } while (truncated);
    hash.Final((unsigned char *)final_etag);

    buf_to_hex((unsigned char *)final_etag, sizeof(final_etag), final_etag_str);
    snprintf(&final_etag_str[CEPH_CRYPTO_MD5_DIGESTSIZE * 2],
	     sizeof(final_etag_str) - CEPH_CRYPTO_MD5_DIGESTSIZE * 2,
           "-%lld", (long long)part_etags.size());
    etag = final_etag_str;
    ldpp_dout(dpp, 10) << "calculated etag: " << etag << dendl;

    etag_bl.append(etag);

    attrs[RGW_ATTR_ETAG] = etag_bl;

    /* XXX: handle compression ? */

    /* Rename all the object data entries with original object name (i.e
     * from 'head_obj.name + "." + upload_id' to head_obj.name) */

    /* Original head object */
    DB::Object op_target(store->getDB(),
			     target_obj->get_bucket()->get_info(),
			     target_obj->get_obj(), get_upload_id());
    DB::Object::Write obj_op(&op_target);
    ret = obj_op.prepare(dpp);

    obj_op.meta.owner = owner.get_id();
    obj_op.meta.flags = PUT_OBJ_CREATE;
    obj_op.meta.category = RGWObjCategory::Main;
    obj_op.meta.modify_tail = true;
    obj_op.meta.completeMultipart = true;

    ret = obj_op.write_meta(dpp, ofs, accounted_size, attrs);
    if (ret < 0)
      return ret;

    /* No need to delete Meta obj here. It is deleted from sal */
    return ret;
  }

  int DBMultipartUpload::get_info(const DoutPrefixProvider *dpp, optional_yield y, rgw_placement_rule** rule, rgw::sal::Attrs* attrs)
  {
    if (!rule && !attrs) {
      return 0;
    }

    if (rule) {
      if (!placement.empty()) {
        *rule = &placement;
        if (!attrs) {
	    /* Don't need attrs, done */
	      return 0;
        }
      } else {
        *rule = nullptr;
      }
    }

    /* We need either attributes or placement, so we need a read */
    std::unique_ptr<rgw::sal::Object> meta_obj;
    meta_obj = get_meta_obj();
    meta_obj->set_in_extra_data(true);

    multipart_upload_info upload_info;
    bufferlist headbl;

    /* Read the obj head which contains the multipart_upload_info */
    std::unique_ptr<rgw::sal::Object::ReadOp> read_op = meta_obj->get_read_op();
    int ret = read_op->prepare(y, dpp);
    if (ret < 0) {
      if (ret == -ENOENT) {
        return -ERR_NO_SUCH_UPLOAD;
      }
      return ret;
    }

    if (attrs) {
      /* Attrs are filled in by prepare */
      *attrs = meta_obj->get_attrs();
      if (!rule || *rule != nullptr) {
        /* placement was cached; don't actually read */
        return 0;
      }
    }

    /* Now read the placement from the head */
    ret = read_op->read(0, store->getDB()->get_max_head_size(), headbl, y, dpp);
    if (ret < 0) {
      if (ret == -ENOENT) {
        return -ERR_NO_SUCH_UPLOAD;
      }
      return ret;
    }

    if (headbl.length() <= 0) {
      return -ERR_NO_SUCH_UPLOAD;
    }

    /* Decode multipart_upload_info */
    auto hiter = headbl.cbegin();
    try {
      decode(upload_info, hiter);
    } catch (buffer::error& err) {
      ldpp_dout(dpp, 0) << "ERROR: failed to decode multipart upload info" << dendl;
      return -EIO;
    }
    placement = upload_info.dest_placement;
    *rule = &placement;

    return 0;
  }

  std::unique_ptr<Writer> DBMultipartUpload::get_writer(
				  const DoutPrefixProvider *dpp,
				  optional_yield y,
				  rgw::sal::Object* obj,
				  const rgw_user& owner,
				  const rgw_placement_rule *ptail_placement_rule,
				  uint64_t part_num,
				  const std::string& part_num_str)
  {
    return std::make_unique<DBMultipartWriter>(dpp, y, this, obj, store, owner,
				 ptail_placement_rule, part_num, part_num_str);
  }

  DBMultipartWriter::DBMultipartWriter(const DoutPrefixProvider *dpp,
	    	    optional_yield y,
                MultipartUpload* upload,
		        rgw::sal::Object* obj,
		        DBStore* _driver,
    		    const rgw_user& _owner,
	    	    const rgw_placement_rule *_ptail_placement_rule,
                uint64_t _part_num, const std::string& _part_num_str):
			StoreWriter(dpp, y),
			store(_driver),
                owner(_owner),
                ptail_placement_rule(_ptail_placement_rule),
                head_obj(obj),
                upload_id(upload->get_upload_id()),
                part_num(_part_num),
                oid(head_obj->get_name() + "." + upload_id +
                    "." + std::to_string(part_num)),
                meta_obj(((DBMultipartUpload*)upload)->get_meta_obj()),
                op_target(_driver->getDB(), head_obj->get_bucket()->get_info(), head_obj->get_obj(), upload_id),
                parent_op(&op_target),
                part_num_str(_part_num_str) {}

  int DBMultipartWriter::prepare(optional_yield y)
  {
    parent_op.prepare(NULL);
    parent_op.set_mp_part_str(upload_id + "." + std::to_string(part_num));
    // XXX: do we need to handle part_num_str??
    return 0;
  }

  int DBMultipartWriter::process(bufferlist&& data, uint64_t offset)
  {
    /* XXX: same as AtomicWriter..consolidate code */
    total_data_size += data.length();

    /* XXX: Optimize all bufferlist copies in this function */

    /* copy head_data into meta. But for multipart we do not
     * need to write head_data */
    uint64_t max_chunk_size = store->getDB()->get_max_chunk_size();
    int excess_size = 0;

    /* Accumulate tail_data till max_chunk_size or flush op */
    bufferlist tail_data;

    if (data.length() != 0) {
        parent_op.meta.data = &head_data; /* Null data ?? */

      /* handle tail )parts.
       * First accumulate and write data into dbstore in its chunk_size
       * parts
       */
      if (!tail_part_size) { /* new tail part */
        tail_part_offset = offset;
      }
      data.begin(0).copy(data.length(), tail_data);
      tail_part_size += tail_data.length();
      tail_part_data.append(tail_data);

      if (tail_part_size < max_chunk_size)  {
        return 0;
      } else {
        int write_ofs = 0;
        while (tail_part_size >= max_chunk_size) {
          excess_size = tail_part_size - max_chunk_size;
          bufferlist tmp;
          tail_part_data.begin(write_ofs).copy(max_chunk_size, tmp);
          /* write tail objects data */
          int ret = parent_op.write_data(dpp, tmp, tail_part_offset);

          if (ret < 0) {
            return ret;
          }

          tail_part_size -= max_chunk_size;
          write_ofs += max_chunk_size;
          tail_part_offset += max_chunk_size;
        }
        /* reset tail parts or update if excess data */
        if (excess_size > 0) { /* wrote max_chunk_size data */
          tail_part_size = excess_size;
          bufferlist tmp;
          tail_part_data.begin(write_ofs).copy(excess_size, tmp);
          tail_part_data = tmp;
        } else {
          tail_part_size = 0;
          tail_part_data.clear();
          tail_part_offset = 0;
        }
      }
    } else {
      if (tail_part_size == 0) {
        return 0; /* nothing more to write */
      }

      /* flush watever tail data is present */
      int ret = parent_op.write_data(dpp, tail_part_data, tail_part_offset);
      if (ret < 0) {
        return ret;
      }
      tail_part_size = 0;
      tail_part_data.clear();
      tail_part_offset = 0;
    }

    return 0;
  }

  int DBMultipartWriter::complete(size_t accounted_size, const std::string& etag,
                       ceph::real_time *mtime, 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)
  {
    int ret = 0;
    /* XXX: same as AtomicWriter..consolidate code */
    parent_op.meta.mtime = mtime;
    parent_op.meta.delete_at = delete_at;
    parent_op.meta.if_match = if_match;
    parent_op.meta.if_nomatch = if_nomatch;
    parent_op.meta.user_data = user_data;
    parent_op.meta.zones_trace = zones_trace;
    
    /* XXX: handle accounted size */
    accounted_size = total_data_size;

    if (ret < 0)
      return ret;

    RGWUploadPartInfo info;
    info.num = part_num;
    info.etag = etag;
    info.size = total_data_size;
    info.accounted_size = accounted_size;
    info.modified = real_clock::now();
    //info.manifest = manifest;

    DB::Object op_target(store->getDB(),
        meta_obj->get_bucket()->get_info(), meta_obj->get_obj());
    ret = op_target.add_mp_part(dpp, info);
    if (ret < 0) {
      return ret == -ENOENT ? -ERR_NO_SUCH_UPLOAD : ret;
    }

    return 0;
  }

  DBAtomicWriter::DBAtomicWriter(const DoutPrefixProvider *dpp,
	    	    optional_yield y,
		        rgw::sal::Object* _obj,
		        DBStore* _driver,
    		    const rgw_user& _owner,
	    	    const rgw_placement_rule *_ptail_placement_rule,
		        uint64_t _olh_epoch,
		        const std::string& _unique_tag) :
			StoreWriter(dpp, y),
			store(_driver),
                owner(_owner),
                ptail_placement_rule(_ptail_placement_rule),
                olh_epoch(_olh_epoch),
                unique_tag(_unique_tag),
                obj(_driver, _obj->get_key(), _obj->get_bucket()),
                op_target(_driver->getDB(), obj.get_bucket()->get_info(), obj.get_obj()),
                parent_op(&op_target) {}

  int DBAtomicWriter::prepare(optional_yield y)
  {
    return parent_op.prepare(NULL); /* send dpp */
  }

  int DBAtomicWriter::process(bufferlist&& data, uint64_t offset)
  {
    total_data_size += data.length();

    /* XXX: Optimize all bufferlist copies in this function */

    /* copy head_data into meta. */
    uint64_t head_size = store->getDB()->get_max_head_size();
    unsigned head_len = 0;
    uint64_t max_chunk_size = store->getDB()->get_max_chunk_size();
    int excess_size = 0;

    /* Accumulate tail_data till max_chunk_size or flush op */
    bufferlist tail_data;

    if (data.length() != 0) {
      if (offset < head_size) {
        /* XXX: handle case (if exists) where offset > 0 & < head_size */
        head_len = std::min((uint64_t)data.length(),
                                    head_size - offset);
        bufferlist tmp;
        data.begin(0).copy(head_len, tmp);
        head_data.append(tmp);

        parent_op.meta.data = &head_data;
        if (head_len == data.length()) {
          return 0;
        }

        /* Move offset by copy_len */
        offset = head_len;
      }

      /* handle tail parts.
       * First accumulate and write data into dbstore in its chunk_size
       * parts
       */
      if (!tail_part_size) { /* new tail part */
        tail_part_offset = offset;
      }
      data.begin(head_len).copy(data.length() - head_len, tail_data);
      tail_part_size += tail_data.length();
      tail_part_data.append(tail_data);

      if (tail_part_size < max_chunk_size)  {
        return 0;
      } else {
        int write_ofs = 0;
        while (tail_part_size >= max_chunk_size) {
          excess_size = tail_part_size - max_chunk_size;
          bufferlist tmp;
          tail_part_data.begin(write_ofs).copy(max_chunk_size, tmp);
          /* write tail objects data */
          int ret = parent_op.write_data(dpp, tmp, tail_part_offset);

          if (ret < 0) {
            return ret;
          }

          tail_part_size -= max_chunk_size;
          write_ofs += max_chunk_size;
          tail_part_offset += max_chunk_size;
        }
        /* reset tail parts or update if excess data */
        if (excess_size > 0) { /* wrote max_chunk_size data */
          tail_part_size = excess_size;
          bufferlist tmp;
          tail_part_data.begin(write_ofs).copy(excess_size, tmp);
          tail_part_data = tmp;
        } else {
          tail_part_size = 0;
          tail_part_data.clear();
          tail_part_offset = 0;
        }
      }
    } else {
      if (tail_part_size == 0) {
        return 0; /* nothing more to write */
      }

      /* flush watever tail data is present */
      int ret = parent_op.write_data(dpp, tail_part_data, tail_part_offset);
      if (ret < 0) {
        return ret;
      }
      tail_part_size = 0;
      tail_part_data.clear();
      tail_part_offset = 0;
    }

    return 0;
  }

  int DBAtomicWriter::complete(size_t accounted_size, const std::string& etag,
                         ceph::real_time *mtime, 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)
  {
    parent_op.meta.mtime = mtime;
    parent_op.meta.delete_at = delete_at;
    parent_op.meta.if_match = if_match;
    parent_op.meta.if_nomatch = if_nomatch;
    parent_op.meta.user_data = user_data;
    parent_op.meta.zones_trace = zones_trace;
    parent_op.meta.category = RGWObjCategory::Main;
    
    /* XXX: handle accounted size */
    accounted_size = total_data_size;
    int ret = parent_op.write_meta(dpp, total_data_size, accounted_size, attrs);
    if (canceled) {
      *canceled = parent_op.meta.canceled;
    }

    return ret;

  }

  std::unique_ptr<RGWRole> DBStore::get_role(std::string name,
      std::string tenant,
      std::string path,
      std::string trust_policy,
      std::string max_session_duration_str,
      std::multimap<std::string,std::string> tags)
  {
    RGWRole* p = nullptr;
    return std::unique_ptr<RGWRole>(p);
  }

  std::unique_ptr<RGWRole> DBStore::get_role(std::string id)
  {
    RGWRole* p = nullptr;
    return std::unique_ptr<RGWRole>(p);
  }

  std::unique_ptr<RGWRole> DBStore::get_role(const RGWRoleInfo& info)
  {
    RGWRole* p = nullptr;
    return std::unique_ptr<RGWRole>(p);
  }

  int DBStore::get_roles(const DoutPrefixProvider *dpp,
      optional_yield y,
      const std::string& path_prefix,
      const std::string& tenant,
      vector<std::unique_ptr<RGWRole>>& roles)
  {
    return 0;
  }

  std::unique_ptr<RGWOIDCProvider> DBStore::get_oidc_provider()
  {
    RGWOIDCProvider* p = nullptr;
    return std::unique_ptr<RGWOIDCProvider>(p);
  }

  int DBStore::get_oidc_providers(const DoutPrefixProvider *dpp,
      const std::string& tenant,
      vector<std::unique_ptr<RGWOIDCProvider>>& providers)
  {
    return 0;
  }

  std::unique_ptr<Writer> DBStore::get_append_writer(const DoutPrefixProvider *dpp,
				  optional_yield y,
				  rgw::sal::Object* obj,
				  const rgw_user& owner,
				  const rgw_placement_rule *ptail_placement_rule,
				  const std::string& unique_tag,
				  uint64_t position,
				  uint64_t *cur_accounted_size) {
    return nullptr;
  }

  std::unique_ptr<Writer> DBStore::get_atomic_writer(const DoutPrefixProvider *dpp,
				  optional_yield y,
				  rgw::sal::Object* obj,
				  const rgw_user& owner,
				  const rgw_placement_rule *ptail_placement_rule,
				  uint64_t olh_epoch,
				  const std::string& unique_tag) {
    return std::make_unique<DBAtomicWriter>(dpp, y, obj, this, owner,
                    ptail_placement_rule, olh_epoch, unique_tag);
  }

  const std::string& DBStore::get_compression_type(const rgw_placement_rule& rule) {
    return zone.get_rgw_params().get_compression_type(rule);
  }

  bool DBStore::valid_placement(const rgw_placement_rule& rule)
  {
    // XXX: Till zonegroup, zone and storage-classes can be configured
    // for dbstore return true
    return true; //zone.get_rgw_params().valid_placement(rule);
  }

  std::unique_ptr<User> DBStore::get_user(const rgw_user &u)
  {
    return std::make_unique<DBUser>(this, u);
  }

  int DBStore::get_user_by_access_key(const DoutPrefixProvider *dpp, const std::string& key, optional_yield y, std::unique_ptr<User>* user)
  {
    RGWUserInfo uinfo;
    User *u;
    int ret = 0;
    RGWObjVersionTracker objv_tracker;

    ret = getDB()->get_user(dpp, string("access_key"), key, uinfo, nullptr,
        &objv_tracker);

    if (ret < 0)
      return ret;

    u = new DBUser(this, uinfo);

    if (!u)
      return -ENOMEM;

    u->get_version_tracker() = objv_tracker;
    user->reset(u);

    return 0;
  }

  int DBStore::get_user_by_email(const DoutPrefixProvider *dpp, const std::string& email, optional_yield y, std::unique_ptr<User>* user)
  {
    RGWUserInfo uinfo;
    User *u;
    int ret = 0;
    RGWObjVersionTracker objv_tracker;

    ret = getDB()->get_user(dpp, string("email"), email, uinfo, nullptr,
        &objv_tracker);

    if (ret < 0)
      return ret;

    u = new DBUser(this, uinfo);

    if (!u)
      return -ENOMEM;

    u->get_version_tracker() = objv_tracker;
    user->reset(u);

    return ret;
  }

  int DBStore::get_user_by_swift(const DoutPrefixProvider *dpp, const std::string& user_str, optional_yield y, std::unique_ptr<User>* user)
  {
    /* Swift keys and subusers are not supported for now */
    return -ENOTSUP;
  }

  std::string DBStore::get_cluster_id(const DoutPrefixProvider* dpp,  optional_yield y)
  {
    return "PLACEHOLDER"; // for instance unique identifier
  }

  std::unique_ptr<Object> DBStore::get_object(const rgw_obj_key& k)
  {
    return std::make_unique<DBObject>(this, k);
  }


  int DBStore::get_bucket(const DoutPrefixProvider *dpp, User* u, const rgw_bucket& b, std::unique_ptr<Bucket>* bucket, optional_yield y)
  {
    int ret;
    Bucket* bp;

    bp = new DBBucket(this, b, u);
    ret = bp->load_bucket(dpp, y);
    if (ret < 0) {
      delete bp;
      return ret;
    }

    bucket->reset(bp);
    return 0;
  }

  int DBStore::get_bucket(User* u, const RGWBucketInfo& i, std::unique_ptr<Bucket>* bucket)
  {
    Bucket* bp;

    bp = new DBBucket(this, i, u);
    /* Don't need to fetch the bucket info, use the provided one */

    bucket->reset(bp);
    return 0;
  }

  int DBStore::get_bucket(const DoutPrefixProvider *dpp, User* u, const std::string& tenant, const std::string& name, std::unique_ptr<Bucket>* bucket, optional_yield y)
  {
    rgw_bucket b;

    b.tenant = tenant;
    b.name = name;

    return get_bucket(dpp, u, b, bucket, y);
  }

  bool DBStore::is_meta_master()
  {
    return true;
  }

  int DBStore::forward_request_to_master(const DoutPrefixProvider *dpp, User* user, obj_version *objv,
      bufferlist& in_data,
      JSONParser *jp, req_info& info,
      optional_yield y)
  {
    return 0;
  }

  int DBStore::forward_iam_request_to_master(const DoutPrefixProvider *dpp, const RGWAccessKey& key, obj_version* objv,
					     bufferlist& in_data,
					     RGWXMLDecoder::XMLParser* parser, req_info& info,
					     optional_yield y)
  {
      return 0;
  }

  std::string DBStore::zone_unique_id(uint64_t unique_num)
  {
    return "";
  }

  std::string DBStore::zone_unique_trans_id(const uint64_t unique_num)
  {
    return "";
  }

  int DBStore::get_zonegroup(const std::string& id, std::unique_ptr<ZoneGroup>* zg)
  {
    /* XXX: for now only one zonegroup supported */
    ZoneGroup* group = new DBZoneGroup(this, std::make_unique<RGWZoneGroup>());
    if (!group)
      return -ENOMEM;

    zg->reset(group);
    return 0;
  }

  int DBStore::list_all_zones(const DoutPrefixProvider* dpp,
			      std::list<std::string>& zone_ids)
  {
    zone_ids.push_back(zone.get_id());
    return 0;
  }

  int DBStore::cluster_stat(RGWClusterStat& stats)
  {
    return 0;
  }

  std::unique_ptr<Lifecycle> DBStore::get_lifecycle(void)
  {
    return std::make_unique<DBLifecycle>(this);
  }

  std::unique_ptr<Completions> DBStore::get_completions(void)
  {
    return 0;
  }

  int DBLifecycle::get_entry(const std::string& oid, const std::string& marker,
			      std::unique_ptr<LCEntry>* entry)
  {
    return store->getDB()->get_entry(oid, marker, entry);
  }

  int DBLifecycle::get_next_entry(const std::string& oid, const std::string& marker,
				  std::unique_ptr<LCEntry>* entry)
  {
    return store->getDB()->get_next_entry(oid, marker, entry);
  }

  int DBLifecycle::set_entry(const std::string& oid, LCEntry& entry)
  {
    return store->getDB()->set_entry(oid, entry);
  }

  int DBLifecycle::list_entries(const std::string& oid, const std::string& marker,
  				 uint32_t max_entries, vector<std::unique_ptr<LCEntry>>& entries)
  {
    return store->getDB()->list_entries(oid, marker, max_entries, entries);
  }

  int DBLifecycle::rm_entry(const std::string& oid, LCEntry& entry)
  {
    return store->getDB()->rm_entry(oid, entry);
  }

  int DBLifecycle::get_head(const std::string& oid, std::unique_ptr<LCHead>* head)
  {
    return store->getDB()->get_head(oid, head);
  }

  int DBLifecycle::put_head(const std::string& oid, LCHead& head)
  {
    return store->getDB()->put_head(oid, head);
  }

  std::unique_ptr<LCSerializer> DBLifecycle::get_serializer(const std::string& lock_name,
							    const std::string& oid,
							    const std::string& cookie)
  {
    return std::make_unique<LCDBSerializer>(store, oid, lock_name, cookie);
  }

  std::unique_ptr<Notification> DBStore::get_notification(
    rgw::sal::Object* obj, rgw::sal::Object* src_obj, req_state* s,
    rgw::notify::EventType event_type, optional_yield y,
    const std::string* object_name)
  {
    return std::make_unique<DBNotification>(obj, src_obj, event_type);
  }

  std::unique_ptr<Notification> DBStore::get_notification(
    const DoutPrefixProvider* dpp, rgw::sal::Object* obj,
    rgw::sal::Object* src_obj,
    rgw::notify::EventType event_type, rgw::sal::Bucket* _bucket,
    std::string& _user_id, std::string& _user_tenant, std::string& _req_id,
    optional_yield y)
  {
    return std::make_unique<DBNotification>(obj, src_obj, event_type);
  }

  RGWLC* DBStore::get_rgwlc(void) {
    return lc;
  }

  int DBStore::log_usage(const DoutPrefixProvider *dpp, map<rgw_user_bucket, RGWUsageBatch>& usage_info)
  {
    return 0;
  }

  int DBStore::log_op(const DoutPrefixProvider *dpp, string& oid, bufferlist& bl)
  {
    return 0;
  }

  int DBStore::register_to_service_map(const DoutPrefixProvider *dpp, const string& daemon_type,
      const map<string, string>& meta)
  {
    return 0;
  }

  void DBStore::get_ratelimit(RGWRateLimitInfo& bucket_ratelimit, RGWRateLimitInfo& user_ratelimit, RGWRateLimitInfo& anon_ratelimit)
  {
    return;
  }

  void DBStore::get_quota(RGWQuota& quota)
  {
    // XXX: Not handled for the first pass 
    return;
  }

  int DBStore::set_buckets_enabled(const DoutPrefixProvider *dpp, vector<rgw_bucket>& buckets, bool enabled)
  {
    int ret = 0;

    vector<rgw_bucket>::iterator iter;

    for (iter = buckets.begin(); iter != buckets.end(); ++iter) {
      rgw_bucket& bucket = *iter;
      if (enabled) {
        ldpp_dout(dpp, 20) << "enabling bucket name=" << bucket.name << dendl;
      } else {
        ldpp_dout(dpp, 20) << "disabling bucket name=" << bucket.name << dendl;
      }

      RGWBucketInfo info;
      map<string, bufferlist> attrs;
      int r = getDB()->get_bucket_info(dpp, string("name"), "", info, &attrs,
          nullptr, nullptr);
      if (r < 0) {
        ldpp_dout(dpp, 0) << "NOTICE: get_bucket_info on bucket=" << bucket.name << " returned err=" << r << ", skipping bucket" << dendl;
        ret = r;
        continue;
      }
      if (enabled) {
        info.flags &= ~BUCKET_SUSPENDED;
      } else {
        info.flags |= BUCKET_SUSPENDED;
      }

      r = getDB()->update_bucket(dpp, "info", info, false, nullptr, &attrs, nullptr, &info.objv_tracker);
      if (r < 0) {
        ldpp_dout(dpp, 0) << "NOTICE: put_bucket_info on bucket=" << bucket.name << " returned err=" << r << ", skipping bucket" << dendl;
        ret = r;
        continue;
      }
    }
    return ret;
  }

  int DBStore::get_sync_policy_handler(const DoutPrefixProvider *dpp,
      std::optional<rgw_zone_id> zone,
      std::optional<rgw_bucket> bucket,
      RGWBucketSyncPolicyHandlerRef *phandler,
      optional_yield y)
  {
    return 0;
  }

  RGWDataSyncStatusManager* DBStore::get_data_sync_manager(const rgw_zone_id& source_zone)
  {
    return 0;
  }

  int DBStore::read_all_usage(const DoutPrefixProvider *dpp, uint64_t start_epoch, uint64_t end_epoch, 
      uint32_t max_entries, bool *is_truncated,
      RGWUsageIter& usage_iter,
      map<rgw_user_bucket, rgw_usage_log_entry>& usage)
  {
    return 0;
  }

  int DBStore::trim_all_usage(const DoutPrefixProvider *dpp, uint64_t start_epoch, uint64_t end_epoch)
  {
    return 0;
  }

  int DBStore::get_config_key_val(string name, bufferlist *bl)
  {
    return -ENOTSUP;
  }

  int DBStore::meta_list_keys_init(const DoutPrefixProvider *dpp, const string& section, const string& marker, void** phandle)
  {
    return 0;
  }

  int DBStore::meta_list_keys_next(const DoutPrefixProvider *dpp, void* handle, int max, list<string>& keys, bool* truncated)
  {
    return 0;
  }

  void DBStore::meta_list_keys_complete(void* handle)
  {
    return;
  }

  std::string DBStore::meta_get_marker(void* handle)
  {
    return "";
  }

  int DBStore::meta_remove(const DoutPrefixProvider *dpp, string& metadata_key, optional_yield y)
  {
    return 0;
  }

  int DBStore::initialize(CephContext *_cct, const DoutPrefixProvider *_dpp) {
    int ret = 0;
    cct = _cct;
    dpp = _dpp;

    lc = new RGWLC();
    lc->initialize(cct, this);

    if (use_lc_thread) {
      ret = db->createLCTables(dpp);
      lc->start_processor();
    }

    ret = db->createGC(dpp);
    if (ret < 0) {
      ldpp_dout(dpp, 0) <<"GC thread creation failed: ret = " << ret << dendl;
    }

    return ret;
  }

  int DBLuaManager::get_script(const DoutPrefixProvider* dpp, optional_yield y, const std::string& key, std::string& script)
  {
    return -ENOENT;
  }

  int DBLuaManager::put_script(const DoutPrefixProvider* dpp, optional_yield y, const std::string& key, const std::string& script)
  {
    return -ENOENT;
  }

  int DBLuaManager::del_script(const DoutPrefixProvider* dpp, optional_yield y, const std::string& key)
  {
    return -ENOENT;
  }

  int DBLuaManager::add_package(const DoutPrefixProvider* dpp, optional_yield y, const std::string& package_name)
  {
    return -ENOENT;
  }

  int DBLuaManager::remove_package(const DoutPrefixProvider* dpp, optional_yield y, const std::string& package_name)
  {
    return -ENOENT;
  }

  int DBLuaManager::list_packages(const DoutPrefixProvider* dpp, optional_yield y, rgw::lua::packages_t& packages)
  {
    return -ENOENT;
  }
} // namespace rgw::sal

extern "C" {

  void *newDBStore(CephContext *cct)
  {
    rgw::sal::DBStore *driver = new rgw::sal::DBStore();
    DBStoreManager *dbsm = new DBStoreManager(cct);

    DB *db = dbsm->getDB();
    if (!db) {
      delete dbsm;
      delete driver;
      return nullptr;
    }

    driver->setDBStoreManager(dbsm);
    driver->setDB(db);
    db->set_driver((rgw::sal::Driver*)driver);
    db->set_context(cct);

    return driver;
  }

}