[native] Improve semi join parallelism

Run build side in parallel similar to other joins

Author: arhimondr

presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/AddLocalExchanges.java

@@ -868,7 +868,8 @@ public PlanWithProperties visitSemiJoin(SemiJoinNode node, StreamPreferredProper
                     parentPreferences.constrainTo(node.getSource().getOutputVariables()).withDefaultParallelism(session));
 
             // this filter source consumes the input completely, so we do not pass through parent preferences
-            PlanWithProperties filteringSource = planAndEnforce(node.getFilteringSource(), singleStream(), singleStream());
+            StreamPreferredProperties filteringPreference = nativeExecution ? defaultParallelism(session) : singleStream();
+            PlanWithProperties filteringSource = planAndEnforce(node.getFilteringSource(), filteringPreference, filteringPreference);
 
             return rebaseAndDeriveProperties(node, ImmutableList.of(source, filteringSource));
         }

presto-native-execution/src/test/java/com/facebook/presto/nativeworker/AbstractTestNativeJoinQueries.java

@@ -14,8 +14,10 @@
 package com.facebook.presto.nativeworker;
 
 import com.facebook.presto.Session;
+import com.facebook.presto.sql.analyzer.FeaturesConfig;
 import com.facebook.presto.testing.QueryRunner;
 import com.facebook.presto.tests.AbstractTestQueryFramework;
+import com.google.common.collect.ImmutableMap;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
@@ -28,6 +30,13 @@
 import static com.facebook.presto.nativeworker.NativeQueryRunnerUtils.createOrdersEx;
 import static com.facebook.presto.nativeworker.NativeQueryRunnerUtils.createPartitionedNation;
 import static com.facebook.presto.nativeworker.NativeQueryRunnerUtils.createRegion;
+import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.anyTree;
+import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.exchange;
+import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.semiJoin;
+import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.tableScan;
+import static com.facebook.presto.sql.planner.plan.ExchangeNode.Scope.REMOTE_STREAMING;
+import static com.facebook.presto.sql.planner.plan.ExchangeNode.Type.REPARTITION;
+import static com.facebook.presto.sql.planner.plan.ExchangeNode.Type.REPLICATE;
 
 public abstract class AbstractTestNativeJoinQueries
         extends AbstractTestQueryFramework
@@ -47,6 +56,12 @@ protected void createTables()
         createBucketedCustomer(queryRunner);
     }
 
+    @Override
+    protected FeaturesConfig createFeaturesConfig()
+    {
+        return new FeaturesConfig().setNativeExecutionEnabled(true);
+    }
+
     @Test(dataProvider = "joinTypeProvider")
     public void testInnerJoin(Session joinTypeSession)
     {
@@ -67,6 +82,35 @@ public void testBucketedInnerJoin(Session joinTypeSession)
                 "WHERE b.name=c.name AND \"$bucket\" = 5");
     }
 
+    @Test
+    public void testSemiJoinPlan()
+    {
+        String sql = "SELECT orderkey FROM orders WHERE orderdate IN (SELECT shipdate FROM lineitem)";
+        assertPlan(
+                partitionedJoin(),
+                sql,
+                anyTree(
+                        semiJoin("orderdate", "shipdate", "orderkey_1",
+                                exchange(REMOTE_STREAMING, REPARTITION,
+                                        tableScan("orders", ImmutableMap.of(
+                                                "orderkey", "orderkey",
+                                                "orderdate", "orderdate"))),
+                                exchange(REMOTE_STREAMING, REPARTITION,
+                                        tableScan("lineitem", ImmutableMap.of(
+                                                "shipdate", "shipdate"))))));
+        assertPlan(
+                broadcastJoin(),
+                sql,
+                anyTree(
+                        semiJoin("orderdate", "shipdate", "orderkey_1",
+                                tableScan("orders", ImmutableMap.of(
+                                        "orderkey", "orderkey",
+                                        "orderdate", "orderdate")),
+                                exchange(REMOTE_STREAMING, REPLICATE,
+                                        tableScan("lineitem", ImmutableMap.of(
+                                                "shipdate", "shipdate"))))));
+    }
+
     @Test(dataProvider = "joinTypeProvider")
     public void testSemiJoin(Session joinTypeSession)
     {