Allow single thread execution mode for local partition to support spa…

…rk union for gluten (#11531)

Summary:
Pull Request resolved: #11531

Local partition could be used to support spark union for gluten. For union, it will be translated to local partition
with a followup distinct aggregation. Gluten is single thread execution mode using API like task::next to pull data.
Currently, Velox only allow single thread execution mode if the front of a pipeline is not local exchange which is not necessary.
Task::next actually pull results from all the pipelines so it should be good. This PR removes that check and verified with the
unit test.

Reviewed By: bikramSingh91, oerling

Differential Revision: D65917749

fbshipit-source-id: 1ebe8c5782d7d36d4a77cdc792169856c03d0dbd

velox/exec/Driver.h

@@ -656,8 +656,7 @@ struct DriverFactory {
   static void registerAdapter(DriverAdapter adapter);
 
   bool supportsSerialExecution() const {
-    return !needsPartitionedOutput() && !needsExchangeClient() &&
-        !needsLocalExchange();
+    return !needsPartitionedOutput() && !needsExchangeClient();
   }
 
   const core::PlanNodeId& leafNodeId() const {

velox/exec/tests/LocalPartitionTest.cpp

@@ -613,8 +613,12 @@ TEST_F(LocalPartitionTest, unionAllLocalExchange) {
   auto data1 = makeRowVector({"d0"}, {makeFlatVector<StringView>({"x"})});
   auto data2 = makeRowVector({"e0"}, {makeFlatVector<StringView>({"y"})});
 
-  auto planNodeIdGenerator = std::make_shared<core::PlanNodeIdGenerator>();
-  auto plan = PlanBuilder(planNodeIdGenerator)
+  for (bool serialExecutionMode : {false, true}) {
+    SCOPED_TRACE(fmt::format("serialExecutionMode {}", serialExecutionMode));
+    auto planNodeIdGenerator = std::make_shared<core::PlanNodeIdGenerator>();
+    AssertQueryBuilder(duckDbQueryRunner_)
+        .serialExecution(serialExecutionMode)
+        .plan(PlanBuilder(planNodeIdGenerator)
                   .localPartitionRoundRobin(
                       {PlanBuilder(planNodeIdGenerator)
                            .values({data1})
@@ -625,12 +629,11 @@ TEST_F(LocalPartitionTest, unionAllLocalExchange) {
                            .project({"e0 as c0"})
                            .planNode()})
                   .project({"length(c0)"})
-                  .planNode();
-
-  assertQuery(
-      plan,
-      "SELECT length(c0) FROM ("
-      "   SELECT * FROM (VALUES ('x')) as t1(c0) UNION ALL "
-      "   SELECT * FROM (VALUES ('y')) as t2(c0)"
-      ")");
+                  .planNode())
+        .assertResults(
+            "SELECT length(c0) FROM ("
+            "   SELECT * FROM (VALUES ('x')) as t1(c0) UNION ALL "
+            "   SELECT * FROM (VALUES ('y')) as t2(c0)"
+            ")");
+  }
 }