Add file system error injection in table writer fuzzer

velox/common/file/tests/FaultyFileSystem.h

@@ -26,7 +26,6 @@
 namespace facebook::velox::tests::utils {
 
 using namespace filesystems;
-
 /// Implements faulty filesystem for io fault injection in unit test. It is a
 /// wrapper on top of a real file system, and by default it delegates the the
 /// file operation to the real file system underneath.
@@ -55,11 +54,11 @@ class FaultyFileSystem : public FileSystem {
 
   std::unique_ptr<ReadFile> openFileForRead(
       std::string_view path,
-      const FileOptions& options) override;
+      const FileOptions& options = {}) override;
 
   std::unique_ptr<WriteFile> openFileForWrite(
       std::string_view path,
-      const FileOptions& options) override;
+      const FileOptions& options = {}) override;
 
   void remove(std::string_view path) override;
 

velox/dwio/common/FileSink.cpp

@@ -145,6 +145,18 @@ LocalFileSink::LocalFileSink(const std::string& name, const Options& options)
   writeFile_ = fs->openFileForWrite(name_);
 }
 
+LocalFileSink::LocalFileSink(
+    const std::string& name,
+    const Options& options,
+    const bool initializeWriter)
+    : FileSink{name, options}, writeFile_() {
+  VELOX_CHECK(!initializeWriter);
+  const auto dir = fs::path(name_).parent_path();
+  if (!fs::exists(dir)) {
+    VELOX_CHECK(velox::common::generateFileDirectory(dir.c_str()));
+  }
+}
+
 void LocalFileSink::doClose() {
   LOG(INFO) << "closing file: " << name()
             << ",  total size: " << succinctBytes(size_);

velox/dwio/common/FileSink.h

@@ -166,9 +166,14 @@ class LocalFileSink : public FileSink {
   static void registerFactory();
 
  protected:
+  // Only used by FaultyFileSink to initialize writeFile_ as FaultyWriteFile.
+  LocalFileSink(
+      const std::string& name,
+      const Options& options,
+      const bool initializeWriter);
   void doClose() override;
 
- private:
+ protected:
   std::unique_ptr<WriteFile> writeFile_;
 };
 

velox/dwio/common/tests/CMakeLists.txt

@@ -13,6 +13,9 @@
 # limitations under the License.
 
 add_subdirectory(utils)
+velox_add_library(velox_dwio_faulty_file_sink FaultyFileSink.cpp)
+velox_link_libraries(velox_dwio_faulty_file_sink velox_file_test_utils
+                     velox_dwio_common)
 
 add_executable(
   velox_dwio_common_test

velox/dwio/common/tests/FaultyFileSink.cpp

@@ -0,0 +1,52 @@
+/*
+ * 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/dwio/common/tests/FaultyFileSink.h"
+#include "velox/common/base/Fs.h"
+#include "velox/common/file/FileSystems.h"
+#include "velox/common/file/tests/FaultyFileSystem.h"
+#include "velox/dwio/common/FileSink.h"
+#include "velox/dwio/common/exception/Exception.h"
+
+namespace facebook::velox::dwio::common {
+namespace {
+using tests::utils::FaultyFileSystem;
+
+std::unique_ptr<FileSink> createFaultyFileSink(
+    const std::string& filePath,
+    const FileSink::Options& options) {
+  if (filePath.find("faulty:") == 0) {
+    return std::make_unique<FaultyFileSink>(
+        filePath, filePath.substr(7), options);
+  }
+  return nullptr;
+}
+} // namespace
+
+FaultyFileSink::FaultyFileSink(
+    const std::string& faultyName,
+    const std::string& name,
+    const Options& options)
+    : LocalFileSink{name, options, false}, faultyName_(faultyName) {
+  auto fs = filesystems::getFileSystem(faultyName_, nullptr);
+  writeFile_ = fs->openFileForWrite(faultyName_);
+}
+
+void registerFaultyFileSinks() {
+  facebook::velox::dwio::common::FileSink::registerFactory(
+      (createFaultyFileSink));
+}
+} // namespace facebook::velox::dwio::common

velox/dwio/common/tests/FaultyFileSink.h

@@ -0,0 +1,46 @@
+/*
+ * 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 <chrono>
+
+#include "velox/common/config/Config.h"
+#include "velox/common/file/File.h"
+#include "velox/common/file/tests/FaultyFile.h"
+#include "velox/common/io/IoStatistics.h"
+#include "velox/dwio/common/Closeable.h"
+#include "velox/dwio/common/DataBuffer.h"
+#include "velox/dwio/common/FileSink.h"
+#include "velox/dwio/common/MetricsLog.h"
+
+namespace facebook::velox::dwio::common {
+using namespace facebook::velox::io;
+
+class FaultyFileSink : public LocalFileSink {
+ public:
+  FaultyFileSink(
+      const std::string& faultyName,
+      const std::string& name,
+      const Options& options);
+
+ private:
+  std::string faultyName_;
+};
+
+void registerFaultyFileSinks();
+
+} // namespace facebook::velox::dwio::common

velox/exec/fuzzer/CMakeLists.txt

@@ -29,7 +29,8 @@ target_link_libraries(
   velox_hive_connector
   velox_dwio_dwrf_reader
   velox_dwio_dwrf_writer
-  velox_dwio_catalog_fbhive)
+  velox_dwio_catalog_fbhive
+  velox_dwio_faulty_file_sink)
 
 add_library(velox_aggregation_fuzzer_base AggregationFuzzerBase.cpp)
 
@@ -102,7 +103,9 @@ target_link_libraries(
   velox_exec_test_lib
   velox_expression_test_utility
   velox_temp_path
-  velox_vector_test_lib)
+  velox_vector_test_lib
+  velox_dwio_faulty_file_sink
+  velox_file_test_utils)
 
 add_library(velox_memory_arbitration_fuzzer MemoryArbitrationFuzzer.cpp)
 

velox/exec/fuzzer/WriterFuzzer.cpp

@@ -22,6 +22,7 @@
 #include "velox/common/base/Fs.h"
 #include "velox/common/encode/Base64.h"
 #include "velox/common/file/FileSystems.h"
+#include "velox/common/file/tests/FaultyFileSystem.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/connectors/hive/HiveConnectorSplit.h"
 #include "velox/connectors/hive/TableHandle.h"
@@ -36,6 +37,11 @@
 #include "velox/vector/VectorSaver.h"
 #include "velox/vector/fuzzer/VectorFuzzer.h"
 
+DEFINE_bool(
+    file_system_error_injection,
+    true,
+    "When enabled, inject file system write error with certain possibility");
+
 DEFINE_int32(steps, 10, "Number of plans to generate and test.");
 
 DEFINE_int32(
@@ -63,6 +69,8 @@ using namespace facebook::velox::test;
 namespace facebook::velox::exec::test {
 
 namespace {
+using facebook::velox::filesystems::FileSystem;
+using tests::utils::FaultyFileSystem;
 
 class WriterFuzzer {
  public:
@@ -113,6 +121,11 @@ class WriterFuzzer {
       std::vector<TypePtr> types,
       size_t partitionOffset);
 
+  // Generates file system write error with certain possibility
+  bool injectWriterError(
+      const std::shared_ptr<FaultyFileSystem>& faultyFs,
+      const std::vector<std::string>& partitionKeys);
+
   void verifyWriter(
       const std::vector<RowVectorPtr>& input,
       const std::vector<std::string>& names,
@@ -123,7 +136,9 @@ class WriterFuzzer {
       const std::vector<std::string>& bucketColumns,
       int32_t sortColumnOffset,
       const std::vector<std::shared_ptr<const HiveSortingColumn>>& sortBy,
-      const std::string& outputDirectoryPath);
+      const std::string& outputFaultyDirectoryPath,
+      const std::string& outputDirectoryPath,
+      const bool writeErrorInjected);
 
   // Generates table column handles based on table column properties
   std::unordered_map<std::string, std::shared_ptr<connector::ColumnHandle>>
@@ -244,6 +259,8 @@ class WriterFuzzer {
   std::shared_ptr<memory::MemoryPool> writerPool_{
       rootPool_->addAggregateChild("writerFuzzerWriter")};
   VectorFuzzer vectorFuzzer_;
+
+  const std::string injectedErrorMsg_ = "Injected Faulty File Error";
 };
 } // namespace
 
@@ -340,7 +357,11 @@ void WriterFuzzer::go() {
     }
     auto input = generateInputData(names, types, partitionOffset);
 
-    auto tempDirPath = exec::test::TempDirectoryPath::create();
+    auto tempFaultyDirPath = exec::test::TempDirectoryPath::create(true);
+    auto faultyFs = std::dynamic_pointer_cast<FaultyFileSystem>(
+        filesystems::getFileSystem(tempFaultyDirPath->getPath(), {}));
+    bool writeErrorInjected = injectWriterError(faultyFs, partitionKeys);
+
     verifyWriter(
         input,
         names,
@@ -351,7 +372,11 @@ void WriterFuzzer::go() {
         bucketColumns,
         sortColumnOffset,
         sortBy,
-        tempDirPath->getPath());
+        tempFaultyDirPath->getPath(),
+        std::string(faultyFs->extractPath(tempFaultyDirPath->getPath())),
+        writeErrorInjected);
+
+    faultyFs->clearFileFaultInjections();
 
     LOG(INFO) << "==============================> Done with iteration "
               << iteration++;
@@ -413,6 +438,24 @@ std::vector<RowVectorPtr> WriterFuzzer::generateInputData(
   return input;
 }
 
+bool WriterFuzzer::injectWriterError(
+    const std::shared_ptr<FaultyFileSystem>& faultyFs,
+    const std::vector<std::string>& partitionKeys) {
+  if (FLAGS_file_system_error_injection && partitionKeys.empty() &&
+      vectorFuzzer_.coinToss(0.01)) {
+    std::exception_ptr fileError;
+    try {
+      VELOX_FAIL(injectedErrorMsg_);
+    } catch (VeloxRuntimeError&) {
+      fileError = std::current_exception();
+    }
+    faultyFs->setFileInjectionError(
+        fileError, {tests::utils::FaultFileOperation::Type::kWrite});
+    return true;
+  }
+  return false;
+}
+
 void WriterFuzzer::verifyWriter(
     const std::vector<RowVectorPtr>& input,
     const std::vector<std::string>& names,
@@ -423,11 +466,13 @@ void WriterFuzzer::verifyWriter(
     const std::vector<std::string>& bucketColumns,
     const int32_t sortColumnOffset,
     const std::vector<std::shared_ptr<const HiveSortingColumn>>& sortBy,
-    const std::string& outputDirectoryPath) {
+    const std::string& outputFaultyDirectoryPath,
+    const std::string& outputDirectoryPath,
+    const bool writeErrorInjected) {
   const auto plan = PlanBuilder()
                         .values(input)
                         .tableWrite(
-                            outputDirectoryPath,
+                            outputFaultyDirectoryPath,
                             partitionKeys,
                             bucketCount,
                             bucketColumns,
@@ -436,7 +481,17 @@ void WriterFuzzer::verifyWriter(
 
   const auto maxDrivers =
       boost::random::uniform_int_distribution<int32_t>(1, 16)(rng_);
-  const auto result = veloxToPrestoResult(execute(plan, maxDrivers));
+  RowVectorPtr result;
+  try {
+    result = veloxToPrestoResult(execute(plan, maxDrivers));
+  } catch (VeloxRuntimeError& error) {
+    if (writeErrorInjected) {
+      VELOX_CHECK(
+          error.message() == injectedErrorMsg_,
+          "write plan failed with different error code");
+      return;
+    }
+  }
 
   const auto dropSql = "DROP TABLE IF EXISTS tmp_write";
   const auto sql = referenceQueryRunner_->toSql(plan).value();

velox/exec/fuzzer/WriterFuzzerRunner.h

@@ -23,8 +23,10 @@
 #include <vector>
 
 #include "velox/common/file/FileSystems.h"
+#include "velox/common/file/tests/FaultyFileSystem.h"
 #include "velox/connectors/hive/HiveConnector.h"
 #include "velox/dwio/common/FileSink.h"
+#include "velox/dwio/common/tests/FaultyFileSink.h"
 #include "velox/dwio/dwrf/RegisterDwrfReader.h"
 #include "velox/dwio/dwrf/RegisterDwrfWriter.h"
 #include "velox/exec/fuzzer/FuzzerUtil.h"
@@ -74,6 +76,7 @@ class WriterFuzzerRunner {
       size_t seed,
       std::unique_ptr<ReferenceQueryRunner> referenceQueryRunner) {
     filesystems::registerLocalFileSystem();
+    tests::utils::registerFaultyFileSystem();
     connector::registerConnectorFactory(
         std::make_shared<connector::hive::HiveConnectorFactory>());
     auto hiveConnector =
@@ -87,6 +90,7 @@ class WriterFuzzerRunner {
     dwrf::registerDwrfReaderFactory();
     dwrf::registerDwrfWriterFactory();
     dwio::common::registerFileSinks();
+    dwio::common::registerFaultyFileSinks();
     facebook::velox::exec::test::writerFuzzer(
         seed, std::move(referenceQueryRunner));
     // Calling gtest here so that it can be recognized as tests in CI systems.