Back out "Readd: [velox][PR] Add runner for local distributed executi…

…on" (facebookincubator#11529)

Summary:
Pull Request resolved: facebookincubator#11529

Backing out to unbreak OSS CI. The PR facebookincubator#11507 seems to have broken pyVelox builds. Backing out so we can investigate and fix.

Original commit changeset: 4e52927c05ff

Original Phabricator Diff: D65779177

Reviewed By: xiaoxmeng

Differential Revision: D65900074

fbshipit-source-id: 97be486003428caa77c118dbd4b3ae1e14d280d4

velox/CMakeLists.txt

@@ -58,7 +58,6 @@ add_subdirectory(connectors)
 
 if(${VELOX_ENABLE_EXEC})
   add_subdirectory(exec)
-  add_subdirectory(runner)
 endif()
 
 if(${VELOX_ENABLE_DUCKDB})

velox/connectors/hive/HiveConnectorSplit.h

@@ -111,112 +111,4 @@ struct HiveConnectorSplit : public connector::ConnectorSplit {
   static void registerSerDe();
 };
 
-class HiveConnectorSplitBuilder {
- public:
-  explicit HiveConnectorSplitBuilder(std::string filePath)
-      : filePath_{std::move(filePath)} {
-    infoColumns_["$path"] = filePath_;
-  }
-
-  HiveConnectorSplitBuilder& start(uint64_t start) {
-    start_ = start;
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& length(uint64_t length) {
-    length_ = length;
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& splitWeight(int64_t splitWeight) {
-    splitWeight_ = splitWeight;
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& fileFormat(dwio::common::FileFormat format) {
-    fileFormat_ = format;
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& infoColumn(
-      const std::string& name,
-      const std::string& value) {
-    infoColumns_.emplace(std::move(name), std::move(value));
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& partitionKey(
-      std::string name,
-      std::optional<std::string> value) {
-    partitionKeys_.emplace(std::move(name), std::move(value));
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& tableBucketNumber(int32_t bucket) {
-    tableBucketNumber_ = bucket;
-    infoColumns_["$bucket"] = std::to_string(bucket);
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& customSplitInfo(
-      const std::unordered_map<std::string, std::string>& customSplitInfo) {
-    customSplitInfo_ = customSplitInfo;
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& extraFileInfo(
-      const std::shared_ptr<std::string>& extraFileInfo) {
-    extraFileInfo_ = extraFileInfo;
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& serdeParameters(
-      const std::unordered_map<std::string, std::string>& serdeParameters) {
-    serdeParameters_ = serdeParameters;
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& connectorId(const std::string& connectorId) {
-    connectorId_ = connectorId;
-    return *this;
-  }
-
-  HiveConnectorSplitBuilder& fileProperties(FileProperties fileProperties) {
-    fileProperties_ = fileProperties;
-    return *this;
-  }
-
-  std::shared_ptr<connector::hive::HiveConnectorSplit> build() const {
-    return std::make_shared<connector::hive::HiveConnectorSplit>(
-        connectorId_,
-        filePath_,
-        fileFormat_,
-        start_,
-        length_,
-        partitionKeys_,
-        tableBucketNumber_,
-        customSplitInfo_,
-        extraFileInfo_,
-        serdeParameters_,
-        splitWeight_,
-        infoColumns_,
-        fileProperties_);
-  }
-
- private:
-  const std::string filePath_;
-  dwio::common::FileFormat fileFormat_{dwio::common::FileFormat::DWRF};
-  uint64_t start_{0};
-  uint64_t length_{std::numeric_limits<uint64_t>::max()};
-  std::unordered_map<std::string, std::optional<std::string>> partitionKeys_;
-  std::optional<int32_t> tableBucketNumber_;
-  std::unordered_map<std::string, std::string> customSplitInfo_ = {};
-  std::shared_ptr<std::string> extraFileInfo_ = {};
-  std::unordered_map<std::string, std::string> serdeParameters_ = {};
-  std::unordered_map<std::string, std::string> infoColumns_ = {};
-  std::string connectorId_;
-  int64_t splitWeight_{0};
-  std::optional<FileProperties> fileProperties_;
-};
-
 } // namespace facebook::velox::connector::hive

velox/connectors/hive/tests/CMakeLists.txt

@@ -21,7 +21,6 @@ add_executable(
   HiveConnectorSerDeTest.cpp
   HivePartitionFunctionTest.cpp
   HivePartitionUtilTest.cpp
-  HiveSplitTest.cpp
   PartitionIdGeneratorTest.cpp
   TableHandleTest.cpp)
 add_test(velox_hive_connector_test velox_hive_connector_test)

velox/exec/tests/TableScanTest.cpp

@@ -180,7 +180,7 @@ class TableScanTest : public virtual HiveConnectorTestBase {
       const std::string& filePath,
       const TypePtr& partitionType,
       const std::optional<std::string>& partitionValue) {
-    auto split = exec::test::HiveConnectorSplitBuilder(filePath)
+    auto split = HiveConnectorSplitBuilder(filePath)
                      .partitionKey("pkey", partitionValue)
                      .build();
     auto outputType =
@@ -411,7 +411,7 @@ TEST_F(TableScanTest, partitionKeyAlias) {
       {"a", regularColumn("c0", BIGINT())},
       {"ds_alias", partitionKey("ds", VARCHAR())}};
 
-  auto split = exec::test::HiveConnectorSplitBuilder(filePath->getPath())
+  auto split = HiveConnectorSplitBuilder(filePath->getPath())
                    .partitionKey("ds", "2021-12-02")
                    .build();
 
@@ -1806,8 +1806,7 @@ TEST_F(TableScanTest, splitOffsetAndLength) {
 }
 
 TEST_F(TableScanTest, fileNotFound) {
-  auto split =
-      exec::test::HiveConnectorSplitBuilder("/path/to/nowhere.orc").build();
+  auto split = HiveConnectorSplitBuilder("/path/to/nowhere.orc").build();
   auto assertMissingFile = [&](bool ignoreMissingFiles) {
     AssertQueryBuilder(tableScanNode())
         .connectorSessionProperty(
@@ -1830,7 +1829,7 @@ TEST_F(TableScanTest, validFileNoData) {
 
   auto filePath = facebook::velox::test::getDataFilePath(
       "velox/exec/tests", "data/emptyPresto.dwrf");
-  auto split = exec::test::HiveConnectorSplitBuilder(filePath)
+  auto split = HiveConnectorSplitBuilder(filePath)
                    .start(0)
                    .length(fs::file_size(filePath) / 2)
                    .build();
@@ -1950,7 +1949,7 @@ TEST_F(TableScanTest, partitionedTableDateKey) {
 
   // Test partition filter on date column.
   {
-    auto split = exec::test::HiveConnectorSplitBuilder(filePath->getPath())
+    auto split = HiveConnectorSplitBuilder(filePath->getPath())
                      .partitionKey("pkey", partitionValue)
                      .build();
     auto outputType = ROW({"pkey", "c0", "c1"}, {DATE(), BIGINT(), DOUBLE()});
@@ -2852,10 +2851,9 @@ TEST_F(TableScanTest, bucket) {
     writeToFile(filePaths[i]->getPath(), rowVector);
     rowVectors.emplace_back(rowVector);
 
-    splits.emplace_back(
-        exec::test::HiveConnectorSplitBuilder(filePaths[i]->getPath())
-            .tableBucketNumber(bucket)
-            .build());
+    splits.emplace_back(HiveConnectorSplitBuilder(filePaths[i]->getPath())
+                            .tableBucketNumber(bucket)
+                            .build());
   }
 
   createDuckDbTable(rowVectors);
@@ -2879,7 +2877,7 @@ TEST_F(TableScanTest, bucket) {
 
   for (int i = 0; i < buckets.size(); ++i) {
     int bucketValue = buckets[i];
-    auto hsplit = exec::test::HiveConnectorSplitBuilder(filePaths[i]->getPath())
+    auto hsplit = HiveConnectorSplitBuilder(filePaths[i]->getPath())
                       .tableBucketNumber(bucketValue)
                       .build();
 
@@ -2899,7 +2897,7 @@ TEST_F(TableScanTest, bucket) {
 
     // Filter on bucket column, but don't project it out
     auto rowTypes = ROW({"c0", "c1"}, {INTEGER(), BIGINT()});
-    hsplit = exec::test::HiveConnectorSplitBuilder(filePaths[i]->getPath())
+    hsplit = HiveConnectorSplitBuilder(filePaths[i]->getPath())
                  .tableBucketNumber(bucketValue)
                  .build();
     op = PlanBuilder()
@@ -4170,7 +4168,7 @@ TEST_F(TableScanTest, reuseRowVector) {
                   .tableScan(rowType, {}, "c0 < 5")
                   .project({"c1.c0"})
                   .planNode();
-  auto split = exec::test::HiveConnectorSplitBuilder(file->getPath()).build();
+  auto split = HiveConnectorSplitBuilder(file->getPath()).build();
   auto expected = makeRowVector(
       {makeFlatVector<int32_t>(10, [](auto i) { return i % 5; })});
   AssertQueryBuilder(plan).splits({split, split}).assertResults(expected);
@@ -4751,7 +4749,7 @@ TEST_F(TableScanTest, varbinaryPartitionKey) {
       {"a", regularColumn("c0", BIGINT())},
       {"ds_alias", partitionKey("ds", VARBINARY())}};
 
-  auto split = exec::test::HiveConnectorSplitBuilder(filePath->getPath())
+  auto split = HiveConnectorSplitBuilder(filePath->getPath())
                    .partitionKey("ds", "2021-12-02")
                    .build();
 
@@ -4790,7 +4788,7 @@ TEST_F(TableScanTest, timestampPartitionKey) {
   ColumnHandleMap assignments = {{"t", partitionKey("t", TIMESTAMP())}};
   std::vector<std::shared_ptr<connector::ConnectorSplit>> splits;
   for (auto& t : inputs) {
-    splits.push_back(exec::test::HiveConnectorSplitBuilder(filePath->getPath())
+    splits.push_back(HiveConnectorSplitBuilder(filePath->getPath())
                          .partitionKey("t", t)
                          .build());
   }
@@ -4809,7 +4807,7 @@ TEST_F(TableScanTest, partitionKeyNotMatchPartitionKeysHandle) {
   writeToFile(filePath->getPath(), vectors);
   createDuckDbTable(vectors);
 
-  auto split = exec::test::HiveConnectorSplitBuilder(filePath->getPath())
+  auto split = HiveConnectorSplitBuilder(filePath->getPath())
                    .partitionKey("ds", "2021-12-02")
                    .build();
 

velox/exec/tests/utils/CMakeLists.txt

@@ -22,10 +22,8 @@ add_library(
   AssertQueryBuilder.cpp
   ArbitratorTestUtil.cpp
   Cursor.cpp
-  DistributedPlanBuilder.cpp
   HiveConnectorTestBase.cpp
   LocalExchangeSource.cpp
-  LocalRunnerTestBase.cpp
   OperatorTestBase.cpp
   PlanBuilder.cpp
   QueryAssertions.cpp
@@ -37,7 +35,6 @@ add_library(
 
 target_link_libraries(
   velox_exec_test_lib
-  velox_local_runner
   velox_vector_test_lib
   velox_temp_path
   velox_core

velox/exec/tests/utils/Cursor.cpp

@@ -275,10 +275,6 @@ class MultiThreadedTaskCursor : public TaskCursorBase {
   /// Starts the task if not started yet.
   bool moveNext() override {
     start();
-    if (error_) {
-      std::rethrow_exception(error_);
-    }
-
     current_ = queue_->dequeue();
     if (task_->error()) {
       // Wait for the task to finish (there's' a small period of time between
@@ -306,13 +302,6 @@ class MultiThreadedTaskCursor : public TaskCursorBase {
     return current_;
   }
 
-  void setError(std::exception_ptr error) override {
-    error_ = error;
-    if (task_) {
-      task_->setError(error);
-    }
-  }
-
   const std::shared_ptr<Task>& task() override {
     return task_;
   }
@@ -327,7 +316,6 @@ class MultiThreadedTaskCursor : public TaskCursorBase {
   std::shared_ptr<exec::Task> task_;
   RowVectorPtr current_;
   bool atEnd_{false};
-  std::exception_ptr error_;
 };
 
 class SingleThreadedTaskCursor : public TaskCursorBase {
@@ -403,13 +391,6 @@ class SingleThreadedTaskCursor : public TaskCursorBase {
     return current_;
   }
 
-  void setError(std::exception_ptr error) override {
-    error_ = error;
-    if (task_) {
-      task_->setError(error);
-    }
-  }
-
   const std::shared_ptr<Task>& task() override {
     return task_;
   }
@@ -418,7 +399,6 @@ class SingleThreadedTaskCursor : public TaskCursorBase {
   std::shared_ptr<exec::Task> task_;
   RowVectorPtr current_;
   RowVectorPtr next_;
-  std::exception_ptr error_;
 };
 
 std::unique_ptr<TaskCursor> TaskCursor::create(const CursorParameters& params) {

velox/exec/tests/utils/Cursor.h

@@ -139,8 +139,6 @@ class TaskCursor {
 
   virtual RowVectorPtr& current() = 0;
 
-  virtual void setError(std::exception_ptr error) = 0;
-
   virtual const std::shared_ptr<Task>& task() = 0;
 };