diff --git a/cloud/aws/aws_file_system.cc b/cloud/aws/aws_file_system.cc
index b8d98806d92..59e1881bd64 100644
--- a/cloud/aws/aws_file_system.cc
+++ b/cloud/aws/aws_file_system.cc
@@ -1,22 +1,14 @@
 //  Copyright (c) 2016-present, Rockset, Inc.  All rights reserved.
 //
+#include "rocksdb/cloud/cloud_file_system.h"
 #ifndef ROCKSDB_LITE
-#include "cloud/aws/aws_file_system.h"
-
-#include <chrono>
-#include <cinttypes>
-#include <fstream>
-#include <iostream>
 #include <memory>
-#include <set>
 
+#include "cloud/aws/aws_file_system.h"
 #include "cloud/cloud_log_controller_impl.h"
-#include "cloud/cloud_scheduler.h"
-#include "rocksdb/cloud/cloud_storage_provider_impl.h"
-#include "cloud/filename.h"
-#include "port/port.h"
 #include "rocksdb/cloud/cloud_log_controller.h"
 #include "rocksdb/cloud/cloud_storage_provider.h"
+#include "rocksdb/cloud/cloud_storage_provider_impl.h"
 #include "rocksdb/env.h"
 #include "rocksdb/status.h"
 #include "rocksdb/utilities/object_registry.h"
@@ -28,7 +20,6 @@
 #include <aws/core/auth/AWSCredentialsProvider.h>
 #endif
 
-#include "cloud/aws/aws_file.h"
 #include "cloud/db_cloud_impl.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -175,8 +166,7 @@ Status AwsCloudAccessCredentials::GetCredentialsProvider(
 AwsFileSystem::AwsFileSystem(const std::shared_ptr<FileSystem>& underlying_fs,
                              const CloudFileSystemOptions& _cloud_fs_options,
                              const std::shared_ptr<Logger>& info_log)
-    : CloudFileSystemImpl(_cloud_fs_options, underlying_fs, info_log) {
-}
+    : CloudFileSystemImpl(_cloud_fs_options, underlying_fs, info_log) {}
 
 // If you do not specify a region, then S3 buckets are created in the
 // standard-region which might not satisfy read-your-own-writes. So,
@@ -223,8 +213,8 @@ Status AwsFileSystem::NewAwsFileSystem(
   }
   std::unique_ptr<AwsFileSystem> afs(
       new AwsFileSystem(fs, cloud_options, info_log));
-
-  auto env = afs->NewCompositeEnvFromThis(Env::Default());
+  auto env =
+      CloudFileSystemEnv::NewCompositeEnvFromFs(afs.get(), Env::Default());
   ConfigOptions config_options;
   config_options.env = env.get();
   status = afs->PrepareOptions(config_options);
diff --git a/cloud/cloud_file_system.cc b/cloud/cloud_file_system.cc
index 8d659931e0f..ff9a4f9924d 100644
--- a/cloud/cloud_file_system.cc
+++ b/cloud/cloud_file_system.cc
@@ -2,7 +2,6 @@
 #ifndef ROCKSDB_LITE
 
 #include "rocksdb/cloud/cloud_file_system.h"
-
 #ifndef _WIN32_WINNT
 #include <unistd.h>
 #else
@@ -11,16 +10,16 @@
 #include <unordered_map>
 
 #include "cloud/aws/aws_file_system.h"
-#include "rocksdb/cloud/cloud_file_system_impl.h"
 #include "cloud/cloud_log_controller_impl.h"
-#include "rocksdb/cloud/cloud_storage_provider_impl.h"
 #include "cloud/db_cloud_impl.h"
 #include "cloud/filename.h"
 #include "env/composite_env_wrapper.h"
 #include "options/configurable_helper.h"
 #include "options/options_helper.h"
 #include "port/likely.h"
+#include "rocksdb/cloud/cloud_file_system_impl.h"
 #include "rocksdb/cloud/cloud_log_controller.h"
