Add faulty file system for io failure injection

Add faulty file system to allow us to inject failure for low-level file operations in unit and fuzzer test.
It is implemented under velox filesystem. It is wrapper on top of the real file system, it either inject
errors or delegate the file operations to the underlying file system. We extend TempDirectoryPath or
TempFilePath to allow user to specify whether enable fault injection or not in test. If it does, it returns
the file or directory path with faulty file system scheme prefix (faulty:). This allows the velox file system
to open the file through faulty file system and by removing the faulty fs scheme prefix, it can delegate
the file operation to the corresponding underlying file system.

We support three kinds of file fault injections: (1) error injection which throws for a set (or all) file operation
types; (2) delay injection for a set (or all) file operation types; (3) custom injection with user provided fault
injection hook. We define the base structure FaultFileOperation to capture the file fault injection parameters
and each file api will extend with its own operation type.

This PR only supports fault injection for read file operations. Next we will extend to other file operation types
as well as fs operation types.

velox/common/base/SpillConfig.h

@@ -35,7 +35,7 @@ namespace facebook::velox::common {
 
 /// Defining type for a callback function that returns the spill directory path.
 /// Implementations can use it to ensure the path exists before returning.
-using GetSpillDirectoryPathCB = std::function<const std::string&()>;
+using GetSpillDirectoryPathCB = std::function<const std::string()>;
 
 /// The callback used to update the aggregated spill bytes of a query. If the
 /// query spill limit is set, the callback throws if the aggregated spilled

velox/common/base/tests/FsTest.cpp

@@ -24,7 +24,8 @@ namespace facebook::velox::common {
 class FsTest : public testing::Test {};
 
 TEST_F(FsTest, createDirectory) {
-  auto rootPath = exec::test::TempDirectoryPath::createTempDirectory();
+  auto dir = exec::test::TempDirectoryPath::create();
+  auto rootPath = dir->path();
   auto tmpDirectoryPath = rootPath + "/first/second/third";
   // First time should generate directory successfully.
   EXPECT_FALSE(fs::exists(tmpDirectoryPath.c_str()));
@@ -34,7 +35,7 @@ TEST_F(FsTest, createDirectory) {
   // Directory already exist, not creating but should return success.
   EXPECT_TRUE(generateFileDirectory(tmpDirectoryPath.c_str()));
   EXPECT_TRUE(fs::exists(tmpDirectoryPath.c_str()));
-  boost::filesystem::remove_all(rootPath);
+  dir.reset();
   EXPECT_FALSE(fs::exists(rootPath.c_str()));
 }
 

velox/common/caching/tests/AsyncDataCacheTest.cpp

@@ -106,7 +106,7 @@ class AsyncDataCacheTest : public testing::Test {
         tempDirectory_ = exec::test::TempDirectoryPath::create();
       }
       ssdCache = std::make_unique<SsdCache>(
-          fmt::format("{}/cache", tempDirectory_->path),
+          fmt::format("{}/cache", tempDirectory_->path()),
           ssdBytes,
           4,
           executor(),
@@ -816,7 +816,7 @@ TEST_F(AsyncDataCacheTest, DISABLED_ssd) {
 
   cache_->ssdCache()->clear();
   // We cut the tail off one of the cache shards.
-  corruptFile(fmt::format("{}/cache0.cpt", tempDirectory_->path));
+  corruptFile(fmt::format("{}/cache0.cpt", tempDirectory_->path()));
   // We open the cache from checkpoint. Reading checks the data integrity, here
   // we check that more data was read than written.
   initializeCache(kRamBytes, kSsdBytes);

velox/common/caching/tests/SsdFileTest.cpp

@@ -67,7 +67,7 @@ class SsdFileTest : public testing::Test {
 
     tempDirectory_ = exec::test::TempDirectoryPath::create();
     ssdFile_ = std::make_unique<SsdFile>(
-        fmt::format("{}/ssdtest", tempDirectory_->path),
+        fmt::format("{}/ssdtest", tempDirectory_->path()),
         0, // shardId
         bits::roundUp(ssdBytes, SsdFile::kRegionSize) / SsdFile::kRegionSize,
         0, // checkpointInternalBytes

velox/common/file/File.cpp

@@ -238,10 +238,8 @@ LocalWriteFile::LocalWriteFile(
   {
     if (shouldThrowOnFileAlreadyExists) {
       FILE* exists = fopen(buf.get(), "rb");
-      VELOX_CHECK(
-          !exists,
-          "Failure in LocalWriteFile: path '{}' already exists.",
-          path);
+      VELOX_CHECK_NULL(
+          exists, "Failure in LocalWriteFile: path '{}' already exists.", path);
     }
   }
   auto* file = fopen(buf.get(), "ab");

velox/common/file/File.h

@@ -225,14 +225,15 @@ class InMemoryWriteFile final : public WriteFile {
   std::string* file_;
 };
 
-// Current implementation for the local version is quite simple (e.g. no
-// internal arenaing), as local disk writes are expected to be cheap. Local
-// files match against any filepath starting with '/'.
-
+/// Current implementation for the local version is quite simple (e.g. no
+/// internal arenaing), as local disk writes are expected to be cheap. Local
+/// files match against any filepath starting with '/'.
 class LocalReadFile final : public ReadFile {
  public:
   explicit LocalReadFile(std::string_view path);
 
+  /// TODO: deprecate this after creating local file all through velox fs
+  /// interface.
   explicit LocalReadFile(int32_t fd);
 
   ~LocalReadFile();

velox/common/file/tests/CMakeLists.txt

@@ -12,11 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+add_subdirectory(utils)
+
 add_library(velox_file_test_utils TestUtils.cpp)
 target_link_libraries(velox_file_test_utils PUBLIC velox_file)
 
 add_executable(velox_file_test FileTest.cpp UtilsTest.cpp)
 add_test(velox_file_test velox_file_test)
 target_link_libraries(
-  velox_file_test PRIVATE velox_file velox_file_test_utils velox_temp_path
-                          gmock gtest gtest_main)
+        velox_file_test PRIVATE velox_file velox_file_test_utils velox_temp_path
+        gmock gtest gtest_main)