diff --git a/db/corruption_test.cc b/db/corruption_test.cc
index 203c34fa40..da020967de 100644
--- a/db/corruption_test.cc
+++ b/db/corruption_test.cc
@@ -190,7 +190,6 @@ class CorruptionTest : public testing::Test {
     ASSERT_TRUE(s.ok()) << s.ToString();
     Options options;
     EnvOptions env_options;
-    options.file_system.reset(new LegacyFileSystemWrapper(options.env));
     ASSERT_NOK(VerifySstFileChecksum(options, env_options, fname));
   }
 
diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc
index 23b28bbcb4..af18ee8685 100644
--- a/db/db_impl/db_impl.cc
+++ b/db/db_impl/db_impl.cc
@@ -150,7 +150,7 @@ DBImpl::DBImpl(const DBOptions& options, const std::string& dbname,
       own_info_log_(options.info_log == nullptr),
       initial_db_options_(SanitizeOptions(dbname, options)),
       env_(initial_db_options_.env),
-      fs_(initial_db_options_.file_system),
+      fs_(initial_db_options_.env->GetFileSystem()),
       immutable_db_options_(initial_db_options_),
       mutable_db_options_(initial_db_options_),
       stats_(immutable_db_options_.statistics.get()),
@@ -3478,12 +3478,8 @@ Status DBImpl::Close() {
 Status DB::ListColumnFamilies(const DBOptions& db_options,
                               const std::string& name,
                               std::vector<std::string>* column_families) {
-  FileSystem* fs = db_options.file_system.get();
-  LegacyFileSystemWrapper legacy_fs(db_options.env);
-  if (!fs) {
-    fs = &legacy_fs;
-  }
-  return VersionSet::ListColumnFamilies(column_families, name, fs);
+  const std::shared_ptr<FileSystem>& fs = db_options.env->GetFileSystem();
+  return VersionSet::ListColumnFamilies(column_families, name, fs.get());
 }
 
 Snapshot::~Snapshot() {}
diff --git a/db/db_impl/db_impl_open.cc b/db/db_impl/db_impl_open.cc
index 09728b6ab6..a1ff1ed8e6 100644
--- a/db/db_impl/db_impl_open.cc
+++ b/db/db_impl/db_impl_open.cc
@@ -35,16 +35,8 @@ Options SanitizeOptions(const std::string& dbname, const Options& src) {
 DBOptions SanitizeOptions(const std::string& dbname, const DBOptions& src) {
   DBOptions result(src);
 
-  if (result.file_system == nullptr) {
-    if (result.env == Env::Default()) {
-      result.file_system = FileSystem::Default();
-    } else {
-      result.file_system.reset(new LegacyFileSystemWrapper(result.env));
-    }
-  } else {
-    if (result.env == nullptr) {
-      result.env = Env::Default();
-    }
+  if (result.env == nullptr) {
+    result.env = Env::Default();
   }
 
   // result.max_open_files means an "infinite" open files.
diff --git a/db/error_handler_fs_test.cc b/db/error_handler_fs_test.cc
index be5917e64c..9dbab3f1ea 100644
--- a/db/error_handler_fs_test.cc
+++ b/db/error_handler_fs_test.cc
@@ -44,7 +44,7 @@ class DBErrorHandlingFSTest : public DBTestBase {
 class DBErrorHandlingFS : public FileSystemWrapper {
  public:
   DBErrorHandlingFS()
-      : FileSystemWrapper(FileSystem::Default().get()),
+      : FileSystemWrapper(FileSystem::Default()),
         trig_no_space(false),
         trig_io_error(false) {}
 
@@ -150,12 +150,13 @@ class ErrorHandlerFSListener : public EventListener {
 };
 
 TEST_F(DBErrorHandlingFSTest, FLushWriteError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   std::shared_ptr<ErrorHandlerFSListener> listener(
       new ErrorHandlerFSListener());
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.listeners.emplace_back(listener);
   Status s;
@@ -181,12 +182,13 @@ TEST_F(DBErrorHandlingFSTest, FLushWriteError) {
 }
 
 TEST_F(DBErrorHandlingFSTest, ManifestWriteError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   std::shared_ptr<ErrorHandlerFSListener> listener(
       new ErrorHandlerFSListener());
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.listeners.emplace_back(listener);
   Status s;
@@ -223,12 +225,13 @@ TEST_F(DBErrorHandlingFSTest, ManifestWriteError) {
 }
 
 TEST_F(DBErrorHandlingFSTest, DoubleManifestWriteError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   std::shared_ptr<ErrorHandlerFSListener> listener(
       new ErrorHandlerFSListener());
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.listeners.emplace_back(listener);
   Status s;
@@ -272,12 +275,13 @@ TEST_F(DBErrorHandlingFSTest, DoubleManifestWriteError) {
 }
 
 TEST_F(DBErrorHandlingFSTest, CompactionManifestWriteError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   std::shared_ptr<ErrorHandlerFSListener> listener(
       new ErrorHandlerFSListener());
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.level0_file_num_compaction_trigger = 2;
   options.listeners.emplace_back(listener);
@@ -344,12 +348,13 @@ TEST_F(DBErrorHandlingFSTest, CompactionManifestWriteError) {
 }
 
 TEST_F(DBErrorHandlingFSTest, CompactionWriteError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   std::shared_ptr<ErrorHandlerFSListener> listener(
       new ErrorHandlerFSListener());
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.level0_file_num_compaction_trigger = 2;
   options.listeners.emplace_back(listener);
@@ -387,10 +392,11 @@ TEST_F(DBErrorHandlingFSTest, CompactionWriteError) {
 }
 
 TEST_F(DBErrorHandlingFSTest, CorruptionError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.level0_file_num_compaction_trigger = 2;
   Status s;
@@ -426,12 +432,13 @@ TEST_F(DBErrorHandlingFSTest, CorruptionError) {
 }
 
 TEST_F(DBErrorHandlingFSTest, AutoRecoverFlushError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   std::shared_ptr<ErrorHandlerFSListener> listener(
       new ErrorHandlerFSListener());
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.listeners.emplace_back(listener);
   Status s;
@@ -460,12 +467,13 @@ TEST_F(DBErrorHandlingFSTest, AutoRecoverFlushError) {
 }
 
 TEST_F(DBErrorHandlingFSTest, FailRecoverFlushError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   std::shared_ptr<ErrorHandlerFSListener> listener(
       new ErrorHandlerFSListener());
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.listeners.emplace_back(listener);
   Status s;
@@ -487,12 +495,13 @@ TEST_F(DBErrorHandlingFSTest, FailRecoverFlushError) {
 }
 
 TEST_F(DBErrorHandlingFSTest, WALWriteError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   std::shared_ptr<ErrorHandlerFSListener> listener(
       new ErrorHandlerFSListener());
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.writable_file_max_buffer_size = 32768;
   options.listeners.emplace_back(listener);
@@ -558,12 +567,13 @@ TEST_F(DBErrorHandlingFSTest, WALWriteError) {
 }
 
 TEST_F(DBErrorHandlingFSTest, MultiCFWALWriteError) {
-  FaultInjectionTestFS* fault_fs =
-      new FaultInjectionTestFS(FileSystem::Default().get());
+  std::shared_ptr<FaultInjectionTestFS> fault_fs(
+      new FaultInjectionTestFS(FileSystem::Default()));
+  std::unique_ptr<Env> fault_fs_env(NewCompositeEnv(fault_fs));
   std::shared_ptr<ErrorHandlerFSListener> listener(
       new ErrorHandlerFSListener());
   Options options = GetDefaultOptions();
-  options.file_system.reset(fault_fs);
+  options.env = fault_fs_env.get();
   options.create_if_missing = true;
   options.writable_file_max_buffer_size = 32768;
   options.listeners.emplace_back(listener);
@@ -643,6 +653,7 @@ TEST_F(DBErrorHandlingFSTest, MultiCFWALWriteError) {
 
 TEST_F(DBErrorHandlingFSTest, MultiDBCompactionError) {
   FaultInjectionTestEnv* def_env = new FaultInjectionTestEnv(Env::Default());
+  std::vector<std::unique_ptr<Env>> fault_envs;
   std::vector<FaultInjectionTestFS*> fault_fs;
   std::vector<Options> options;
   std::vector<std::shared_ptr<ErrorHandlerFSListener>> listener;
@@ -654,12 +665,13 @@ TEST_F(DBErrorHandlingFSTest, MultiDBCompactionError) {
   for (auto i = 0; i < kNumDbInstances; ++i) {
     listener.emplace_back(new ErrorHandlerFSListener());
     options.emplace_back(GetDefaultOptions());
-    fault_fs.emplace_back(
-        new FaultInjectionTestFS(FileSystem::Default().get()));
+    fault_fs.emplace_back(new FaultInjectionTestFS(FileSystem::Default()));
+    std::shared_ptr<FileSystem> fs(fault_fs.back());
+    fault_envs.emplace_back(new CompositeEnvWrapper(def_env, fs));
+    options[i].env = fault_envs.back().get();
     options[i].create_if_missing = true;
     options[i].level0_file_num_compaction_trigger = 2;
     options[i].writable_file_max_buffer_size = 32768;
-    options[i].file_system.reset(fault_fs[i]);
     options[i].listeners.emplace_back(listener[i]);
     options[i].sst_file_manager = sfm;
     DB* dbptr;
@@ -742,6 +754,7 @@ TEST_F(DBErrorHandlingFSTest, MultiDBCompactionError) {
 
 TEST_F(DBErrorHandlingFSTest, MultiDBVariousErrors) {
   FaultInjectionTestEnv* def_env = new FaultInjectionTestEnv(Env::Default());
+  std::vector<std::unique_ptr<Env>> fault_envs;
   std::vector<FaultInjectionTestFS*> fault_fs;
   std::vector<Options> options;
   std::vector<std::shared_ptr<ErrorHandlerFSListener>> listener;
@@ -753,12 +766,13 @@ TEST_F(DBErrorHandlingFSTest, MultiDBVariousErrors) {
   for (auto i = 0; i < kNumDbInstances; ++i) {
     listener.emplace_back(new ErrorHandlerFSListener());
     options.emplace_back(GetDefaultOptions());
-    fault_fs.emplace_back(
-        new FaultInjectionTestFS(FileSystem::Default().get()));
+    fault_fs.emplace_back(new FaultInjectionTestFS(FileSystem::Default()));
+    std::shared_ptr<FileSystem> fs(fault_fs.back());
+    fault_envs.emplace_back(new CompositeEnvWrapper(def_env, fs));
+    options[i].env = fault_envs.back().get();
     options[i].create_if_missing = true;
     options[i].level0_file_num_compaction_trigger = 2;
     options[i].writable_file_max_buffer_size = 32768;
-    options[i].file_system.reset(fault_fs[i]);
     options[i].listeners.emplace_back(listener[i]);
     options[i].sst_file_manager = sfm;
     DB* dbptr;
diff --git a/db/memtable_list_test.cc b/db/memtable_list_test.cc
index 5cc3d50651..8981c12bad 100644
--- a/db/memtable_list_test.cc
+++ b/db/memtable_list_test.cc
@@ -92,7 +92,6 @@ class MemTableListTest : public testing::Test {
     CreateDB();
     // Create a mock VersionSet
     DBOptions db_options;
-    db_options.file_system = FileSystem::Default();
     ImmutableDBOptions immutable_db_options(db_options);
     EnvOptions env_options;
     std::shared_ptr<Cache> table_cache(NewLRUCache(50000, 16));
@@ -139,7 +138,6 @@ class MemTableListTest : public testing::Test {
     CreateDB();
     // Create a mock VersionSet
     DBOptions db_options;
-    db_options.file_system.reset(new LegacyFileSystemWrapper(db_options.env));
 
     ImmutableDBOptions immutable_db_options(db_options);
     EnvOptions env_options;
diff --git a/db/repair.cc b/db/repair.cc
index 383ffe3a4b..b2df33f068 100644
--- a/db/repair.cc
+++ b/db/repair.cc
@@ -672,10 +672,6 @@ Status RepairDB(const std::string& dbname, const DBOptions& db_options,
 
 Status RepairDB(const std::string& dbname, const Options& options) {
   Options opts(options);
-  if (opts.file_system == nullptr) {
-    opts.file_system.reset(new LegacyFileSystemWrapper(opts.env));
-    ;
-  }
 
   DBOptions db_options(opts);
   ColumnFamilyOptions cf_options(opts);
diff --git a/db/version_set.cc b/db/version_set.cc
index 4d21d22b36..b5637c753d 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -4928,9 +4928,10 @@ Status VersionSet::DumpManifest(Options& options, std::string& dscname,
   Status s;
   {
     std::unique_ptr<FSSequentialFile> file;
-    s = options.file_system->NewSequentialFile(
+    const std::shared_ptr<FileSystem>& fs = options.env->GetFileSystem();
+    s = fs->NewSequentialFile(
         dscname,
-        options.file_system->OptimizeForManifestRead(file_options_), &file,
+        fs->OptimizeForManifestRead(file_options_), &file,
         nullptr);
     if (!s.ok()) {
       return s;
diff --git a/env/composite_env_wrapper.h b/env/composite_env_wrapper.h
index 24ac9e9de8..936d54f718 100644
--- a/env/composite_env_wrapper.h
+++ b/env/composite_env_wrapper.h
@@ -291,20 +291,18 @@ class CompositeEnvWrapper : public Env {
  public:
   // Initialize a CompositeEnvWrapper that delegates all thread/time related
   // calls to env, and all file operations to fs
-  explicit CompositeEnvWrapper(Env* env, FileSystem* fs)
-      : env_target_(env), fs_env_target_(fs) {}
+  explicit CompositeEnvWrapper(Env* env, std::shared_ptr<FileSystem> fs)
+      : Env(fs), env_target_(env) {}
   ~CompositeEnvWrapper() {}
 
   // Return the target to which this Env forwards all calls
   Env* env_target() const { return env_target_; }
 
-  FileSystem* fs_env_target() const { return fs_env_target_; }
-
   Status RegisterDbPaths(const std::vector<std::string>& paths) override {
-    return fs_env_target_->RegisterDbPaths(paths);
+    return file_system_->RegisterDbPaths(paths);
   }
   Status UnregisterDbPaths(const std::vector<std::string>& paths) override {
-    return fs_env_target_->UnregisterDbPaths(paths);
+    return file_system_->UnregisterDbPaths(paths);
   }
 
   // The following text is boilerplate that forwards all methods to target()
@@ -315,7 +313,7 @@ class CompositeEnvWrapper : public Env {
     std::unique_ptr<FSSequentialFile> file;
     Status status;
     status =
-        fs_env_target_->NewSequentialFile(f, FileOptions(options), &file, &dbg);
+        file_system_->NewSequentialFile(f, FileOptions(options), &file, &dbg);
     if (status.ok()) {
       r->reset(new CompositeSequentialFileWrapper(file));
     }
@@ -327,8 +325,8 @@ class CompositeEnvWrapper : public Env {
     IODebugContext dbg;
     std::unique_ptr<FSRandomAccessFile> file;
     Status status;
-    status = fs_env_target_->NewRandomAccessFile(f, FileOptions(options), &file,
-                                                 &dbg);
+    status =
+        file_system_->NewRandomAccessFile(f, FileOptions(options), &file, &dbg);
     if (status.ok()) {
       r->reset(new CompositeRandomAccessFileWrapper(file));
     }
@@ -340,7 +338,7 @@ class CompositeEnvWrapper : public Env {
     std::unique_ptr<FSWritableFile> file;
     Status status;
     status =
-        fs_env_target_->NewWritableFile(f, FileOptions(options), &file, &dbg);
+        file_system_->NewWritableFile(f, FileOptions(options), &file, &dbg);
     if (status.ok()) {
       r->reset(new CompositeWritableFileWrapper(file));
     }
@@ -352,8 +350,8 @@ class CompositeEnvWrapper : public Env {
     IODebugContext dbg;
     Status status;
     std::unique_ptr<FSWritableFile> file;
-    status = fs_env_target_->ReopenWritableFile(fname, FileOptions(options),
-                                                &file, &dbg);
+    status = file_system_->ReopenWritableFile(fname, FileOptions(options),
+                                              &file, &dbg);
     if (status.ok()) {
       result->reset(new CompositeWritableFileWrapper(file));
     }
@@ -366,8 +364,8 @@ class CompositeEnvWrapper : public Env {
     IODebugContext dbg;
     Status status;
     std::unique_ptr<FSWritableFile> file;
-    status = fs_env_target_->ReuseWritableFile(
-        fname, old_fname, FileOptions(options), &file, &dbg);
+    status = file_system_->ReuseWritableFile(fname, old_fname,
+                                             FileOptions(options), &file, &dbg);
     if (status.ok()) {
       r->reset(new CompositeWritableFileWrapper(file));
     }
@@ -379,8 +377,8 @@ class CompositeEnvWrapper : public Env {
     IODebugContext dbg;
     std::unique_ptr<FSRandomRWFile> file;
     Status status;
-    status = fs_env_target_->NewRandomRWFile(fname, FileOptions(options), &file,
-                                             &dbg);
+    status =
+        file_system_->NewRandomRWFile(fname, FileOptions(options), &file, &dbg);
     if (status.ok()) {
       result->reset(new CompositeRandomRWFileWrapper(file));
     }
@@ -389,7 +387,7 @@ class CompositeEnvWrapper : public Env {
   Status NewMemoryMappedFileBuffer(
       const std::string& fname,
       std::unique_ptr<MemoryMappedFileBuffer>* result) override {
-    return fs_env_target_->NewMemoryMappedFileBuffer(fname, result);
+    return file_system_->NewMemoryMappedFileBuffer(fname, result);
   }
   Status NewDirectory(const std::string& name,
                       std::unique_ptr<Directory>* result) override {
@@ -397,7 +395,7 @@ class CompositeEnvWrapper : public Env {
     IODebugContext dbg;
     std::unique_ptr<FSDirectory> dir;
     Status status;
-    status = fs_env_target_->NewDirectory(name, io_opts, &dir, &dbg);
+    status = file_system_->NewDirectory(name, io_opts, &dir, &dbg);
     if (status.ok()) {
       result->reset(new CompositeDirectoryWrapper(dir));
     }
@@ -406,102 +404,108 @@ class CompositeEnvWrapper : public Env {
   Status FileExists(const std::string& f) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->FileExists(f, io_opts, &dbg);
+    return file_system_->FileExists(f, io_opts, &dbg);
   }
   Status GetChildren(const std::string& dir,
                      std::vector<std::string>* r) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->GetChildren(dir, io_opts, r, &dbg);
+    return file_system_->GetChildren(dir, io_opts, r, &dbg);
   }
   Status GetChildrenFileAttributes(
       const std::string& dir, std::vector<FileAttributes>* result) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->GetChildrenFileAttributes(dir, io_opts, result,
-                                                     &dbg);
+    return file_system_->GetChildrenFileAttributes(dir, io_opts, result, &dbg);
   }
   Status DeleteFile(const std::string& f) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->DeleteFile(f, io_opts, &dbg);
+    return file_system_->DeleteFile(f, io_opts, &dbg);
   }
   Status Truncate(const std::string& fname, size_t size) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->Truncate(fname, size, io_opts, &dbg);
+    return file_system_->Truncate(fname, size, io_opts, &dbg);
   }
   Status CreateDir(const std::string& d) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->CreateDir(d, io_opts, &dbg);
+    return file_system_->CreateDir(d, io_opts, &dbg);
   }
   Status CreateDirIfMissing(const std::string& d) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->CreateDirIfMissing(d, io_opts, &dbg);
+    return file_system_->CreateDirIfMissing(d, io_opts, &dbg);
   }
   Status DeleteDir(const std::string& d) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->DeleteDir(d, io_opts, &dbg);
+    return file_system_->DeleteDir(d, io_opts, &dbg);
   }
   Status GetFileSize(const std::string& f, uint64_t* s) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->GetFileSize(f, io_opts, s, &dbg);
+    return file_system_->GetFileSize(f, io_opts, s, &dbg);
   }
 
   Status GetFileModificationTime(const std::string& fname,
                                  uint64_t* file_mtime) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->GetFileModificationTime(fname, io_opts, file_mtime,
-                                                   &dbg);
+    return file_system_->GetFileModificationTime(fname, io_opts, file_mtime,
+                                                 &dbg);
   }
 
   Status RenameFile(const std::string& s, const std::string& t) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->RenameFile(s, t, io_opts, &dbg);
+    return file_system_->RenameFile(s, t, io_opts, &dbg);
   }
 
   Status LinkFile(const std::string& s, const std::string& t) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->LinkFile(s, t, io_opts, &dbg);
+    return file_system_->LinkFile(s, t, io_opts, &dbg);
   }
 
   Status NumFileLinks(const std::string& fname, uint64_t* count) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->NumFileLinks(fname, io_opts, count, &dbg);
+    return file_system_->NumFileLinks(fname, io_opts, count, &dbg);
   }
 
   Status AreFilesSame(const std::string& first, const std::string& second,
                       bool* res) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->AreFilesSame(first, second, io_opts, res, &dbg);
+    return file_system_->AreFilesSame(first, second, io_opts, res, &dbg);
   }
 
   Status LockFile(const std::string& f, FileLock** l) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->LockFile(f, io_opts, l, &dbg);
+    return file_system_->LockFile(f, io_opts, l, &dbg);
   }
 
   Status UnlockFile(FileLock* l) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->UnlockFile(l, io_opts, &dbg);
+    return file_system_->UnlockFile(l, io_opts, &dbg);
   }
 
   Status GetAbsolutePath(const std::string& db_path,
                          std::string* output_path) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->GetAbsolutePath(db_path, io_opts, output_path, &dbg);
+    return file_system_->GetAbsolutePath(db_path, io_opts, output_path, &dbg);
+  }
+
+  Status NewLogger(const std::string& fname,
+                   std::shared_ptr<Logger>* result) override {
+    IOOptions io_opts;
+    IODebugContext dbg;
+    return file_system_->NewLogger(fname, io_opts, result, &dbg);
   }
 
 #if !defined(OS_WIN) && !defined(ROCKSDB_NO_DYNAMIC_EXTENSION)
@@ -531,10 +535,6 @@ class CompositeEnvWrapper : public Env {
   Status GetTestDirectory(std::string* path) override {
     return env_target_->GetTestDirectory(path);
   }
-  Status NewLogger(const std::string& fname,
-                   std::shared_ptr<Logger>* result) override {
-    return env_target_->NewLogger(fname, result);
-  }
   uint64_t NowMicros() override { return env_target_->NowMicros(); }
   uint64_t NowNanos() override { return env_target_->NowNanos(); }
   uint64_t NowCPUNanos() override { return env_target_->NowCPUNanos(); }
@@ -590,46 +590,41 @@ class CompositeEnvWrapper : public Env {
   }
 
   EnvOptions OptimizeForLogRead(const EnvOptions& env_options) const override {
-    return fs_env_target_->OptimizeForLogRead(FileOptions(env_options));
+    return file_system_->OptimizeForLogRead(FileOptions(env_options));
   }
   EnvOptions OptimizeForManifestRead(
       const EnvOptions& env_options) const override {
-    return fs_env_target_->OptimizeForManifestRead(
-                                FileOptions(env_options));
+    return file_system_->OptimizeForManifestRead(FileOptions(env_options));
   }
   EnvOptions OptimizeForLogWrite(const EnvOptions& env_options,
                                  const DBOptions& db_options) const override {
-    return fs_env_target_->OptimizeForLogWrite(FileOptions(env_options),
-                                               db_options);
+    return file_system_->OptimizeForLogWrite(FileOptions(env_options),
+                                             db_options);
   }
   EnvOptions OptimizeForManifestWrite(
       const EnvOptions& env_options) const override {
-    return fs_env_target_->OptimizeForManifestWrite(
-                                FileOptions(env_options));
+    return file_system_->OptimizeForManifestWrite(FileOptions(env_options));
   }
   EnvOptions OptimizeForCompactionTableWrite(
       const EnvOptions& env_options,
       const ImmutableDBOptions& immutable_ops) const override {
-    return fs_env_target_->OptimizeForCompactionTableWrite(
-                                FileOptions(env_options),
-                                immutable_ops);
+    return file_system_->OptimizeForCompactionTableWrite(
+        FileOptions(env_options), immutable_ops);
   }
   EnvOptions OptimizeForCompactionTableRead(
       const EnvOptions& env_options,
       const ImmutableDBOptions& db_options) const override {
-    return fs_env_target_->OptimizeForCompactionTableRead(
-                                FileOptions(env_options),
-                                db_options);
+    return file_system_->OptimizeForCompactionTableRead(
+        FileOptions(env_options), db_options);
   }
   Status GetFreeSpace(const std::string& path, uint64_t* diskfree) override {
     IOOptions io_opts;
     IODebugContext dbg;
-    return fs_env_target_->GetFreeSpace(path, io_opts, diskfree, &dbg);
+    return file_system_->GetFreeSpace(path, io_opts, diskfree, &dbg);
   }
 
  private:
   Env* env_target_;
-  FileSystem* fs_env_target_;
 };
 
 class LegacySequentialFileWrapper : public FSSequentialFile {
@@ -1067,6 +1062,10 @@ class LegacyFileSystemWrapper : public FileSystem {
     return status_to_io_status(target_->NewLogger(fname, result));
   }
 
+  void SanitizeFileOptions(FileOptions* opts) const override {
+    target_->SanitizeEnvOptions(opts);
+  }
+
   FileOptions OptimizeForLogRead(
                   const FileOptions& file_options) const override {
     return target_->OptimizeForLogRead(file_options);
diff --git a/env/env.cc b/env/env.cc
index 70f4b29f7a..c6eef1448a 100644
--- a/env/env.cc
+++ b/env/env.cc
@@ -22,6 +22,14 @@
 
 namespace ROCKSDB_NAMESPACE {
 
+Env::Env() : thread_status_updater_(nullptr) {
+  file_system_ = std::make_shared<LegacyFileSystemWrapper>(this);
+}
+
+Env::Env(std::shared_ptr<FileSystem> fs)
+  : thread_status_updater_(nullptr),
+    file_system_(fs) {}
+
 Env::~Env() {
 }
 
@@ -472,4 +480,14 @@ Status NewEnvLogger(const std::string& fname, Env* env,
   return Status::OK();
 }
 
+const std::shared_ptr<FileSystem>& Env::GetFileSystem() const {
+  return file_system_;
+}
+
+#ifdef OS_WIN
+std::unique_ptr<Env> NewCompositeEnv(std::shared_ptr<FileSystem> fs) {
+  return std::unique_ptr<Env>(new CompositeEnvWrapper(Env::Default(), fs));
+}
+#endif
+
 }  // namespace ROCKSDB_NAMESPACE
diff --git a/env/env_posix.cc b/env/env_posix.cc
index 861fbcf627..d899263867 100644
--- a/env/env_posix.cc
+++ b/env/env_posix.cc
@@ -128,21 +128,25 @@ class PosixDynamicLibrary : public DynamicLibrary {
 
 class PosixEnv : public CompositeEnvWrapper {
  public:
-  PosixEnv();
+  // This constructor is for constructing non-default Envs, mainly by
+  // NewCompositeEnv(). It allows new instances to share the same
+  // threadpool and other resources as the default Env, while allowing
+  // a non-default FileSystem implementation
+  PosixEnv(const PosixEnv* default_env, std::shared_ptr<FileSystem> fs);
 
   ~PosixEnv() override {
-    for (const auto tid : threads_to_join_) {
-      pthread_join(tid, nullptr);
-    }
-    for (int pool_id = 0; pool_id < Env::Priority::TOTAL; ++pool_id) {
-      thread_pools_[pool_id].JoinAllThreads();
-    }
-    // Delete the thread_status_updater_ only when the current Env is not
-    // Env::Default().  This is to avoid the free-after-use error when
-    // Env::Default() is destructed while some other child threads are
-    // still trying to update thread status.
-    if (this != Env::Default()) {
-      delete thread_status_updater_;
+    if (this == Env::Default()) {
+      for (const auto tid : threads_to_join_) {
+        pthread_join(tid, nullptr);
+      }
+      for (int pool_id = 0; pool_id < Env::Priority::TOTAL; ++pool_id) {
+        thread_pools_[pool_id].JoinAllThreads();
+      }
+      // Do not delete the thread_status_updater_ in order to avoid the
+      // free after use when Env::Default() is destructed while some other
+      // child threads are still trying to update thread status. All
+      // PosixEnv instances use the same thread_status_updater_, so never
+      // explicitly delete it.
     }
   }
 
@@ -252,34 +256,6 @@ class PosixEnv : public CompositeEnvWrapper {
 
   uint64_t GetThreadID() const override { return gettid(pthread_self()); }
 
-  Status NewLogger(const std::string& fname,
-                   std::shared_ptr<Logger>* result) override {
-    FILE* f;
-    {
-      IOSTATS_TIMER_GUARD(open_nanos);
-      f = fopen(fname.c_str(),
-                "w"
-#ifdef __GLIBC_PREREQ
-#if __GLIBC_PREREQ(2, 7)
-                "e"  // glibc extension to enable O_CLOEXEC
-#endif
-#endif
-      );
-    }
-    if (f == nullptr) {
-      result->reset();
-      return IOError("when fopen a file for new logger", fname, errno);
-    } else {
-      int fd = fileno(f);
-#ifdef ROCKSDB_FALLOCATE_PRESENT
-      fallocate(fd, FALLOC_FL_KEEP_SIZE, 0, 4 * 1024);
-#endif
-      SetFD_CLOEXEC(fd, nullptr);
-      result->reset(new PosixLogger(f, &PosixEnv::gettid, this));
-      return Status::OK();
-    }
-  }
-
   uint64_t NowMicros() override {
     struct timeval tv;
     gettimeofday(&tv, nullptr);
@@ -406,18 +382,34 @@ class PosixEnv : public CompositeEnvWrapper {
   }
 
  private:
-  std::vector<ThreadPoolImpl> thread_pools_;
-  pthread_mutex_t mu_;
-  std::vector<pthread_t> threads_to_join_;
+  friend Env* Env::Default();
+  // Constructs the default Env, a singleton
+  PosixEnv();
+
+  // The below 4 members are only used by the default PosixEnv instance.
+  // Non-default instances simply maintain references to the backing
+  // members in te default instance
+  std::vector<ThreadPoolImpl> thread_pools_storage_;
+  pthread_mutex_t mu_storage_;
+  std::vector<pthread_t> threads_to_join_storage_;
+  bool allow_non_owner_access_storage_;
+
+  std::vector<ThreadPoolImpl>& thread_pools_;
+  pthread_mutex_t& mu_;
+  std::vector<pthread_t>& threads_to_join_;
   // If true, allow non owner read access for db files. Otherwise, non-owner
   //  has no access to db files.
-  bool allow_non_owner_access_;
+  bool& allow_non_owner_access_;
 };
 
 PosixEnv::PosixEnv()
-    : CompositeEnvWrapper(this, FileSystem::Default().get()),
-      thread_pools_(Priority::TOTAL),
-      allow_non_owner_access_(true) {
+    : CompositeEnvWrapper(this, FileSystem::Default()),
+      thread_pools_storage_(Priority::TOTAL),
+      allow_non_owner_access_storage_(true),
+      thread_pools_(thread_pools_storage_),
+      mu_(mu_storage_),
+      threads_to_join_(threads_to_join_storage_),
+      allow_non_owner_access_(allow_non_owner_access_storage_) {
   ThreadPoolImpl::PthreadCall("mutex_init", pthread_mutex_init(&mu_, nullptr));
   for (int pool_id = 0; pool_id < Env::Priority::TOTAL; ++pool_id) {
     thread_pools_[pool_id].SetThreadPriority(
@@ -428,6 +420,15 @@ PosixEnv::PosixEnv()
   thread_status_updater_ = CreateThreadStatusUpdater();
 }
 
+PosixEnv::PosixEnv(const PosixEnv* default_env, std::shared_ptr<FileSystem> fs)
+  : CompositeEnvWrapper(this, fs),
+    thread_pools_(default_env->thread_pools_),
+    mu_(default_env->mu_),
+    threads_to_join_(default_env->threads_to_join_),
+    allow_non_owner_access_(default_env->allow_non_owner_access_) {
+  thread_status_updater_ = default_env->thread_status_updater_;
+}
+
 void PosixEnv::Schedule(void (*function)(void* arg1), void* arg, Priority pri,
                         void* tag, void (*unschedFunction)(void* arg)) {
   assert(pri >= Priority::BOTTOM && pri <= Priority::HIGH);
@@ -519,9 +520,12 @@ Env* Env::Default() {
   CompressionContextCache::InitSingleton();
   INIT_SYNC_POINT_SINGLETONS();
   static PosixEnv default_env;
-  static CompositeEnvWrapper composite_env(&default_env,
-                                           FileSystem::Default().get());
-  return &composite_env;
+  return &default_env;
+}
+
+std::unique_ptr<Env> NewCompositeEnv(std::shared_ptr<FileSystem> fs) {
+  PosixEnv* default_env = static_cast<PosixEnv*>(Env::Default());
+  return std::unique_ptr<Env>(new PosixEnv(default_env, fs));
 }
 
 }  // namespace ROCKSDB_NAMESPACE
diff --git a/env/env_test.cc b/env/env_test.cc
index 4f31dbd146..9ef5abbe39 100644
--- a/env/env_test.cc
+++ b/env/env_test.cc
@@ -35,6 +35,8 @@
 #include "port/malloc.h"
 #include "port/port.h"
 #include "rocksdb/env.h"
+#include "test_util/fault_injection_test_env.h"
+#include "test_util/fault_injection_test_fs.h"
 #include "test_util/sync_point.h"
 #include "test_util/testharness.h"
 #include "test_util/testutil.h"
@@ -1272,7 +1274,7 @@ TEST_F(EnvPosixTest, MultiReadNonAlignedLargeNum) {
     ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
   }
 }
-  
+
 // Only works in linux platforms
 #ifdef OS_WIN
 TEST_P(EnvPosixTestWithParam, DISABLED_InvalidateCache) {
@@ -1980,6 +1982,114 @@ INSTANTIATE_TEST_CASE_P(
     ::testing::Values(std::pair<Env*, bool>(chroot_env.get(), true)));
 #endif  // !defined(ROCKSDB_LITE) && !defined(OS_WIN)
 
+class EnvFSTestWithParam
+    : public ::testing::Test,
+      public ::testing::WithParamInterface<std::tuple<bool, bool, bool>> {
+ public:
+  EnvFSTestWithParam() {
+    bool env_non_null = std::get<0>(GetParam());
+    bool env_default = std::get<1>(GetParam());
+    bool fs_default = std::get<2>(GetParam());
+
+    env_ = env_non_null ? (env_default ? Env::Default() : nullptr) : nullptr;
+    fs_ = fs_default
+              ? FileSystem::Default()
+              : std::make_shared<FaultInjectionTestFS>(FileSystem::Default());
+    if (env_non_null && env_default && !fs_default) {
+      env_ptr_ = NewCompositeEnv(fs_);
+    }
+    if (env_non_null && !env_default && fs_default) {
+      env_ptr_ = std::unique_ptr<Env>(new FaultInjectionTestEnv(Env::Default()));
+      fs_.reset();
+    }
+    if (env_non_null && !env_default && !fs_default) {
+      env_ptr_.reset(new FaultInjectionTestEnv(Env::Default()));
+      composite_env_ptr_.reset(new CompositeEnvWrapper(env_ptr_.get(), fs_));
+      env_ = composite_env_ptr_.get();
+    } else {
+      env_ = env_ptr_.get();
+    }
+
+    dbname1_ = test::PerThreadDBPath("env_fs_test1");
+    dbname2_ = test::PerThreadDBPath("env_fs_test2");
+  }
+
+  ~EnvFSTestWithParam() = default;
+
+  Env* env_;
+  std::unique_ptr<Env> env_ptr_;
+  std::unique_ptr<Env> composite_env_ptr_;
+  std::shared_ptr<FileSystem> fs_;
+  std::string dbname1_;
+  std::string dbname2_;
+};
+
+TEST_P(EnvFSTestWithParam, OptionsTest) {
+  Options opts;
+  opts.env = env_;
+  opts.create_if_missing = true;
+  std::string dbname = dbname1_;
+
+  if (env_) {
+    if (fs_) {
+      ASSERT_EQ(fs_.get(), env_->GetFileSystem().get());
+    } else {
+      ASSERT_NE(FileSystem::Default().get(), env_->GetFileSystem().get());
+    }
+  }
+  for (int i = 0; i < 2; ++i) {
+    DB* db;
+    Status s = DB::Open(opts, dbname, &db);
+    ASSERT_OK(s);
+
+    WriteOptions wo;
+    db->Put(wo, "a", "a");
+    db->Flush(FlushOptions());
+    db->Put(wo, "b", "b");
+    db->Flush(FlushOptions());
+    db->CompactRange(CompactRangeOptions(), nullptr, nullptr);
+
+    std::string val;
+    ASSERT_OK(db->Get(ReadOptions(), "a", &val));
+    ASSERT_EQ("a", val);
+    ASSERT_OK(db->Get(ReadOptions(), "b", &val));
+    ASSERT_EQ("b", val);
+
+    db->Close();
+    delete db;
+    DestroyDB(dbname, opts);
+
+    dbname = dbname2_;
+  }
+}
+
+// The parameters are as follows -
+// 1. True means Options::env is non-null, false means null
+// 2. True means use Env::Default, false means custom
+// 3. True means use FileSystem::Default, false means custom
+INSTANTIATE_TEST_CASE_P(
+    EnvFSTest, EnvFSTestWithParam,
+    ::testing::Combine(::testing::Bool(), ::testing::Bool(),
+                       ::testing::Bool()));
+
+// This test ensures that default Env and those allocated by
+// NewCompositeEnv() all share the same threadpool
+TEST_F(EnvTest, MultipleCompositeEnv) {
+  std::shared_ptr<FaultInjectionTestFS> fs1 =
+    std::make_shared<FaultInjectionTestFS>(FileSystem::Default());
+  std::shared_ptr<FaultInjectionTestFS> fs2 =
+    std::make_shared<FaultInjectionTestFS>(FileSystem::Default());
+  std::unique_ptr<Env> env1 = NewCompositeEnv(fs1);
+  std::unique_ptr<Env> env2 = NewCompositeEnv(fs2);
+  Env::Default()->SetBackgroundThreads(8, Env::HIGH);
+  Env::Default()->SetBackgroundThreads(16, Env::LOW);
+
+  ASSERT_EQ(env1->GetBackgroundThreads(Env::LOW), 16);
+  ASSERT_EQ(env1->GetBackgroundThreads(Env::HIGH), 8);
+  ASSERT_EQ(env2->GetBackgroundThreads(Env::LOW), 16);
+  ASSERT_EQ(env2->GetBackgroundThreads(Env::HIGH), 8);
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {
diff --git a/env/file_system.cc b/env/file_system.cc
index 646e6e3d11..00840bc867 100644
--- a/env/file_system.cc
+++ b/env/file_system.cc
@@ -26,6 +26,18 @@ Status FileSystem::Load(const std::string& value,
   return s;
 }
 
+IOStatus FileSystem::ReuseWritableFile(const std::string& fname,
+                                       const std::string& old_fname,
+                                       const FileOptions& opts,
+                                       std::unique_ptr<FSWritableFile>* result,
+                                       IODebugContext* dbg) {
+  IOStatus s = RenameFile(old_fname, fname, opts.io_options, dbg);
+  if (!s.ok()) {
+    return s;
+  }
+  return NewWritableFile(fname, opts, result, dbg);
+}
+
 FileOptions FileSystem::OptimizeForLogRead(
               const FileOptions& file_options) const {
   FileOptions optimized_file_options(file_options);
diff --git a/env/fs_posix.cc b/env/fs_posix.cc
index 893827cdd2..4601472a07 100644
--- a/env/fs_posix.cc
+++ b/env/fs_posix.cc
@@ -47,6 +47,7 @@
 #include <set>
 #include <vector>
 
+#include "env/composite_env_wrapper.h"
 #include "env/io_posix.h"
 #include "logging/logging.h"
 #include "logging/posix_logger.h"
@@ -81,6 +82,10 @@ inline mode_t GetDBFileMode(bool allow_non_owner_access) {
   return allow_non_owner_access ? 0644 : 0600;
 }
 
+static uint64_t gettid() {
+  return Env::Default()->GetThreadID();
+}
+
 // list of pathnames that are locked
 // Only used for error message.
 struct LockHoldingInfo {
@@ -540,10 +545,34 @@ class PosixFileSystem : public FileSystem {
     return IOStatus::OK();
   }
 
-  IOStatus NewLogger(const std::string& /*fname*/, const IOOptions& /*opts*/,
-                     std::shared_ptr<ROCKSDB_NAMESPACE::Logger>* /*ptr*/,
-                     IODebugContext* /*dbg*/) override {
-    return IOStatus::NotSupported();
+  IOStatus NewLogger(const std::string& fname, const IOOptions& /*opts*/,
+                   std::shared_ptr<Logger>* result,
+                   IODebugContext* /*dbg*/) override {
+    FILE* f;
+    {
+      IOSTATS_TIMER_GUARD(open_nanos);
+      f = fopen(fname.c_str(),
+                "w"
+#ifdef __GLIBC_PREREQ
+#if __GLIBC_PREREQ(2, 7)
+                "e"  // glibc extension to enable O_CLOEXEC
+#endif
+#endif
+      );
+    }
+    if (f == nullptr) {
+      result->reset();
+      return status_to_io_status(
+              IOError("when fopen a file for new logger", fname, errno));
+    } else {
+      int fd = fileno(f);
+#ifdef ROCKSDB_FALLOCATE_PRESENT
+      fallocate(fd, FALLOC_FL_KEEP_SIZE, 0, 4 * 1024);
+#endif
+      SetFD_CLOEXEC(fd, nullptr);
+      result->reset(new PosixLogger(f, &gettid, Env::Default()));
+      return IOStatus::OK();
+    }
   }
 
   IOStatus FileExists(const std::string& fname, const IOOptions& /*opts*/,
diff --git a/include/rocksdb/env.h b/include/rocksdb/env.h
index a62eaa6662..fb9c0c9deb 100644
--- a/include/rocksdb/env.h
+++ b/include/rocksdb/env.h
@@ -57,6 +57,7 @@ struct MutableDBOptions;
 class RateLimiter;
 class ThreadStatusUpdater;
 struct ThreadStatus;
+class FileSystem;
 
 const size_t kDefaultPageSize = 4 * 1024;
 
@@ -140,7 +141,9 @@ class Env {
     uint64_t size_bytes;
   };
 
-  Env() : thread_status_updater_(nullptr) {}
+  Env();
+  // Construct an Env with a separate FileSystem implementation
+  Env(std::shared_ptr<FileSystem> fs);
   // No copying allowed
   Env(const Env&) = delete;
   void operator=(const Env&) = delete;
@@ -539,12 +542,19 @@ class Env {
 
   virtual void SanitizeEnvOptions(EnvOptions* /*env_opts*/) const {}
 
+  // Get the FileSystem implementation this Env was constructed with. It
+  // could be a fully implemented one, or a wrapper class around the Env
+  const std::shared_ptr<FileSystem>& GetFileSystem() const;
+
   // If you're adding methods here, remember to add them to EnvWrapper too.
 
  protected:
   // The pointer to an internal structure that will update the
   // status of each thread.
   ThreadStatusUpdater* thread_status_updater_;
+
+  // Pointer to the underlying FileSystem implementation
+  std::shared_ptr<FileSystem> file_system_;
 };
 
 // The factory function to construct a ThreadStatusUpdater.  Any Env
@@ -1603,4 +1613,6 @@ Env* NewTimedEnv(Env* base_env);
 Status NewEnvLogger(const std::string& fname, Env* env,
                     std::shared_ptr<Logger>* result);
 
+std::unique_ptr<Env> NewCompositeEnv(std::shared_ptr<FileSystem> fs);
+
 }  // namespace ROCKSDB_NAMESPACE
diff --git a/include/rocksdb/file_system.h b/include/rocksdb/file_system.h
index 8a7592d017..66b94680ed 100644
--- a/include/rocksdb/file_system.h
+++ b/include/rocksdb/file_system.h
@@ -102,6 +102,9 @@ struct FileOptions : EnvOptions {
 
   FileOptions(const EnvOptions& opts)
     : EnvOptions(opts) {}
+
+  FileOptions(const FileOptions& opts)
+    : EnvOptions(opts), io_options(opts.io_options) {}
 };
 
 // A structure to pass back some debugging information from the FileSystem
@@ -263,7 +266,7 @@ class FileSystem {
                                      const std::string& old_fname,
                                      const FileOptions& file_opts,
                                      std::unique_ptr<FSWritableFile>* result,
-                                     IODebugContext* dbg) = 0;
+                                     IODebugContext* dbg);
 
   // Open `fname` for random read and write, if file doesn't exist the file
   // will be created.  On success, stores a pointer to the new file in
@@ -465,6 +468,10 @@ class FileSystem {
                                    std::string* output_path,
                                    IODebugContext* dbg) = 0;
 
+  // Sanitize the FileOptions. Typically called by a FileOptions/EnvOptions
+  // copy constructor
+  virtual void SanitizeFileOptions(FileOptions* /*opts*/) const {}
+
   // OptimizeForLogRead will create a new FileOptions object that is a copy of
   // the FileOptions in the parameters, but is optimized for reading log files.
   virtual FileOptions OptimizeForLogRead(const FileOptions& file_options) const;
@@ -1001,11 +1008,13 @@ class FSDirectory {
 class FileSystemWrapper : public FileSystem {
  public:
   // Initialize an EnvWrapper that delegates all calls to *t
-  explicit FileSystemWrapper(FileSystem* t) : target_(t) {}
+  explicit FileSystemWrapper(std::shared_ptr<FileSystem> t) : target_(t) {}
   ~FileSystemWrapper() override {}
 
+  const char* Name() const override { return target_->Name(); }
+
   // Return the target to which this Env forwards all calls
-  FileSystem* target() const { return target_; }
+  FileSystem* target() const { return target_.get(); }
 
   // The following text is boilerplate that forwards all methods to target()
   IOStatus NewSequentialFile(const std::string& f,
@@ -1149,6 +1158,10 @@ class FileSystemWrapper : public FileSystem {
     return target_->NewLogger(fname, options, result, dbg);
   }
 
+  void SanitizeFileOptions(FileOptions* opts) const override {
+    target_->SanitizeFileOptions(opts);
+  }
+
   FileOptions OptimizeForLogRead(
                   const FileOptions& file_options) const override {
     return target_->OptimizeForLogRead(file_options);
@@ -1182,7 +1195,7 @@ class FileSystemWrapper : public FileSystem {
   }
 
  private:
-  FileSystem* target_;
+  std::shared_ptr<FileSystem> target_;
 };
 
 class FSSequentialFileWrapper : public FSSequentialFile {
diff --git a/include/rocksdb/options.h b/include/rocksdb/options.h
index 1976cdd1c6..87a0b94420 100644
--- a/include/rocksdb/options.h
+++ b/include/rocksdb/options.h
@@ -396,11 +396,6 @@ struct DBOptions {
   // Default: Env::Default()
   Env* env = Env::Default();
 
-  // Use the specified object to interact with the storage to
-  // read/write files. This is in addition to env. This option should be used
-  // if the desired storage subsystem provides a FileSystem implementation.
-  std::shared_ptr<FileSystem> file_system = nullptr;
-
   // Use to control write rate of flush and compaction. Flush has higher
   // priority than compaction. Rate limiting is disabled if nullptr.
   // If rate limiter is enabled, bytes_per_sync is set to 1MB by default.
diff --git a/options/db_options.cc b/options/db_options.cc
index 4788977863..2e572963f7 100644
--- a/options/db_options.cc
+++ b/options/db_options.cc
@@ -26,7 +26,7 @@ ImmutableDBOptions::ImmutableDBOptions(const DBOptions& options)
       error_if_exists(options.error_if_exists),
       paranoid_checks(options.paranoid_checks),
       env(options.env),
-      fs(options.file_system),
+      fs(options.env->GetFileSystem()),
       rate_limiter(options.rate_limiter),
       sst_file_manager(options.sst_file_manager),
       info_log(options.info_log),
diff --git a/options/options_helper.cc b/options/options_helper.cc
index 3ebbcee4e2..70b56460d2 100644
--- a/options/options_helper.cc
+++ b/options/options_helper.cc
@@ -38,7 +38,6 @@ DBOptions BuildDBOptions(const ImmutableDBOptions& immutable_db_options,
   options.error_if_exists = immutable_db_options.error_if_exists;
   options.paranoid_checks = immutable_db_options.paranoid_checks;
   options.env = immutable_db_options.env;
-  options.file_system = immutable_db_options.fs;
   options.rate_limiter = immutable_db_options.rate_limiter;
   options.sst_file_manager = immutable_db_options.sst_file_manager;
   options.info_log = immutable_db_options.info_log;
diff --git a/options/options_settable_test.cc b/options/options_settable_test.cc
index 057d772bf3..10c32707f5 100644
--- a/options/options_settable_test.cc
+++ b/options/options_settable_test.cc
@@ -181,8 +181,6 @@ TEST_F(OptionsSettableTest, BlockBasedTableOptionsAllFieldsSettable) {
 TEST_F(OptionsSettableTest, DBOptionsAllFieldsSettable) {
   const OffsetGap kDBOptionsBlacklist = {
       {offsetof(struct DBOptions, env), sizeof(Env*)},
-      {offsetof(struct DBOptions, file_system),
-       sizeof(std::shared_ptr<FileSystem>)},
       {offsetof(struct DBOptions, rate_limiter),
        sizeof(std::shared_ptr<RateLimiter>)},
       {offsetof(struct DBOptions, sst_file_manager),
diff --git a/test_util/fault_injection_test_fs.h b/test_util/fault_injection_test_fs.h
index 30529e0d9f..3c0c53007c 100644
--- a/test_util/fault_injection_test_fs.h
+++ b/test_util/fault_injection_test_fs.h
@@ -137,7 +137,7 @@ class TestFSDirectory : public FSDirectory {
 
 class FaultInjectionTestFS : public FileSystemWrapper {
  public:
-  explicit FaultInjectionTestFS(FileSystem* base)
+  explicit FaultInjectionTestFS(std::shared_ptr<FileSystem> base)
       : FileSystemWrapper(base), filesystem_active_(true) {}
   virtual ~FaultInjectionTestFS() {}
 
diff --git a/tools/ldb_cmd.cc b/tools/ldb_cmd.cc
index d33e4dbca7..0271c9027e 100644
--- a/tools/ldb_cmd.cc
+++ b/tools/ldb_cmd.cc
@@ -295,8 +295,6 @@ void LDBCommand::Run() {
     options_.env = env;
   }
 
-  options_.file_system.reset(new LegacyFileSystemWrapper(options_.env));
-
   if (db_ == nullptr && !NoDBOpen()) {
     OpenDB();
     if (exec_state_.IsFailed() && try_load_options_) {
@@ -1170,7 +1168,7 @@ void GetLiveFilesChecksumInfoFromVersionSet(Options options,
                       /*block_cache_tracer=*/nullptr);
   std::vector<std::string> cf_name_list;
   s = versions.ListColumnFamilies(&cf_name_list, db_path,
-                                  options.file_system.get());
+                                  immutable_db_options.fs.get());
   if (s.ok()) {
     std::vector<ColumnFamilyDescriptor> cf_list;
     for (const auto& name : cf_name_list) {
@@ -1913,8 +1911,6 @@ void ReduceDBLevelsCommand::DoCommand() {
   Status st;
   Options opt = PrepareOptionsForOpenDB();
   int old_level_num = -1;
-  opt.file_system.reset(new LegacyFileSystemWrapper(opt.env));
-  ;
   st = GetOldNumOfLevels(opt, &old_level_num);
   if (!st.ok()) {
     exec_state_ = LDBCommandExecuteResult::Failed(st.ToString());
diff --git a/tools/ldb_cmd_test.cc b/tools/ldb_cmd_test.cc
index 80e2f0d1f4..cc370f4c8a 100644
--- a/tools/ldb_cmd_test.cc
+++ b/tools/ldb_cmd_test.cc
@@ -81,8 +81,6 @@ TEST_F(LdbCmdTest, MemEnv) {
   opts.env = env.get();
   opts.create_if_missing = true;
 
-  opts.file_system.reset(new LegacyFileSystemWrapper(opts.env));
-
   DB* db = nullptr;
   std::string dbname = test::TmpDir();
   ASSERT_OK(DB::Open(opts, dbname, &db));
@@ -199,7 +197,7 @@ class FileChecksumTestHelper {
     std::vector<std::string> cf_name_list;
     Status s;
     s = versions.ListColumnFamilies(&cf_name_list, dbname_,
-                                    options_.file_system.get());
+                                    immutable_db_options.fs.get());
     if (s.ok()) {
       std::vector<ColumnFamilyDescriptor> cf_list;
       for (const auto& name : cf_name_list) {
@@ -264,7 +262,6 @@ TEST_F(LdbCmdTest, DumpFileChecksumNoChecksum) {
   Options opts;
   opts.env = env.get();
   opts.create_if_missing = true;
-  opts.file_system.reset(new LegacyFileSystemWrapper(opts.env));
 
   DB* db = nullptr;
   std::string dbname = test::TmpDir();
@@ -351,7 +348,6 @@ TEST_F(LdbCmdTest, DumpFileChecksumCRC32) {
   opts.create_if_missing = true;
   opts.sst_file_checksum_func =
       std::shared_ptr<FileChecksumFunc>(CreateFileChecksumFuncCrc32c());
-  opts.file_system.reset(new LegacyFileSystemWrapper(opts.env));
 
   DB* db = nullptr;
   std::string dbname = test::TmpDir();