Add faulty fs for io fault injection

velox/common/file/tests/CMakeLists.txt

@@ -11,9 +11,10 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # 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)
+target_link_libraries(velox_file_test_utils PUBLIC velox_file velox_file_test_lib)
 
 add_executable(velox_file_test FileTest.cpp UtilsTest.cpp)
 add_test(velox_file_test velox_file_test)

velox/common/file/tests/utils/CMakeLists.txt

@@ -0,0 +1,22 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+add_library(
+  velox_file_test_lib
+  FaultyFile.cpp
+  FaultyFileSystem.cpp)
+
+target_link_libraries(
+  velox_file_test_lib
+  velox_file)

velox/common/file/tests/utils/FaultyFile.cpp

@@ -0,0 +1,76 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "velox/common/file/tests/utils/FaultyFile.h"
+
+namespace facebook::velox::tests::utils {
+
+FaultyReadFile::FaultyReadFile(
+    std::shared_ptr<ReadFile> delegatedFile,
+    FileFaultInjectionHook injectionHook)
+    : delegatedFile_(std::move(delegatedFile)),
+      injectionHook_(std::move(injectionHook)) {
+  VELOX_CHECK_NOT_NULL(delegatedFile_);
+}
+
+std::string_view
+FaultyReadFile::pread(uint64_t offset, uint64_t length, void* buf) const {
+  if (injectionHook_ != nullptr) {
+    FileReadFaultInjection injection(offset, length, buf);
+    injectionHook_(&injection);
+    if (!injection.delegate) {
+      return injection.injectedReadBuf;
+    }
+  }
+  return delegatedFile_->pread(offset, length, buf);
+}
+
+uint64_t FaultyReadFile::preadv(
+    uint64_t offset,
+    const std::vector<folly::Range<char*>>& buffers) const {
+  if (injectionHook_ != nullptr) {
+    FileReadvFaultInjection injection(offset, buffers);
+    injectionHook_(&injection);
+    if (!injection.delegate) {
+      return injection.readBytes;
+    }
+  }
+  return delegatedFile_->preadv(offset, buffers);
+}
+
+FaultyWriteFile::FaultyWriteFile(
+    std::shared_ptr<WriteFile> delegatedFile,
+    FileFaultInjectionHook injectionHook)
+    : delegatedFile_(std::move(delegatedFile)),
+      injectionHook_(std::move(injectionHook)) {
+  VELOX_CHECK_NOT_NULL(delegatedFile_);
+}
+void FaultyWriteFile::append(std::string_view data) {
+  delegatedFile_->append(data);
+}
+
+void FaultyWriteFile::append(std::unique_ptr<folly::IOBuf> data) {
+  delegatedFile_->append(std::move(data));
+}
+
+void FaultyWriteFile::flush() {
+  delegatedFile_->flush();
+}
+
+void FaultyWriteFile::close() {
+  delegatedFile_->close();
+}
+} // namespace facebook::velox::exec::test

velox/common/file/tests/utils/FaultyFile.h

@@ -0,0 +1,138 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include "velox/common/file/File.h"
+
+namespace facebook::velox::tests::utils {
+
+    /// Defines the per-file operation fault injection.
+struct FileFaultInjectionBase {
+  enum class Type {
+    /// Injects fault for file read operations.
+    kRead,
+    kReadv,
+    /// TODO: add to support fault injections for the other operator types.
+  };
+
+  const Type type;
+
+  /// Indicates to forward this operation to the delegate file or not. If not,
+  /// then the file fault injection hook must have processed the request. For
+  /// instance, if this is a file read injection, then the hook must have filled
+  /// the fake data for read.
+  bool delegate{true};
+
+  explicit FileFaultInjectionBase(Type _type) : type(_type) {}
+};
+
+/// Defines fault injection parameters for file read.
+struct FileReadFaultInjection : FileFaultInjectionBase {
+  const uint64_t offset;
+  const uint64_t length;
+  void* const buf;
+  std::string_view injectedReadBuf;
+
+  FileReadFaultInjection(uint64_t _offset, uint64_t _length, void* _buf)
+      : FileFaultInjectionBase(FileFaultInjectionBase::Type::kRead),
+        offset(_offset),
+        length(_length),
+        buf(_buf) {}
+};
+
+/// Defines fault injection parameters for file readv.
+struct FileReadvFaultInjection : FileFaultInjectionBase {
+  const uint64_t offset;
+  const std::vector<folly::Range<char*>>& buffers;
+  uint64_t readBytes{0};
+
+  FileReadvFaultInjection(
+      uint64_t _offset,
+      const std::vector<folly::Range<char*>>& _buffers)
+      : FileFaultInjectionBase(FileFaultInjectionBase::Type::kReadv),
+        offset(_offset),
+        buffers(_buffers) {}
+};
+
+using FileFaultInjectionHook = std::function<void(FileFaultInjectionBase*)>;
+
+class FaultyReadFile : public ReadFile {
+ public:
+  FaultyReadFile(
+      std::shared_ptr<ReadFile> delegatedFile,
+      FileFaultInjectionHook injectionHook);
+
+  ~FaultyReadFile() override{};
+
+  uint64_t size() const override {
+    return delegatedFile_->size();
+  }
+
+  std::string_view pread(uint64_t offset, uint64_t length, void* buf)
+      const override;
+
+  uint64_t preadv(
+      uint64_t offset,
+      const std::vector<folly::Range<char*>>& buffers) const override;
+
+  uint64_t memoryUsage() const override {
+    return delegatedFile_->memoryUsage();
+  }
+
+  bool shouldCoalesce() const override {
+    return delegatedFile_->shouldCoalesce();
+  }
+
+  std::string getName() const override {
+    return delegatedFile_->getName();
+  }
+
+  uint64_t getNaturalReadSize() const override {
+    return delegatedFile_->getNaturalReadSize();
+  }
+
+ private:
+  const std::shared_ptr<ReadFile> delegatedFile_;
+  const FileFaultInjectionHook injectionHook_;
+};
+
+class FaultyWriteFile : public WriteFile {
+ public:
+  FaultyWriteFile(
+      std::shared_ptr<WriteFile> delegatedFile,
+      FileFaultInjectionHook injectionHook);
+
+  ~FaultyWriteFile() override{};
+
+  void append(std::string_view data) override;
+
+  void append(std::unique_ptr<folly::IOBuf> data) override;
+
+  void flush() override;
+
+  void close() override;
+
+  uint64_t size() const final {
+    return delegatedFile_->size();
+  }
+
+ private:
+  const std::shared_ptr<WriteFile> delegatedFile_;
+  const FileFaultInjectionHook injectionHook_;
+};
+
+} // namespace facebook::velox::exec::test