+#include "rocksdb/cloud/cloud_storage_provider_impl.h"
 #include "rocksdb/db.h"
 #include "rocksdb/env.h"
 #include "rocksdb/options.h"
@@ -291,8 +290,8 @@ int offset_of(T1 CloudFileSystemOptions::*member) {
   return int(size_t(&(dummy_ceo_options.*member)) - size_t(&dummy_ceo_options));
 }
 
-static std::unordered_map<std::string, OptionTypeInfo>
-    cloud_fs_option_type_info = {
+const std::unordered_map<std::string, OptionTypeInfo>
+    CloudFileSystemOptions::cloud_fs_option_type_info = {
         {"keep_local_sst_files",
          {offset_of(&CloudFileSystemOptions::keep_local_sst_files),
           OptionType::kBoolean}},
@@ -417,19 +416,7 @@ Status CloudFileSystemOptions::Serialize(const ConfigOptions& config_options,
       reinterpret_cast<const char*>(this), value);
 }
 
-CloudFileSystem::CloudFileSystem(const CloudFileSystemOptions& options,
-                                 const std::shared_ptr<FileSystem>& base,
-                                 const std::shared_ptr<Logger>& logger)
-    : cloud_fs_options(options), base_fs_(base), info_log_(logger) {
-  RegisterOptions(&cloud_fs_options, &cloud_fs_option_type_info);
-}
-
-CloudFileSystem::~CloudFileSystem() {
-  cloud_fs_options.cloud_log_controller.reset();
-  cloud_fs_options.storage_provider.reset();
-}
-
-Status CloudFileSystem::NewAwsFileSystem(
+Status CloudFileSystemEnv::NewAwsFileSystem(
     const std::shared_ptr<FileSystem>& base_fs,
     const std::string& src_cloud_bucket, const std::string& src_cloud_object,
     const std::string& src_cloud_region, const std::string& dest_cloud_bucket,
@@ -484,7 +471,7 @@ int DoRegisterCloudObjects(ObjectLibrary& library, const std::string& arg) {
   return count;
 }
 
-void CloudFileSystem::RegisterCloudObjects(const std::string& arg) {
+void CloudFileSystemEnv::RegisterCloudObjects(const std::string& arg) {
   static std::once_flag do_once;
   std::call_once(do_once, [&]() {
     auto library = ObjectLibrary::Default();
@@ -492,15 +479,16 @@ void CloudFileSystem::RegisterCloudObjects(const std::string& arg) {
   });
 }
 
-std::unique_ptr<Env> CloudFileSystem::NewCompositeEnvFromThis(Env* env) {
+std::unique_ptr<Env> CloudFileSystemEnv::NewCompositeEnvFromFs(FileSystem* fs,
+                                                               Env* env) {
   // We need a shared_ptr<FileSystem> pointing to "this", to initialize the
   // env wrapper, but we don't want that shared_ptr to own the lifecycle for
   // "this". Creating a shared_ptr with a custom no-op deleter instead.
-  std::shared_ptr<FileSystem> fs(this, [](auto* /*p*/) { /*noop*/ });
-  return std::make_unique<CompositeEnvWrapper>(env, fs);
+  std::shared_ptr<FileSystem> fsPtr(fs, [](auto* /*p*/) { /*noop*/ });
+  return std::make_unique<CompositeEnvWrapper>(env, fsPtr);
 }
 
-Status CloudFileSystem::CreateFromString(
+Status CloudFileSystemEnv::CreateFromString(
     const ConfigOptions& config_options, const std::string& value,
     std::unique_ptr<CloudFileSystem>* result) {
   RegisterCloudObjects();
@@ -529,13 +517,13 @@ Status CloudFileSystem::CreateFromString(
   copy.invoke_prepare_options = false;  // Prepare here, not there
   s = ObjectRegistry::NewInstance()->NewUniqueObject<FileSystem>(id, &fs);
   if (s.ok()) {
-    auto* cfs = dynamic_cast<CloudFileSystem*>(fs.get());
+    auto* cfs = dynamic_cast<CloudFileSystemImpl*>(fs.get());
     assert(cfs);
     if (!options.empty()) {
       s = cfs->ConfigureFromMap(copy, options);
     }
     if (s.ok() && config_options.invoke_prepare_options) {
-      auto env = cfs->NewCompositeEnvFromThis(copy.env);
+      auto env = NewCompositeEnvFromFs(cfs, copy.env);
       copy.invoke_prepare_options = config_options.invoke_prepare_options;
       copy.env = env.get();
       s = cfs->PrepareOptions(copy);
@@ -553,7 +541,7 @@ Status CloudFileSystem::CreateFromString(
   return s;
 }
 
-Status CloudFileSystem::CreateFromString(
+Status CloudFileSystemEnv::CreateFromString(
     const ConfigOptions& config_options, const std::string& value,
     const CloudFileSystemOptions& cloud_options,
     std::unique_ptr<CloudFileSystem>* result) {
@@ -583,7 +571,7 @@ Status CloudFileSystem::CreateFromString(
   copy.invoke_prepare_options = false;  // Prepare here, not there
   s = ObjectRegistry::NewInstance()->NewUniqueObject<FileSystem>(id, &fs);
   if (s.ok()) {
-    auto* cfs = dynamic_cast<CloudFileSystem*>(fs.get());
+    auto* cfs = dynamic_cast<CloudFileSystemImpl*>(fs.get());
     assert(cfs);
     auto copts = cfs->GetOptions<CloudFileSystemOptions>();
     *copts = cloud_options;
@@ -591,7 +579,7 @@ Status CloudFileSystem::CreateFromString(
       s = cfs->ConfigureFromMap(copy, options);
     }
     if (s.ok() && config_options.invoke_prepare_options) {
-      auto env = cfs->NewCompositeEnvFromThis(copy.env);
+      auto env = NewCompositeEnvFromFs(cfs, copy.env);
       copy.invoke_prepare_options = config_options.invoke_prepare_options;
       copy.env = env.get();
       s = cfs->PrepareOptions(copy);
@@ -610,18 +598,18 @@ Status CloudFileSystem::CreateFromString(
 }
 
 #ifndef USE_AWS
-Status CloudFileSystem::NewAwsFileSystem(
+Status CloudFileSystemEnv::NewAwsFileSystem(
     const std::shared_ptr<FileSystem>& /*base_fs*/,
     const CloudFileSystemOptions& /*options*/,
     const std::shared_ptr<Logger>& /*logger*/, CloudFileSystem** /*cfs*/) {
   return Status::NotSupported("RocksDB Cloud not compiled with AWS support");
 }
 #else
-Status CloudFileSystem::NewAwsFileSystem(
+Status CloudFileSystemEnv::NewAwsFileSystem(
     const std::shared_ptr<FileSystem>& base_fs,
     const CloudFileSystemOptions& options,
     const std::shared_ptr<Logger>& logger, CloudFileSystem** cfs) {
-  CloudFileSystem::RegisterCloudObjects();
+  CloudFileSystemEnv::RegisterCloudObjects();
   // Dump out cloud fs options
   options.Dump(logger.get());
 
@@ -640,9 +628,8 @@ Status CloudFileSystem::NewAwsFileSystem(
 }
 #endif
 
-std::unique_ptr<Env> CloudFileSystem::NewCompositeEnv(
-    Env* env,
-    const std::shared_ptr<FileSystem>& fs) {
+std::unique_ptr<Env> CloudFileSystemEnv::NewCompositeEnv(
+    Env* env, const std::shared_ptr<FileSystem>& fs) {
   return std::make_unique<CompositeEnvWrapper>(env, fs);
 }
 
diff --git a/cloud/cloud_file_system_impl.cc b/cloud/cloud_file_system_impl.cc
index 259dd2ef17c..3d064242237 100644
--- a/cloud/cloud_file_system_impl.cc
+++ b/cloud/cloud_file_system_impl.cc
@@ -1,23 +1,20 @@
 // Copyright (c) 2017 Rockset.
+#include "rocksdb/utilities/options_type.h"
 #ifndef ROCKSDB_LITE
 
-#include "rocksdb/cloud/cloud_file_system_impl.h"
-
 #include <cinttypes>
 
-#include "rocksdb/cloud/cloud_file_deletion_scheduler.h"
 #include "cloud/cloud_log_controller_impl.h"
 #include "cloud/cloud_manifest.h"
 #include "cloud/cloud_scheduler.h"
 #include "cloud/filename.h"
 #include "cloud/manifest_reader.h"
-#include "db/db_impl/replication_codec.h"
-#include "env/composite_env_wrapper.h"
 #include "file/file_util.h"
 #include "file/filename.h"
 #include "file/writable_file_writer.h"
-#include "port/likely.h"
 #include "port/port_posix.h"
+#include "rocksdb/cloud/cloud_file_deletion_scheduler.h"
+#include "rocksdb/cloud/cloud_file_system_impl.h"
 #include "rocksdb/cloud/cloud_log_controller.h"
 #include "rocksdb/cloud/cloud_storage_provider.h"
 #include "rocksdb/db.h"
@@ -32,8 +29,13 @@ namespace ROCKSDB_NAMESPACE {
 
 CloudFileSystemImpl::CloudFileSystemImpl(
     const CloudFileSystemOptions& opts, const std::shared_ptr<FileSystem>& base,
-    const std::shared_ptr<Logger>& l)
-    : CloudFileSystem(opts, base, l), purger_is_running_(true) {
+    const std::shared_ptr<Logger>& logger)
+    : info_log_(logger),
+      cloud_fs_options(opts),
+      base_fs_(base),
+      purger_is_running_(true) {
+  RegisterOptions(&cloud_fs_options,
+                  &CloudFileSystemOptions::cloud_fs_option_type_info);
   if (opts.cloud_file_deletion_delay) {
     cloud_file_deletion_scheduler_ = CloudFileDeletionScheduler::Create(
         CloudScheduler::Get(), *opts.cloud_file_deletion_delay);
@@ -45,6 +47,8 @@ CloudFileSystemImpl::~CloudFileSystemImpl() {
     cloud_fs_options.cloud_log_controller->StopTailingStream();
   }
   StopPurger();
+  cloud_fs_options.cloud_log_controller.reset();
+  cloud_fs_options.storage_provider.reset();
 }
 
 IOStatus CloudFileSystemImpl::ExistsCloudObject(const std::string& fname) {
diff --git a/cloud/cloud_file_system_test.cc b/cloud/cloud_file_system_test.cc
index 59032aab352..ad3b72381d2 100644
--- a/cloud/cloud_file_system_test.cc
+++ b/cloud/cloud_file_system_test.cc
@@ -3,9 +3,9 @@
 #include "rocksdb/cloud/cloud_file_system.h"
 
 #include "cloud/cloud_log_controller_impl.h"
-#include "rocksdb/cloud/cloud_storage_provider_impl.h"
 #include "rocksdb/cloud/cloud_log_controller.h"
 #include "rocksdb/cloud/cloud_storage_provider.h"
+#include "rocksdb/cloud/cloud_storage_provider_impl.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/env.h"
 #include "test_util/testharness.h"
@@ -214,8 +214,8 @@ TEST(CloudFileSystemTest, DISABLED_ConfigureKinesisController) {
   ASSERT_OK(CloudFileSystem::CreateFromString(
       config_options, "id=aws; controller=kinesis; TEST=dbcloud:/test", &cfs));
   ASSERT_STREQ(cfs->Name(), "aws");
-  ASSERT_NE(cfs->GetLogController(), nullptr);
-  ASSERT_STREQ(cfs->GetLogController()->Name(),
+  ASSERT_NE(cfs->GetCloudFileSystemOptions().cloud_log_controller, nullptr);
+  ASSERT_STREQ(cfs->GetCloudFileSystemOptions().cloud_log_controller->Name(),
                CloudLogControllerImpl::kKinesis());
 #endif
 }
@@ -229,8 +229,8 @@ TEST(CloudFileSystemTest, ConfigureKafkaController) {
 #ifdef USE_KAFKA
   ASSERT_OK(s);
   ASSERT_NE(cfs, nullptr);
-  ASSERT_NE(cfs->GetLogController(), nullptr);
-  ASSERT_STREQ(cfs->GetLogController()->Name(),
+  ASSERT_NE(cfs->GetCloudFileSystemOptions().cloud_log_controller, nullptr);
+  ASSERT_STREQ(cfs->GetCloudFileSystemOptions().cloud_log_controller->Name(),
                CloudLogControllerImpl::kKafka());
 #else
   ASSERT_NOK(s);
@@ -244,4 +244,3 @@ int main(int argc, char** argv) {
   ::testing::InitGoogleTest(&argc, argv);
   return RUN_ALL_TESTS();
 }
-  
diff --git a/cloud/cloud_storage_provider.cc b/cloud/cloud_storage_provider.cc
index 483646bcd05..7bae847e270 100644
--- a/cloud/cloud_storage_provider.cc
+++ b/cloud/cloud_storage_provider.cc
@@ -4,20 +4,17 @@
 #include "rocksdb/cloud/cloud_storage_provider.h"
 
 #include <cinttypes>
-#include <mutex>
-#include <set>
 
-#include "rocksdb/cloud/cloud_file_system_impl.h"
-#include "rocksdb/cloud/cloud_storage_provider_impl.h"
 #include "cloud/filename.h"
 #include "file/filename.h"
 #include "rocksdb/cloud/cloud_file_system.h"
+#include "rocksdb/cloud/cloud_file_system_impl.h"
+#include "rocksdb/cloud/cloud_storage_provider_impl.h"
 #include "rocksdb/convenience.h"
 #include "rocksdb/env.h"
 #include "rocksdb/options.h"
 #include "rocksdb/status.h"
 #include "rocksdb/utilities/object_registry.h"
-#include "util/coding.h"
 #include "util/random.h"
 #include "util/string_util.h"
 
@@ -251,7 +248,8 @@ Status CloudStorageProvider::CreateFromString(
     provider->reset();
     return Status::OK();
   } else {
-    return ObjectRegistry::NewInstance()->NewSharedObject<CloudStorageProvider>(id, provider);
+    return ObjectRegistry::NewInstance()->NewSharedObject<CloudStorageProvider>(
+        id, provider);
   }
 }
 
@@ -289,7 +287,7 @@ Status CloudStorageProviderImpl::PrepareOptions(const ConfigOptions& options) {
 }
 
 CloudStorageProviderImpl::CloudStorageProviderImpl()
-  : rng_(std::make_unique<Random64>(time(nullptr))) {}
+    : rng_(std::make_unique<Random64>(time(nullptr))) {}
 
 CloudStorageProviderImpl::~CloudStorageProviderImpl() {}
 
@@ -315,8 +313,8 @@ IOStatus CloudStorageProviderImpl::GetCloudObject(
       local_destination + ".tmp-" + std::to_string(rng_->Next());
 
   uint64_t remote_size;
-  auto s = DoGetCloudObject(bucket_name, object_path, tmp_destination,
-                            &remote_size);
+  auto s =
+      DoGetCloudObject(bucket_name, object_path, tmp_destination, &remote_size);
   const IOOptions io_opts;
   IODebugContext* dbg = nullptr;
   if (!s.ok()) {
diff --git a/include/rocksdb/cloud/cloud_file_system.h b/include/rocksdb/cloud/cloud_file_system.h
index 6651e51dcd2..3a6c6e3531e 100644
--- a/include/rocksdb/cloud/cloud_file_system.h
+++ b/include/rocksdb/cloud/cloud_file_system.h
@@ -397,6 +397,10 @@ class CloudFileSystemOptions {
   // Default: 1 hour
   std::optional<std::chrono::seconds> cloud_file_deletion_delay;
 
+  // Type info map for this class.
+  static const std::unordered_map<std::string, OptionTypeInfo>
+      cloud_fs_option_type_info;
+
   CloudFileSystemOptions(
       CloudType _cloud_type = CloudType::kCloudAws,
       LogType _log_type = LogType::kLogKafka,
@@ -488,45 +492,15 @@ struct CloudManifestDelta {
   std::string epoch;  // epoch for the new manifest file
 };
 
-//
-// The Cloud file system
-//
-// NOTE: The AWS SDK must be initialized before the CloudFileSystem is
-// constructed, and remain active (Aws::ShutdownAPI() not called) as long as any
-// CloudFileSystem objects exist.
+/** Extension of the FileSystem API for "the Cloud" */
 class CloudFileSystem : public FileSystem {
- protected:
-  CloudFileSystemOptions cloud_fs_options;
-  std::shared_ptr<FileSystem> base_fs_;  // The underlying file system
-
-  // Creates a new CompositeEnv from "env" and "this".
-  // The returned Env must not outlive "this"
-  std::unique_ptr<Env> NewCompositeEnvFromThis(Env* env);
-
-  CloudFileSystem(const CloudFileSystemOptions& options,
-                  const std::shared_ptr<FileSystem>& base,
-                  const std::shared_ptr<Logger>& logger);
-
  public:
-  mutable std::shared_ptr<Logger> info_log_;  // informational messages
-
-  virtual ~CloudFileSystem();
-
-  static void RegisterCloudObjects(const std::string& mode = "");
-  static Status CreateFromString(const ConfigOptions& config_options,
-                                 const std::string& id,
-                                 std::unique_ptr<CloudFileSystem>* fs);
-  static Status CreateFromString(const ConfigOptions& config_options,
-                                 const std::string& id,
-                                 const CloudFileSystemOptions& cloud_options,
-                                 std::unique_ptr<CloudFileSystem>* fs);
   static const char* kCloud() { return "cloud"; }
   static const char* kAws() { return "aws"; }
-  virtual const char* Name() const { return "cloud-env"; }
+
   // Returns the underlying file system
-  const std::shared_ptr<FileSystem>& GetBaseFileSystem() const {
-    return base_fs_;
-  }
+  virtual const std::shared_ptr<FileSystem>& GetBaseFileSystem() const = 0;
+
   virtual IOStatus PreloadCloudManifest(const std::string& local_dbname) = 0;
   // This method will migrate the database that is using pure RocksDB into
   // RocksDB-Cloud. Call this before opening the database with RocksDB-Cloud.
@@ -550,52 +524,6 @@ class CloudFileSystem : public FileSystem {
   virtual IOStatus DeleteDbid(const std::string& bucket_prefix,
                               const std::string& dbid) = 0;
 
-  Logger* GetLogger() const { return info_log_.get(); }
-  const std::shared_ptr<CloudStorageProvider>& GetStorageProvider() const {
-    return cloud_fs_options.storage_provider;
-  }
-
-  const std::shared_ptr<CloudLogController>& GetLogController() const {
-    return cloud_fs_options.cloud_log_controller;
-  }
-
-  // The SrcBucketName identifies the cloud storage bucket and
-  // GetSrcObjectPath specifies the path inside that bucket
-  // where data files reside. The specified bucket is used in
-  // a readonly mode by the associated DBCloud instance.
-  const std::string& GetSrcBucketName() const {
-    return cloud_fs_options.src_bucket.GetBucketName();
-  }
-  const std::string& GetSrcObjectPath() const {
-    return cloud_fs_options.src_bucket.GetObjectPath();
-  }
-  bool HasSrcBucket() const { return cloud_fs_options.src_bucket.IsValid(); }
-
-  // The DestBucketName identifies the cloud storage bucket and
-  // GetDestObjectPath specifies the path inside that bucket
-  // where data files reside. The associated DBCloud instance
-  // writes newly created files to this bucket.
-  const std::string& GetDestBucketName() const {
-    return cloud_fs_options.dest_bucket.GetBucketName();
-  }
-  const std::string& GetDestObjectPath() const {
-    return cloud_fs_options.dest_bucket.GetObjectPath();
-  }
-
-  bool HasDestBucket() const { return cloud_fs_options.dest_bucket.IsValid(); }
-  bool SrcMatchesDest() const {
-    if (HasSrcBucket() && HasDestBucket()) {
-      return cloud_fs_options.src_bucket == cloud_fs_options.dest_bucket;
-    } else {
-      return false;
-    }
-  }
-
-  // returns the options used to create this object
-  const CloudFileSystemOptions& GetCloudFileSystemOptions() const {
-    return cloud_fs_options;
-  }
-
   // Deletes file from a destination bucket.
   virtual IOStatus DeleteCloudFileFromDest(const std::string& fname) = 0;
   // Copies a local file to a destination bucket.
@@ -663,6 +591,56 @@ class CloudFileSystem : public FileSystem {
       const std::string& dbname,
       const std::vector<std::string>& active_cookies) = 0;
 
+  // returns the options used to create this object
+  virtual const CloudFileSystemOptions& GetCloudFileSystemOptions() const = 0;
+
+  // The SrcBucketName identifies the cloud storage bucket and
+  // GetSrcObjectPath specifies the path inside that bucket
+  // where data files reside. The specified bucket is used in
+  // a readonly mode by the associated DBCloud instance.
+  virtual const std::string& GetSrcBucketName() const = 0;
+  virtual const std::string& GetSrcObjectPath() const = 0;
+  virtual bool HasSrcBucket() const = 0;
+
+  // The DestBucketName identifies the cloud storage bucket and
+  // GetDestObjectPath specifies the path inside that bucket
+  // where data files reside. The associated DBCloud instance
+  // writes newly created files to this bucket.
+  virtual const std::string& GetDestBucketName() const = 0;
+  virtual const std::string& GetDestObjectPath() const = 0;
+  virtual bool HasDestBucket() const = 0;
+
+  virtual bool SrcMatchesDest() const = 0;
+
+  // Get the storage provider for the FileSystem.
+  virtual const std::shared_ptr<CloudStorageProvider>& GetStorageProvider()
+      const = 0;
+
+  virtual Logger* GetLogger() const = 0;
+};
+
+//
+// The Cloud File System initialization/construction.
+//
+// NOTE: The AWS SDK must be initialized before the CloudFileSystem is
+// constructed, and remain active (Aws::ShutdownAPI() not called) as long as any
+// CloudFileSystem objects exist.
+class CloudFileSystemEnv {
+ public:
+  // Creates a new CompositeEnv from "env" and "this".
+  // The returned Env must not outlive "this"
+  static std::unique_ptr<Env> NewCompositeEnvFromFs(FileSystem* fs, Env* env);
+
+ public:
+  static void RegisterCloudObjects(const std::string& mode = "");
+  static Status CreateFromString(const ConfigOptions& config_options,
+                                 const std::string& id,
+                                 std::unique_ptr<CloudFileSystem>* fs);
+  static Status CreateFromString(const ConfigOptions& config_options,
+                                 const std::string& id,
+                                 const CloudFileSystemOptions& cloud_options,
+                                 std::unique_ptr<CloudFileSystem>* fs);
+
   // Create a new AWS file system.
   // src_bucket_name: bucket name suffix where db data is read from
   // src_object_prefix: all db objects in source bucket are prepended with this
diff --git a/include/rocksdb/cloud/cloud_file_system_impl.h b/include/rocksdb/cloud/cloud_file_system_impl.h
index 23d2873af9c..cb3d58ca3ce 100644
--- a/include/rocksdb/cloud/cloud_file_system_impl.h
+++ b/include/rocksdb/cloud/cloud_file_system_impl.h
@@ -4,8 +4,8 @@
 #include <atomic>
 #include <condition_variable>
 #include <mutex>
-#include <thread>
 #include <set>
+#include <thread>
 
 #include "rocksdb/cloud/cloud_file_system.h"
 #include "rocksdb/file_system.h"
@@ -23,9 +23,11 @@ class CloudFileDeletionScheduler;
 // The Cloud file system
 //
 class CloudFileSystemImpl : public CloudFileSystem {
-  friend class CloudFileSystem;
+  friend class CloudFileSystemEnv;
 
  public:
+  mutable std::shared_ptr<Logger> info_log_;  // informational messages
+
   static int RegisterAwsObjects(ObjectLibrary& library, const std::string& arg);
   // Constructor
   CloudFileSystemImpl(const CloudFileSystemOptions& options,
@@ -34,7 +36,12 @@ class CloudFileSystemImpl : public CloudFileSystem {
 
   virtual ~CloudFileSystemImpl();
   static const char* kClassName() { return kCloud(); }
-  virtual const char* Name() const override { return kClassName(); }
+
+  const std::shared_ptr<FileSystem>& GetBaseFileSystem() const override {
+    return base_fs_;
+  }
+
+  const char* Name() const override { return kClassName(); }
 
   IOStatus NewSequentialFile(const std::string& fname,
                              const FileOptions& file_opts,
@@ -285,6 +292,9 @@ class CloudFileSystemImpl : public CloudFileSystem {
 #endif
 
  protected:
+  CloudFileSystemOptions cloud_fs_options;
+  std::shared_ptr<FileSystem> base_fs_;  // The underlying file system
+
   Status CheckValidity() const;
   // Status TEST_Initialize(const std::string& name) override;
   // The pathname that contains a list of all db's inside a bucket.
@@ -367,6 +377,48 @@ class CloudFileSystemImpl : public CloudFileSystem {
   IOStatus DeleteLocalInvisibleFiles(
       const std::string& dbname,
       const std::vector<std::string>& active_cookies) override;
+
+ public:
+  // returns the options used to create this object
+  const CloudFileSystemOptions& GetCloudFileSystemOptions() const override {
+    return cloud_fs_options;
+  }
+
+  const std::string& GetSrcBucketName() const override {
+    return cloud_fs_options.src_bucket.GetBucketName();
+  }
+  const std::string& GetSrcObjectPath() const override {
+    return cloud_fs_options.src_bucket.GetObjectPath();
+  }
+  bool HasSrcBucket() const override {
+    return cloud_fs_options.src_bucket.IsValid();
+  }
+
+  const std::string& GetDestBucketName() const override {
+    return cloud_fs_options.dest_bucket.GetBucketName();
+  }
+  const std::string& GetDestObjectPath() const override {
+    return cloud_fs_options.dest_bucket.GetObjectPath();
+  }
+
+  bool HasDestBucket() const override {
+    return cloud_fs_options.dest_bucket.IsValid();
+  }
+  bool SrcMatchesDest() const override {
+    if (HasSrcBucket() && HasDestBucket()) {
+      return cloud_fs_options.src_bucket == cloud_fs_options.dest_bucket;
+    } else {
+      return false;
+    }
+  }
+
+  const std::shared_ptr<CloudStorageProvider>& GetStorageProvider()
+      const override {
+    return cloud_fs_options.storage_provider;
+  }
+
+  Logger* GetLogger() const override { return info_log_.get(); }
+
  private:
   // Files are invisibile if:
   // - It's CLOUDMANFIEST file and cookie is not active. NOTE: empty cookie is