Add initial table writer fuzzer

Author: kewang1024

velox/docs/develop/testing/writer-fuzzer.rst

@@ -0,0 +1,45 @@
+===========
+Writer Fuzzer
+===========
+
+Writer fuzzer tests table write plan with up to 5 regular columns and
+up to 3 partition keys.
+
+At each iteration, fuzzer randomly generate a table write plan with different
+table properties, as of now, only support partitioned and unpartitioned table.
+
+The fuzzer then generates inputs and runs the query plan and compares the
+results with PrestoDB.
+As of now, we compare:
+1. How many rows were written.
+2. Output directories have the same directory layout and hierarchy.
+
+How to run
+----------
+
+Use velox_writer_fuzzer_test binary to run join fuzzer:
+
+::
+
+    velox/exec/tests/velox_writer_fuzzer_test
+
+By default, the fuzzer will go through 10 interations. Use --steps
+or --duration-sec flag to run fuzzer for longer. Use --seed to
+reproduce fuzzer failures.
+
+Here is a full list of supported command line arguments.
+
+* ``–-steps``: How many iterations to run. Each iteration generates and
+  evaluates one tale writer plan. Default is 10.
+
+* ``–-duration_sec``: For how long to run in seconds. If both ``-–steps``
+  and ``-–duration_sec`` are specified, –duration_sec takes precedence.
+
+* ``–-seed``: The seed to generate random expressions and input vectors with.
+
+* ``–-batch_size``: The size of input vectors to generate. Default is 100.
+
+* ``--num_batches``: The number of input vectors of size `--batch_size` to
+  generate. Default is 5.
+
+If running from CLion IDE, add ``--logtostderr=1`` to see the full output.

velox/exec/fuzzer/CMakeLists.txt

@@ -57,3 +57,14 @@ target_link_libraries(
   velox_expression_test_utility
   velox_aggregation_fuzzer_base
   velox_temp_path)
+
+add_library(velox_writer_fuzzer WriterFuzzer.cpp)
+
+target_link_libraries(
+  velox_writer_fuzzer
+  velox_fuzzer_util
+  velox_type
+  velox_vector_fuzzer
+  velox_exec_test_lib
+  velox_temp_path
+  velox_vector_test_lib)

velox/exec/fuzzer/PrestoQueryRunner.cpp

@@ -26,6 +26,8 @@
 #include "velox/exec/tests/utils/QueryAssertions.h"
 #include "velox/serializers/PrestoSerializer.h"
 #include "velox/type/parser/TypeParser.h"
+#include "velox/velox/connectors/hive/HiveDataSink.h"
+#include "velox/velox/connectors/hive/TableHandle.h"
 
 #include <utility>
 
@@ -35,16 +37,6 @@ namespace facebook::velox::exec::test {
 
 namespace {
 
-template <typename T>
-T extractSingleValue(const std::vector<RowVectorPtr>& data) {
-  VELOX_CHECK_EQ(1, data.size());
-  VELOX_CHECK_EQ(1, data[0]->childrenSize());
-
-  auto simpleVector = data[0]->childAt(0)->as<SimpleVector<T>>();
-  VELOX_CHECK(!simpleVector->isNullAt(0));
-  return simpleVector->valueAt(0);
-}
-
 void writeToFile(
     const std::string& path,
     const std::vector<RowVectorPtr>& data,
@@ -174,6 +166,11 @@ std::optional<std::string> PrestoQueryRunner::toSql(
     return toSql(aggregationNode);
   }
 
+  if (auto tableWriteNode =
+          std::dynamic_pointer_cast<const core::TableWriteNode>(plan)) {
+    return toSql(tableWriteNode);
+  }
+
   VELOX_NYI();
 }
 
@@ -500,6 +497,40 @@ std::optional<std::string> PrestoQueryRunner::toSql(
   return sql.str();
 }
 
+std::optional<std::string> PrestoQueryRunner::toSql(
+    const std::shared_ptr<const core::TableWriteNode>& tableWriteNode) {
+  std::vector<std::string> partitionKeys;
+  auto insertTableHandle =
+      std::dynamic_pointer_cast<connector::hive::HiveInsertTableHandle>(
+          tableWriteNode->insertTableHandle()->connectorInsertTableHandle());
+
+  // Returns a CREATE sql with specified table properties from TableWriteNode,
+  // example sql:
+  // CREATE TABLE tmp_write (c0 integer, c1 varchar, p0 varchar)
+  // WITH (PARTITIONED_BY = ARRAY['p0']);
+  std::stringstream sql;
+  sql << "CREATE TABLE tmp_write ( ";
+  for (auto i = 0; i < tableWriteNode->columnNames().size(); ++i) {
+    appendComma(i, sql);
+    sql << tableWriteNode->columnNames()[i] << " "
+        << toTypeSql(tableWriteNode->columns()->childAt(i));
+    if (insertTableHandle->inputColumns()[i]->isPartitionKey()) {
+      partitionKeys.push_back(insertTableHandle->inputColumns()[i]->name());
+    }
+  }
+  sql << ")";
+
+  if (insertTableHandle->isPartitioned()) {
+    sql << " WITH (PARTITIONED_BY = ARRAY[";
+    for (int i = 0; i < partitionKeys.size(); ++i) {
+      appendComma(i, sql);
+      sql << "'" << partitionKeys[i] << "'";
+    }
+    sql << "])";
+  }
+  return sql.str();
+}
+
 std::multiset<std::vector<variant>> PrestoQueryRunner::execute(
     const std::string& sql,
     const std::vector<RowVectorPtr>& input,

velox/exec/fuzzer/PrestoQueryRunner.h

@@ -23,6 +23,13 @@
 #include "velox/vector/ComplexVector.h"
 
 namespace facebook::velox::exec::test {
+template <typename T>
+T extractSingleValue(const std::vector<RowVectorPtr>& data) {
+  auto simpleVector = data[0]->childAt(0)->as<SimpleVector<T>>();
+  VELOX_CHECK(!simpleVector->isNullAt(0));
+  return simpleVector->valueAt(0);
+}
+
 class PrestoQueryRunner : public velox::exec::test::ReferenceQueryRunner {
  public:
   /// @param coordinatorUri Presto REST API endpoint, e.g. http://127.0.0.1:8080
@@ -58,7 +65,7 @@ class PrestoQueryRunner : public velox::exec::test::ReferenceQueryRunner {
 
   /// Executes Presto SQL query and returns the results. Tables referenced by
   /// the query must already exist.
-  std::vector<velox::RowVectorPtr> execute(const std::string& sql);
+  std::vector<velox::RowVectorPtr> execute(const std::string& sql) override;
 
   bool supportsVeloxVectorResults() const override;
 
@@ -86,6 +93,9 @@ class PrestoQueryRunner : public velox::exec::test::ReferenceQueryRunner {
   std::optional<std::string> toSql(
       const std::shared_ptr<const velox::core::ProjectNode>& projectNode);
 
+  std::optional<std::string> toSql(
+      const std::shared_ptr<const core::TableWriteNode>& tableWriteNode);
+
   std::string startQuery(const std::string& sql);
 
   std::string fetchNext(const std::string& nextUri);

velox/exec/fuzzer/ReferenceQueryRunner.h

@@ -50,6 +50,10 @@ class ReferenceQueryRunner {
       const RowTypePtr& resultType) {
     VELOX_UNSUPPORTED();
   }
+
+  virtual std::vector<velox::RowVectorPtr> execute(const std::string& sql) {
+    VELOX_UNSUPPORTED();
+  }
 };
 
 } // namespace facebook::velox::exec::test