Add a session property to disable ReplicateSemiJoinInDelete optimizer

presto-main-base/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -328,6 +328,7 @@ public final class SystemSessionProperties
     public static final String INLINE_PROJECTIONS_ON_VALUES = "inline_projections_on_values";
     public static final String INCLUDE_VALUES_NODE_IN_CONNECTOR_OPTIMIZER = "include_values_node_in_connector_optimizer";
     public static final String SINGLE_NODE_EXECUTION_ENABLED = "single_node_execution_enabled";
+    public static final String BROADCAST_SEMI_JOIN_FOR_DELETE = "broadcast_semi_join_for_delete";
     public static final String EXPRESSION_OPTIMIZER_NAME = "expression_optimizer_name";
     public static final String ADD_EXCHANGE_BELOW_PARTIAL_AGGREGATION_OVER_GROUP_ID = "add_exchange_below_partial_aggregation_over_group_id";
     public static final String QUERY_CLIENT_TIMEOUT = "query_client_timeout";
@@ -1895,6 +1896,10 @@ public SystemSessionProperties(
                         "Configure which expression optimizer to use",
                         featuresConfig.getExpressionOptimizerName(),
                         false),
+                booleanProperty(BROADCAST_SEMI_JOIN_FOR_DELETE,
+                        "Enforce broadcast join for semi join in delete",
+                        featuresConfig.isBroadcastSemiJoinForDelete(),
+                        false),
                 booleanProperty(ADD_EXCHANGE_BELOW_PARTIAL_AGGREGATION_OVER_GROUP_ID,
                         "Enable adding an exchange below partial aggregation over a GroupId node to improve partial aggregation performance",
                         featuresConfig.getAddExchangeBelowPartialAggregationOverGroupId(),
@@ -3238,6 +3243,11 @@ public static String getExpressionOptimizerName(Session session)
         return session.getSystemProperty(EXPRESSION_OPTIMIZER_NAME, String.class);
     }
 
+    public static boolean isBroadcastSemiJoinForDeleteEnabled(Session session)
+    {
+        return session.getSystemProperty(BROADCAST_SEMI_JOIN_FOR_DELETE, Boolean.class);
+    }
+
     public static boolean isEnabledAddExchangeBelowGroupId(Session session)
     {
         return session.getSystemProperty(ADD_EXCHANGE_BELOW_PARTIAL_AGGREGATION_OVER_GROUP_ID, Boolean.class);

presto-main-base/src/main/java/com/facebook/presto/sql/analyzer/FeaturesConfig.java

@@ -296,6 +296,7 @@ public class FeaturesConfig
     private boolean innerJoinPushdownEnabled;
     private boolean inEqualityJoinPushdownEnabled;
     private boolean rewriteMinMaxByToTopNEnabled;
+    private boolean broadcastSemiJoinForDelete = true;
 
     private boolean prestoSparkExecutionEnvironment;
     private boolean singleNodeExecutionEnabled;
@@ -2924,6 +2925,19 @@ public FeaturesConfig setRewriteMinMaxByToTopNEnabled(boolean rewriteMinMaxByToT
         return this;
     }
 
+    public boolean isBroadcastSemiJoinForDelete()
+    {
+        return broadcastSemiJoinForDelete;
+    }
+
+    @Config("optimizer.broadcast-semi-join-for-delete")
+    @ConfigDescription("Enforce broadcast semi join in delete queries")
+    public FeaturesConfig setBroadcastSemiJoinForDelete(boolean broadcastSemiJoinForDelete)
+    {
+        this.broadcastSemiJoinForDelete = broadcastSemiJoinForDelete;
+        return this;
+    }
+
     public boolean isInEqualityJoinPushdownEnabled()
     {
         return inEqualityJoinPushdownEnabled;

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

@@ -23,6 +23,7 @@
 import com.facebook.presto.sql.planner.TypeProvider;
 import com.facebook.presto.sql.planner.plan.SimplePlanRewriter;
 
+import static com.facebook.presto.SystemSessionProperties.isBroadcastSemiJoinForDeleteEnabled;
 import static com.facebook.presto.spi.plan.SemiJoinNode.DistributionType.REPLICATED;
 import static java.util.Objects.requireNonNull;
 
@@ -32,10 +33,13 @@ public class ReplicateSemiJoinInDelete
     @Override
     public PlanOptimizerResult optimize(PlanNode plan, Session session, TypeProvider types, VariableAllocator variableAllocator, PlanNodeIdAllocator idAllocator, WarningCollector warningCollector)
     {
-        requireNonNull(plan, "plan is null");
-        Rewriter rewriter = new Rewriter();
-        PlanNode rewrittenPlan = SimplePlanRewriter.rewriteWith(rewriter, plan);
-        return PlanOptimizerResult.optimizerResult(rewrittenPlan, rewriter.isPlanChanged());
+        if (isBroadcastSemiJoinForDeleteEnabled(session)) {
+            requireNonNull(plan, "plan is null");
+            Rewriter rewriter = new Rewriter();
+            PlanNode rewrittenPlan = SimplePlanRewriter.rewriteWith(rewriter, plan);
+            return PlanOptimizerResult.optimizerResult(rewrittenPlan, rewriter.isPlanChanged());
+        }
+        return PlanOptimizerResult.optimizerResult(plan, false);
     }
 
     private static class Rewriter

presto-main-base/src/test/java/com/facebook/presto/sql/analyzer/TestFeaturesConfig.java

@@ -259,6 +259,7 @@ public void testDefaults()
                 .setAddExchangeBelowPartialAggregationOverGroupId(false)
                 .setAddDistinctBelowSemiJoinBuild(false)
                 .setInnerJoinPushdownEnabled(false)
+                .setBroadcastSemiJoinForDelete(true)
                 .setInEqualityJoinPushdownEnabled(false)
                 .setRewriteMinMaxByToTopNEnabled(false)
                 .setPrestoSparkExecutionEnvironment(false));
@@ -459,6 +460,7 @@ public void testExplicitPropertyMappings()
                 .put("eager-plan-validation-enabled", "true")
                 .put("eager-plan-validation-thread-pool-size", "2")
                 .put("optimizer.inner-join-pushdown-enabled", "true")
+                .put("optimizer.broadcast-semi-join-for-delete", "false")
                 .put("optimizer.inequality-join-pushdown-enabled", "true")
                 .put("optimizer.rewrite-minBy-maxBy-to-topN-enabled", "true")
                 .put("presto-spark-execution-environment", "true")
@@ -674,6 +676,7 @@ public void testExplicitPropertyMappings()
                 .setAddExchangeBelowPartialAggregationOverGroupId(true)
                 .setAddDistinctBelowSemiJoinBuild(true)
                 .setInEqualityJoinPushdownEnabled(true)
+                .setBroadcastSemiJoinForDelete(false)
                 .setRewriteMinMaxByToTopNEnabled(true)
                 .setInnerJoinPushdownEnabled(true)
                 .setPrestoSparkExecutionEnvironment(true);