velox/common/file/tests/utils/FaultyFileSystem.cpp

@@ -0,0 +1,129 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "velox/common/file/tests/utils/FaultyFileSystem.h"
+#include <folly/synchronization/CallOnce.h>
+
+#include <filesystem>
+
+namespace facebook::velox::tests::utils {
+namespace {
+inline std::string extractPath(std::string_view path) {
+  VELOX_CHECK_EQ(path.find(FaultyFileSystem::scheme()), 0);
+  return std::string(path.substr(FaultyFileSystem::scheme().length()));
+}
+
+std::function<bool(std::string_view)> schemeMatcher() {
+  // Note: presto behavior is to prefix local paths with 'file:'.
+  // Check for that prefix and prune to absolute regular paths as needed.
+  return [](std::string_view filePath) {
+    return filePath.find(FaultyFileSystem::scheme()) == 0;
+  };
+}
+
+folly::once_flag faultFilesystemInitOnceFlag;
+
+std::function<std::shared_ptr<
+    FileSystem>(std::shared_ptr<const Config>, std::string_view)>
+fileSystemGenerator() {
+  return
+      [](std::shared_ptr<const Config> properties, std::string_view filePath) {
+        // One instance of faulty FileSystem is sufficient. Initializes on first
+        // access and reuse after that.
+        static std::shared_ptr<FileSystem> lfs;
+        folly::call_once(faultFilesystemInitOnceFlag, [&properties]() {
+          lfs = std::make_shared<FaultyFileSystem>(properties);
+        });
+        return lfs;
+      };
+}
+} // namespace
+
+std::unique_ptr<ReadFile> FaultyFileSystem::openFileForRead(
+    std::string_view path,
+    const FileOptions& options) {
+  const std::string delegatedPath = extractPath(path);
+  auto delegatedFile = getFileSystem(delegatedPath, config_)
+                           ->openFileForRead(delegatedPath, options);
+  return std::make_unique<FaultyReadFile>(
+      std::move(delegatedFile), fileInjectionHook_);
+}
+
+std::unique_ptr<WriteFile> FaultyFileSystem::openFileForWrite(
+    std::string_view path,
+    const FileOptions& options) {
+  const std::string delegatedPath = extractPath(path);
+  auto delegatedFile = getFileSystem(delegatedPath, config_)
+                           ->openFileForWrite(delegatedPath, options);
+  return std::make_unique<FaultyWriteFile>(
+      std::move(delegatedFile), fileInjectionHook_);
+}
+
+void FaultyFileSystem::remove(std::string_view path) {
+  const std::string delegatedPath = extractPath(path);
+  getFileSystem(delegatedPath, config_)->remove(path);
+}
+
+void FaultyFileSystem::rename(
+    std::string_view oldPath,
+    std::string_view newPath,
+    bool overwrite) {
+  const auto delegatedOldPath = extractPath(oldPath);
+  const auto delegatedNewPath = extractPath(newPath);
+  getFileSystem(delegatedOldPath, config_)
+      ->rename(delegatedOldPath, delegatedNewPath);
+}
+
+bool FaultyFileSystem::exists(std::string_view path) {
+  const auto delegatedPath = extractPath(path);
+  return getFileSystem(delegatedPath, config_)->exists(delegatedPath);
+}
+
+std::vector<std::string> FaultyFileSystem::list(std::string_view path) {
+  const auto delegatedDirPath = extractPath(path);
+  return getFileSystem(delegatedDirPath, config_)->list(delegatedDirPath);
+}
+
+void FaultyFileSystem::mkdir(std::string_view path) {
+  const auto delegatedDirPath = extractPath(path);
+  getFileSystem(delegatedDirPath, config_)->mkdir(delegatedDirPath);
+}
+
+void FaultyFileSystem::rmdir(std::string_view path) {
+  const auto delegatedDirPath = extractPath(path);
+  getFileSystem(delegatedDirPath, config_)->rmdir(delegatedDirPath);
+}
+
+void FaultyFileSystem::setFileFaultInjection(
+    FileFaultInjectionHook injectionHook) {
+  std::lock_guard<std::mutex> l(mu_);
+  fileInjectionHook_ = std::move(injectionHook);
+}
+
+void FaultyFileSystem::clearFileFaultInjection() {
+  std::lock_guard<std::mutex> l(mu_);
+  fileInjectionHook_ = nullptr;
+}
+
+void registerFaultyFileSystem() {
+  registerFileSystem(schemeMatcher(), fileSystemGenerator());
+}
+
+std::shared_ptr<FaultyFileSystem> faultyFileSystem() {
+  return std::dynamic_pointer_cast<FaultyFileSystem>(
+      getFileSystem(FaultyFileSystem::scheme(), {}));
+}
+} // namespace facebook::velox::exec::test