Switch back to FSWritableFile in external sst file ingestion job (#13791)

Summary:
This patch reverted "NewRandomRWFile" back to "ReopenWritableFile" in external sst file ingestion job when file is linked instead of copied. The reason is that some of the file systems do not support "NewRandomRWFile". A long term fix is being worked in progress.

Pull Request resolved: #13791

Test Plan: Unit test

Reviewed By: pdillinger

Differential Revision: D78697825

Pulled By: xingbowang

fbshipit-source-id: d3651223ab1f2369aac34b772bba8049c6c2c628

Author: Xingbo Wang

db/external_sst_file_ingestion_job.cc

@@ -156,10 +156,6 @@ Status ExternalSstFileIngestionJob::Prepare(
         // It is unsafe to assume application had sync the file and file
         // directory before ingest the file. For integrity of RocksDB we need
         // to sync the file.
-        // Use FSRandomRWFile instead of FSWritableFile, as in encrypted file
-        // system the FSWritableFile will append a new prefix to the end of the
-        // file when the file exists, which causes file corruption. On the
-        // contrary, FSRandomRWFile handles an existing file correctly.
 
         // TODO(xingbo), We should in general be moving away from production
         // uses of ReuseWritableFile (except explicitly for WAL recycling),
@@ -168,9 +164,9 @@ Status ExternalSstFileIngestionJob::Prepare(
         // re-open+sync+close combo but can (a) be reused easily, and (b) be
         // overridden to do that more cleanly, e.g. in EncryptedEnv.
         // https://github.com/facebook/rocksdb/issues/13741
-        std::unique_ptr<FSRandomRWFile> file_to_sync;
-        Status s = fs_->NewRandomRWFile(path_inside_db, env_options_,
-                                        &file_to_sync, nullptr);
+        std::unique_ptr<FSWritableFile> file_to_sync;
+        Status s = fs_->ReopenWritableFile(path_inside_db, env_options_,
+                                           &file_to_sync, nullptr);
         TEST_SYNC_POINT_CALLBACK("ExternalSstFileIngestionJob::Prepare:Reopen",
                                  &s);
         // Some file systems (especially remote/distributed) don't support

env/env_encryption.cc

@@ -665,17 +665,52 @@ class EncryptedFileSystemImpl : public EncryptedFileSystem {
                               std::unique_ptr<FSWritableFile>* result,
                               IODebugContext* dbg) override {
     result->reset();
-    if (options.use_mmap_writes) {
+    if (options.use_mmap_reads || options.use_mmap_writes) {
       return IOStatus::InvalidArgument();
     }
+
+    size_t prefix_length = 0;
+    std::unique_ptr<BlockAccessCipherStream> stream;
+
     // Open file using underlying Env implementation
     std::unique_ptr<FSWritableFile> underlying;
-    IOStatus status =
+    auto status =
         FileSystemWrapper::ReopenWritableFile(fname, options, &underlying, dbg);
     if (!status.ok()) {
       return status;
     }
-    return CreateWritableEncryptedFile(fname, underlying, options, result, dbg);
+
+    if (underlying->GetFileSize(options.io_options, dbg) != 0) {
+      // read the cipher stream from file for non-empty file
+      std::unique_ptr<FSRandomAccessFile> underlying_file_reader;
+      status = FileSystemWrapper::NewRandomAccessFile(
+          fname, options, &underlying_file_reader, dbg);
+      if (!status.ok()) {
+        return status;
+      }
+
+      status = CreateRandomReadCipherStream(
+          fname, underlying_file_reader, options, &prefix_length, &stream, dbg);
+
+      if (!status.ok()) {
+        return status;
+      }
+    } else {
+      // create cipher stream for new or empty file
+      status = CreateWritableCipherStream(fname, underlying, options,
+                                          &prefix_length, &stream, dbg);
+      if (!status.ok()) {
+        return status;
+      }
+    }
+
+    if (stream) {
+      result->reset(new EncryptedWritableFile(
+          std::move(underlying), std::move(stream), prefix_length));
+    } else {
+      result->reset(underlying.release());
+    }
+    return status;
   }
 
   IOStatus ReuseWritableFile(const std::string& fname,