Code cleanup to use concept serial / parallel to replace single-threa…

…ded / multi-threaded as task execution modes (facebookincubator#10792) Summary: A code cleanup for task execution mode concepts, to uniformly use `serial` / `parallel`. Remove usages of `single-threaded` / `multi-threaded`. Fixes facebookincubator#10745 Pull Request resolved: facebookincubator#10792 Reviewed By: Yuhta, bikramSingh91 Differential Revision: D61956385 Pulled By: xiaoxmeng fbshipit-source-id: c601451cc5059aaec304d9e7c34506856f51fbdf
assignUser · Aug 30, 2024 · ede7d0b · ede7d0b
1 parent 7af17de
commit ede7d0b
Show file tree

Hide file tree

Showing 17 changed files with 86 additions and 89 deletions.
diff --git a/velox/common/memory/SharedArbitrator.h b/velox/common/memory/SharedArbitrator.h
@@ -345,8 +345,8 @@ class SharedArbitrator : public memory::MemoryArbitrator {
 
   // Invoked to run global arbitration to reclaim free or used memory from the
   // other queries. The global arbitration run is protected by the exclusive
-  // lock of 'arbitrationLock_' for serial execution. The function returns true
-  // on success, false on failure.
+  // lock of 'arbitrationLock_' for serial execution mode. The function returns
+  // true on success, false on failure.
   bool runGlobalArbitration(ArbitrationOperation* op);
 
   // Gets the mim/max memory capacity growth targets for 'op'. The min and max

diff --git a/velox/common/memory/tests/MockSharedArbitratorTest.cpp b/velox/common/memory/tests/MockSharedArbitratorTest.cpp
@@ -1117,7 +1117,7 @@ TEST_F(MockSharedArbitrationTest, shrinkPools) {
 }
 
 // This test verifies local arbitration runs from the same query has to wait for
-// serial execution.
+// serial execution mode.
 DEBUG_ONLY_TEST_F(
     MockSharedArbitrationTest,
     localArbitrationRunsFromSameQuery) {

diff --git a/velox/common/memory/tests/SharedArbitratorTest.cpp b/velox/common/memory/tests/SharedArbitratorTest.cpp
@@ -305,26 +305,26 @@ class SharedArbitrationTestBase : public exec::test::HiveConnectorTestBase {
 };
 
 namespace {
-std::unique_ptr<folly::Executor> newMultiThreadedExecutor() {
+std::unique_ptr<folly::Executor> newParallelExecutor() {
   return std::make_unique<folly::CPUThreadPoolExecutor>(32);
 }
 
 struct TestParam {
-  bool isSingleThreaded{false};
+  bool isSerialExecutionMode{false};
 };
 } // namespace
 
-/// A test fixture that runs cases within multi-threaded execution mode.
+/// A test fixture that runs cases within parallel execution mode.
 class SharedArbitrationTest : public SharedArbitrationTestBase {
  protected:
   void SetUp() override {
     SharedArbitrationTestBase::SetUp();
-    executor_ = newMultiThreadedExecutor();
+    executor_ = newParallelExecutor();
   }
 };
-/// A test fixture that runs cases within both single-threaded and
-/// multi-threaded execution modes.
-class SharedArbitrationTestWithThreadingModes
+/// A test fixture that runs cases within both serial and
+/// parallel execution modes.
+class SharedArbitrationTestWithExecutionModes
     : public testing::WithParamInterface<TestParam>,
       public SharedArbitrationTestBase {
  public:
@@ -335,15 +335,15 @@ class SharedArbitrationTestWithThreadingModes
  protected:
   void SetUp() override {
     SharedArbitrationTestBase::SetUp();
-    isSingleThreaded_ = GetParam().isSingleThreaded;
-    if (isSingleThreaded_) {
+    isSerialExecutionMode_ = GetParam().isSerialExecutionMode;
+    if (isSerialExecutionMode_) {
       executor_ = nullptr;
     } else {
-      executor_ = newMultiThreadedExecutor();
+      executor_ = newParallelExecutor();
     }
   }
 
-  bool isSingleThreaded_{false};
+  bool isSerialExecutionMode_{false};
 };
 
 DEBUG_ONLY_TEST_F(SharedArbitrationTest, queryArbitrationStateCheck) {
@@ -525,7 +525,7 @@ DEBUG_ONLY_TEST_F(SharedArbitrationTest, skipNonReclaimableTaskTest) {
   ASSERT_EQ(taskPausedCount, 1);
 }
 
-DEBUG_ONLY_TEST_P(SharedArbitrationTestWithThreadingModes, reclaimToOrderBy) {
+DEBUG_ONLY_TEST_P(SharedArbitrationTestWithExecutionModes, reclaimToOrderBy) {
   const int numVectors = 32;
   std::vector<RowVectorPtr> vectors;
   for (int i = 0; i < numVectors; ++i) {
@@ -590,7 +590,7 @@ DEBUG_ONLY_TEST_P(SharedArbitrationTestWithThreadingModes, reclaimToOrderBy) {
       auto task =
           AssertQueryBuilder(duckDbQueryRunner_)
               .queryCtx(orderByQueryCtx)
-              .singleThreaded(isSingleThreaded_)
+              .serialExecution(isSerialExecutionMode_)
               .plan(PlanBuilder()
                         .values(vectors)
                         .orderBy({"c0 ASC NULLS LAST"}, false)
@@ -607,7 +607,7 @@ DEBUG_ONLY_TEST_P(SharedArbitrationTestWithThreadingModes, reclaimToOrderBy) {
       auto task =
           AssertQueryBuilder(duckDbQueryRunner_)
               .queryCtx(fakeMemoryQueryCtx)
-              .singleThreaded(isSingleThreaded_)
+              .serialExecution(isSerialExecutionMode_)
               .plan(PlanBuilder()
                         .values(vectors)
                         .addNode([&](std::string id, core::PlanNodePtr input) {
@@ -628,7 +628,7 @@ DEBUG_ONLY_TEST_P(SharedArbitrationTestWithThreadingModes, reclaimToOrderBy) {
 }
 
 DEBUG_ONLY_TEST_P(
-    SharedArbitrationTestWithThreadingModes,
+    SharedArbitrationTestWithExecutionModes,
     reclaimToAggregation) {
   const int numVectors = 32;
   std::vector<RowVectorPtr> vectors;
@@ -694,7 +694,7 @@ DEBUG_ONLY_TEST_P(
       auto task =
           AssertQueryBuilder(duckDbQueryRunner_)
               .queryCtx(aggregationQueryCtx)
-              .singleThreaded(isSingleThreaded_)
+              .serialExecution(isSerialExecutionMode_)
               .plan(PlanBuilder()
                         .values(vectors)
                         .singleAggregation({"c0", "c1"}, {"array_agg(c2)"})
@@ -712,7 +712,7 @@ DEBUG_ONLY_TEST_P(
       auto task =
           AssertQueryBuilder(duckDbQueryRunner_)
               .queryCtx(fakeMemoryQueryCtx)
-              .singleThreaded(isSingleThreaded_)
+              .serialExecution(isSerialExecutionMode_)
               .plan(PlanBuilder()
                         .values(vectors)
                         .addNode([&](std::string id, core::PlanNodePtr input) {
@@ -733,7 +733,7 @@ DEBUG_ONLY_TEST_P(
 }
 
 DEBUG_ONLY_TEST_P(
-    SharedArbitrationTestWithThreadingModes,
+    SharedArbitrationTestWithExecutionModes,
     reclaimToJoinBuilder) {
   const int numVectors = 32;
   std::vector<RowVectorPtr> vectors;
@@ -800,7 +800,7 @@ DEBUG_ONLY_TEST_P(
       auto task =
           AssertQueryBuilder(duckDbQueryRunner_)
               .queryCtx(joinQueryCtx)
-              .singleThreaded(isSingleThreaded_)
+              .serialExecution(isSerialExecutionMode_)
               .plan(PlanBuilder(planNodeIdGenerator)
                         .values(vectors)
                         .project({"c0 AS t0", "c1 AS t1", "c2 AS t2"})
@@ -828,7 +828,7 @@ DEBUG_ONLY_TEST_P(
       auto task =
           AssertQueryBuilder(duckDbQueryRunner_)
               .queryCtx(fakeMemoryQueryCtx)
-              .singleThreaded(isSingleThreaded_)
+              .serialExecution(isSerialExecutionMode_)
               .plan(PlanBuilder()
                         .values(vectors)
                         .addNode([&](std::string id, core::PlanNodePtr input) {
@@ -1407,10 +1407,10 @@ TEST_F(SharedArbitrationTest, reserveReleaseCounters) {
 }
 
 VELOX_INSTANTIATE_TEST_SUITE_P(
-    SharedArbitrationTestWithThreadingModes,
-    SharedArbitrationTestWithThreadingModes,
+    SharedArbitrationTestWithExecutionModes,
+    SharedArbitrationTestWithExecutionModes,
     testing::ValuesIn(
-        SharedArbitrationTestWithThreadingModes::getTestParams()));
+        SharedArbitrationTestWithExecutionModes::getTestParams()));
 } // namespace facebook::velox::memory
 
 int main(int argc, char** argv) {

diff --git a/velox/core/QueryCtx.h b/velox/core/QueryCtx.h
@@ -65,7 +65,6 @@ class QueryCtx : public std::enable_shared_from_this<QueryCtx> {
 
   folly::Executor* executor() const {
     return executor_;
-    ;
   }
 
   bool isExecutorSupplied() const {

diff --git a/velox/exec/Driver.cpp b/velox/exec/Driver.cpp
@@ -779,7 +779,7 @@ StopReason Driver::runInternal(
                     result->estimateFlatSize(), result->size());
               }
 
-              // This code path is used only in single-threaded execution.
+              // This code path is used only in serial execution mode.
               blockingReason_ = BlockingReason::kWaitForConsumer;
               guard.notThrown();
               return StopReason::kBlock;

diff --git a/velox/exec/Driver.h b/velox/exec/Driver.h
@@ -241,7 +241,7 @@ class BlockingState {
   }
 
   /// Moves out the blocking future stored inside. Can be called only once.
-  /// Used in single-threaded execution.
+  /// Used in serial execution mode.
   ContinueFuture future() {
     return std::move(future_);
   }
@@ -616,7 +616,7 @@ struct DriverFactory {
 
   static void registerAdapter(DriverAdapter adapter);
 
-  bool supportsSingleThreadedExecution() const {
+  bool supportsSerialExecution() const {
     return !needsPartitionedOutput() && !needsExchangeClient() &&
         !needsLocalExchange();
   }

diff --git a/velox/exec/Task.cpp b/velox/exec/Task.cpp
@@ -550,7 +550,7 @@ velox::memory::MemoryPool* Task::addExchangeClientPool(
   return childPools_.back().get();
 }
 
-bool Task::supportsSingleThreadedExecution() const {
+bool Task::supportSerialExecutionMode() const {
   if (consumerSupplier_) {
     return false;
   }
@@ -560,7 +560,7 @@ bool Task::supportsSingleThreadedExecution() const {
       planFragment_, nullptr, &driverFactories, queryCtx_->queryConfig(), 1);
 
   for (const auto& factory : driverFactories) {
-    if (!factory->supportsSingleThreadedExecution()) {
+    if (!factory->supportsSerialExecution()) {
       return false;
     }
   }
@@ -570,18 +570,18 @@ bool Task::supportsSingleThreadedExecution() const {
 
 RowVectorPtr Task::next(ContinueFuture* future) {
   checkExecutionMode(ExecutionMode::kSerial);
-  // NOTE: Task::next() is single-threaded execution so locking is not required
+  // NOTE: Task::next() is serial execution so locking is not required
   // to access Task object.
   VELOX_CHECK_EQ(
       core::ExecutionStrategy::kUngrouped,
       planFragment_.executionStrategy,
-      "Single-threaded execution supports only ungrouped execution");
+      "Serial execution mode supports only ungrouped execution");
 
   if (!splitsStates_.empty()) {
     for (const auto& it : splitsStates_) {
       VELOX_CHECK(
           it.second.noMoreSplits,
-          "Single-threaded execution requires all splits to be added before "
+          "Serial execution mode requires all splits to be added before "
           "calling Task::next().");
     }
   }
@@ -595,7 +595,7 @@ RowVectorPtr Task::next(ContinueFuture* future) {
   if (driverFactories_.empty()) {
     VELOX_CHECK_NULL(
         consumerSupplier_,
-        "Single-threaded execution doesn't support delivering results to a "
+        "Serial execution mode doesn't support delivering results to a "
         "callback");
 
     taskStats_.executionStartTimeMs = getCurrentTimeMs();
@@ -605,7 +605,7 @@ RowVectorPtr Task::next(ContinueFuture* future) {
 
     // In Task::next() we always assume ungrouped execution.
     for (const auto& factory : driverFactories_) {
-      VELOX_CHECK(factory->supportsSingleThreadedExecution());
+      VELOX_CHECK(factory->supportsSerialExecution());
       numDriversUngrouped_ += factory->numDrivers;
       numTotalDrivers_ += factory->numTotalDrivers;
       taskStats_.pipelineStats.emplace_back(
@@ -943,7 +943,7 @@ void Task::resume(std::shared_ptr<Task> self) {
           // event. The Driver gets enqueued by the promise realization.
           //
           // Do not continue the driver if no executor is supplied,
-          // This usually happens in single-threaded execution.
+          // This usually happens in serial execution mode.
           Driver::enqueue(driver);
         }
       }

diff --git a/velox/exec/Task.h b/velox/exec/Task.h
@@ -165,9 +165,9 @@ class Task : public std::enable_shared_from_this<Task> {
   /// splits groups processed concurrently.
   void start(uint32_t maxDrivers, uint32_t concurrentSplitGroups = 1);
 
-  /// If this returns true, this Task supports the single-threaded execution API
+  /// If this returns true, this Task supports the serial execution API
   /// next().
-  bool supportsSingleThreadedExecution() const;
+  bool supportSerialExecutionMode() const;
 
   /// Single-threaded execution API. Runs the query and returns results one
   /// batch at a time. Returns nullptr if query evaluation is finished and no

diff --git a/velox/exec/tests/AssertQueryBuilderTest.cpp b/velox/exec/tests/AssertQueryBuilderTest.cpp
@@ -34,17 +34,17 @@ TEST_F(AssertQueryBuilderTest, basic) {
       .assertResults(data);
 }
 
-TEST_F(AssertQueryBuilderTest, singleThreaded) {
+TEST_F(AssertQueryBuilderTest, serialExecution) {
   auto data = makeRowVector({makeFlatVector<int32_t>({1, 2, 3})});
 
   PlanBuilder builder;
   const auto& plan = builder.values({data}).planNode();
 
   AssertQueryBuilder(plan, duckDbQueryRunner_)
-      .singleThreaded(true)
+      .serialExecution(true)
       .assertResults("VALUES (1), (2), (3)");
 
-  AssertQueryBuilder(plan).singleThreaded(true).assertResults(data);
+  AssertQueryBuilder(plan).serialExecution(true).assertResults(data);
 }
 
 TEST_F(AssertQueryBuilderTest, orderedResults) {

diff --git a/velox/exec/tests/HashJoinTest.cpp b/velox/exec/tests/HashJoinTest.cpp
@@ -6658,7 +6658,7 @@ TEST_F(HashJoinTest, leftJoinPreserveProbeOrder) {
           .planNode();
   auto result = AssertQueryBuilder(plan)
                     .config(core::QueryConfig::kPreferredOutputBatchRows, "1")
-                    .singleThreaded(true)
+                    .serialExecution(true)
                     .copyResults(pool_.get());
   ASSERT_EQ(result->size(), 3);
   auto* v1 =

diff --git a/velox/exec/tests/QueryAssertionsTest.cpp b/velox/exec/tests/QueryAssertionsTest.cpp
@@ -32,15 +32,15 @@ class QueryAssertionsTest : public OperatorTestBase {
   void assertQueryWithThreadingConfigs(
       const core::PlanNodePtr& plan,
       const std::string& duckDbSql) {
-    CursorParameters multiThreadedParams{};
-    multiThreadedParams.planNode = plan;
-    multiThreadedParams.singleThreaded = false;
-    assertQuery(multiThreadedParams, duckDbSql);
-
-    CursorParameters singleThreadedParams{};
-    singleThreadedParams.planNode = plan;
-    singleThreadedParams.singleThreaded = true;
-    assertQuery(singleThreadedParams, duckDbSql);
+    CursorParameters parallelParams{};
+    parallelParams.planNode = plan;
+    parallelParams.serialExecution = false;
+    assertQuery(parallelParams, duckDbSql);
+
+    CursorParameters serialParams{};
+    serialParams.planNode = plan;
+    serialParams.serialExecution = true;
+    assertQuery(serialParams, duckDbSql);
   }
 };
 

diff --git a/velox/exec/tests/TableScanTest.cpp b/velox/exec/tests/TableScanTest.cpp
@@ -3277,7 +3277,7 @@ TEST_F(TableScanTest, remainingFilterLazyWithMultiReferences) {
   writeToFile(file->getPath(), {vector});
   CursorParameters params;
   params.copyResult = false;
-  params.singleThreaded = true;
+  params.serialExecution = true;
   params.planNode =
       PlanBuilder()
           .tableScan(schema, {}, "NOT (c0 % 2 == 0 AND c2 % 3 == 0)")
@@ -3318,7 +3318,7 @@ TEST_F(
   writeToFile(file->getPath(), {vector});
   CursorParameters params;
   params.copyResult = false;
-  params.singleThreaded = true;
+  params.serialExecution = true;
   params.planNode = PlanBuilder()
                         .tableScan(schema, {}, "c0 % 7 == 0 AND c1 % 2 == 0")
                         .planNode();