Add session property for connector optimizer to work on values node

Author: feilong-liu

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

@@ -324,6 +324,7 @@ public final class SystemSessionProperties
     public static final String OPTIMIZER_USE_HISTOGRAMS = "optimizer_use_histograms";
     public static final String WARN_ON_COMMON_NAN_PATTERNS = "warn_on_common_nan_patterns";
     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";
 
     // TODO: Native execution related session properties that are temporarily put here. They will be relocated in the future.
     public static final String NATIVE_AGGREGATION_SPILL_ALL = "native_aggregation_spill_all";
@@ -1820,6 +1821,10 @@ public SystemSessionProperties(
                         "Whether to evaluate project node on values node",
                         featuresConfig.getInlineProjectionsOnValues(),
                         false),
+                booleanProperty(INCLUDE_VALUES_NODE_IN_CONNECTOR_OPTIMIZER,
+                        "Include values node for connector optimizer",
+                        featuresConfig.isIncludeValuesNodeInConnectorOptimizer(),
+                        false),
                 integerProperty(
                         NATIVE_MIN_COLUMNAR_ENCODING_CHANNELS_TO_PREFER_ROW_WISE_ENCODING,
                         "Minimum number of columnar encoding channels to consider row wise encoding for partitioned exchange. Native execution only",
@@ -3096,6 +3101,11 @@ public static boolean isInlineProjectionsOnValues(Session session)
         return session.getSystemProperty(INLINE_PROJECTIONS_ON_VALUES, Boolean.class);
     }
 
+    public static boolean isIncludeValuesNodeInConnectorOptimizer(Session session)
+    {
+        return session.getSystemProperty(INCLUDE_VALUES_NODE_IN_CONNECTOR_OPTIMIZER, Boolean.class);
+    }
+
     public static int getMinColumnarEncodingChannelsToPreferRowWiseEncoding(Session session)
     {
         return session.getSystemProperty(NATIVE_MIN_COLUMNAR_ENCODING_CHANNELS_TO_PREFER_ROW_WISE_ENCODING, Integer.class);

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

@@ -283,6 +283,7 @@ public class FeaturesConfig
     private boolean useHistograms;
 
     private boolean isInlineProjectionsOnValuesEnabled;
+    private boolean includeValuesNodeInConnectorOptimizer = true;
 
     private boolean eagerPlanValidationEnabled;
     private int eagerPlanValidationThreadPoolSize = 20;
@@ -2810,6 +2811,19 @@ public FeaturesConfig setInlineProjectionsOnValues(boolean isInlineProjectionsOn
         return this;
     }
 
+    public boolean isIncludeValuesNodeInConnectorOptimizer()
+    {
+        return includeValuesNodeInConnectorOptimizer;
+    }
+
+    @Config("optimizer.include-values-node-in-connector-optimizer")
+    @ConfigDescription("Include values node in connector optimizer")
+    public FeaturesConfig setIncludeValuesNodeInConnectorOptimizer(boolean includeValuesNodeInConnectorOptimizer)
+    {
+        this.includeValuesNodeInConnectorOptimizer = includeValuesNodeInConnectorOptimizer;
+        return this;
+    }
+
     @Config("eager-plan-validation-enabled")
     @ConfigDescription("Enable eager building and validation of logical plan before queueing")
     public FeaturesConfig setEagerPlanValidationEnabled(boolean eagerPlanValidationEnabled)

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

@@ -57,11 +57,13 @@
 import java.util.Queue;
 import java.util.Set;
 
+import static com.facebook.presto.SystemSessionProperties.isIncludeValuesNodeInConnectorOptimizer;
 import static com.facebook.presto.common.RuntimeUnit.NANO;
 import static com.facebook.presto.sql.OptimizerRuntimeTrackUtil.getOptimizerNameForLog;
 import static com.facebook.presto.sql.OptimizerRuntimeTrackUtil.trackOptimizerRuntime;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static java.util.Objects.requireNonNull;
 
 public class ApplyConnectorOptimization
@@ -144,9 +146,9 @@ public PlanOptimizerResult optimize(PlanNode plan, Session session, TypeProvider
                 //    * The subtree with root `node` is a closure.
                 //    * `node` has no parent, or the subtree with root as `node`'s parent is not a closure.
                 ConnectorPlanNodeContext context = contextMap.get(node);
-                if (!context.isClosure(connectorId) ||
+                if (!context.isClosure(connectorId, session) ||
                         !context.getParent().isPresent() ||
-                        contextMap.get(context.getParent().get()).isClosure(connectorId)) {
+                        contextMap.get(context.getParent().get()).isClosure(connectorId, session)) {
                     continue;
                 }
 
@@ -293,10 +295,12 @@ public Set<Class<? extends PlanNode>> getReachablePlanNodeTypes()
             return reachablePlanNodeTypes;
         }
 
-        boolean isClosure(ConnectorId connectorId)
+        boolean isClosure(ConnectorId connectorId, Session session)
         {
             // check if all children can reach the only connector
-            if (reachableConnectors.size() != 1 || !reachableConnectors.contains(connectorId)) {
+            boolean includeValuesNode = isIncludeValuesNodeInConnectorOptimizer(session);
+            Set<ConnectorId> connectorIds = includeValuesNode ? reachableConnectors.stream().filter(x -> !x.equals(EMPTY_CONNECTOR_ID)).collect(toImmutableSet()) : reachableConnectors;
+            if (connectorIds.size() != 1 || !connectorIds.contains(connectorId)) {
                 return false;
             }
 

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

@@ -244,6 +244,7 @@ public void testDefaults()
                 .setRemoveCrossJoinWithSingleConstantRow(true)
                 .setUseHistograms(false)
                 .setInlineProjectionsOnValues(false)
+                .setIncludeValuesNodeInConnectorOptimizer(true)
                 .setEagerPlanValidationEnabled(false)
                 .setEagerPlanValidationThreadPoolSize(20)
                 .setPrestoSparkExecutionEnvironment(false));
@@ -439,6 +440,7 @@ public void testExplicitPropertyMappings()
                 .put("optimizer.remove-cross-join-with-single-constant-row", "false")
                 .put("optimizer.use-histograms", "true")
                 .put("optimizer.inline-projections-on-values", "true")
+                .put("optimizer.include-values-node-in-connector-optimizer", "false")
                 .put("eager-plan-validation-enabled", "true")
                 .put("eager-plan-validation-thread-pool-size", "2")
                 .put("presto-spark-execution-environment", "true")
@@ -632,6 +634,7 @@ public void testExplicitPropertyMappings()
                 .setRemoveCrossJoinWithSingleConstantRow(false)
                 .setUseHistograms(true)
                 .setInlineProjectionsOnValues(true)
+                .setIncludeValuesNodeInConnectorOptimizer(false)
                 .setEagerPlanValidationEnabled(true)
                 .setEagerPlanValidationThreadPoolSize(2)
                 .setPrestoSparkExecutionEnvironment(true);