Use ExpressionOptimizerProvider

The runtime should consolidate to the `ExpressionOptimizerProvider` factory
so that it can be customized without significant refactoring.

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/JdbcConnector.java

@@ -109,7 +109,7 @@ public ConnectorPlanOptimizerProvider getConnectorPlanOptimizerProvider()
                 functionManager,
                 functionResolution,
                 rowExpressionService.getDeterminismEvaluator(),
-                rowExpressionService.getExpressionOptimizer());
+                rowExpressionService);
     }
 
     @Override

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/optimization/JdbcComputePushdown.java

@@ -29,7 +29,7 @@
 import com.facebook.presto.spi.plan.PlanNodeIdAllocator;
 import com.facebook.presto.spi.plan.TableScanNode;
 import com.facebook.presto.spi.relation.DeterminismEvaluator;
-import com.facebook.presto.spi.relation.ExpressionOptimizer;
+import com.facebook.presto.spi.relation.ExpressionOptimizerProvider;
 import com.facebook.presto.spi.relation.RowExpression;
 
 import java.util.Optional;
@@ -44,15 +44,15 @@
 public class JdbcComputePushdown
         implements ConnectorPlanOptimizer
 {
-    private final ExpressionOptimizer expressionOptimizer;
+    private final ExpressionOptimizerProvider expressionOptimizerProvider;
     private final JdbcFilterToSqlTranslator jdbcFilterToSqlTranslator;
     private final LogicalRowExpressions logicalRowExpressions;
 
     public JdbcComputePushdown(
             FunctionMetadataManager functionMetadataManager,
             StandardFunctionResolution functionResolution,
             DeterminismEvaluator determinismEvaluator,
-            ExpressionOptimizer expressionOptimizer,
+            ExpressionOptimizerProvider expressionOptimizerProvider,
             String identifierQuote,
             Set<Class<?>> functionTranslators)
     {
@@ -62,7 +62,7 @@ public JdbcComputePushdown(
         requireNonNull(determinismEvaluator, "determinismEvaluator is null");
         requireNonNull(functionResolution, "functionResolution is null");
 
-        this.expressionOptimizer = requireNonNull(expressionOptimizer, "expressionOptimizer is null");
+        this.expressionOptimizerProvider = requireNonNull(expressionOptimizerProvider, "expressionOptimizerProvider is null");
         this.jdbcFilterToSqlTranslator = new JdbcFilterToSqlTranslator(
                 functionMetadataManager,
                 buildFunctionTranslator(functionTranslators),
@@ -106,7 +106,7 @@ public PlanNode visitFilter(FilterNode node, RewriteContext<Void> context)
             TableHandle oldTableHandle = oldTableScanNode.getTable();
             JdbcTableHandle oldConnectorTable = (JdbcTableHandle) oldTableHandle.getConnectorHandle();
 
-            RowExpression predicate = expressionOptimizer.optimize(node.getPredicate(), OPTIMIZED, session);
+            RowExpression predicate = expressionOptimizerProvider.getExpressionOptimizer(session).optimize(node.getPredicate(), OPTIMIZED, session);
             predicate = logicalRowExpressions.convertToConjunctiveNormalForm(predicate);
             TranslatedExpression<JdbcExpression> jdbcExpression = translateWith(
                     predicate,

presto-base-jdbc/src/main/java/com/facebook/presto/plugin/jdbc/optimization/JdbcPlanOptimizerProvider.java

@@ -20,7 +20,7 @@
 import com.facebook.presto.spi.function.FunctionMetadataManager;
 import com.facebook.presto.spi.function.StandardFunctionResolution;
 import com.facebook.presto.spi.relation.DeterminismEvaluator;
-import com.facebook.presto.spi.relation.ExpressionOptimizer;
+import com.facebook.presto.spi.relation.ExpressionOptimizerProvider;
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.Inject;
 
@@ -34,7 +34,7 @@ public class JdbcPlanOptimizerProvider
     private final FunctionMetadataManager functionManager;
     private final StandardFunctionResolution functionResolution;
     private final DeterminismEvaluator determinismEvaluator;
-    private final ExpressionOptimizer expressionOptimizer;
+    private final ExpressionOptimizerProvider expressionOptimizerProvider;
     private final String identifierQuote;
 
     @Inject
@@ -43,12 +43,12 @@ public JdbcPlanOptimizerProvider(
             FunctionMetadataManager functionManager,
             StandardFunctionResolution functionResolution,
             DeterminismEvaluator determinismEvaluator,
-            ExpressionOptimizer expressionOptimizer)
+            ExpressionOptimizerProvider expressionOptimizerProvider)
     {
         this.functionManager = requireNonNull(functionManager, "functionManager is null");
         this.functionResolution = requireNonNull(functionResolution, "functionResolution is null");
         this.determinismEvaluator = requireNonNull(determinismEvaluator, "determinismEvaluator is null");
-        this.expressionOptimizer = requireNonNull(expressionOptimizer, "expressionOptimizer is null");
+        this.expressionOptimizerProvider = requireNonNull(expressionOptimizerProvider, "expressionOptimizerProvider is null");
         this.identifierQuote = jdbcClient.getIdentifierQuote();
     }
 
@@ -65,7 +65,7 @@ public Set<ConnectorPlanOptimizer> getPhysicalPlanOptimizers()
                 functionManager,
                 functionResolution,
                 determinismEvaluator,
-                expressionOptimizer,
+                expressionOptimizerProvider,
                 identifierQuote,
                 getFunctionTranslators()));
     }

presto-base-jdbc/src/test/java/com/facebook/presto/plugin/jdbc/optimization/TestJdbcComputePushdown.java

@@ -105,7 +105,7 @@ public TestJdbcComputePushdown()
                 functionAndTypeManager,
                 functionResolution,
                 determinismEvaluator,
-                new RowExpressionOptimizer(METADATA),
+                (ConnectorSession session) -> new RowExpressionOptimizer(METADATA),
                 "'",
                 getFunctionTranslators());
     }

presto-hive/src/test/java/com/facebook/presto/hive/HiveTestUtils.java

@@ -151,7 +151,7 @@ public String formatRowExpression(ConnectorSession session, RowExpression expres
     };
 
     public static final FilterStatsCalculatorService FILTER_STATS_CALCULATOR_SERVICE = new ConnectorFilterStatsCalculatorService(
-            new FilterStatsCalculator(METADATA, new ScalarStatsCalculator(METADATA), new StatsNormalizer()));
+            new FilterStatsCalculator(METADATA, new ScalarStatsCalculator(METADATA, ROW_EXPRESSION_SERVICE), new StatsNormalizer()));
 
     public static final HiveClientConfig HIVE_CLIENT_CONFIG = new HiveClientConfig();
     public static final MetastoreClientConfig METASTORE_CLIENT_CONFIG = new MetastoreClientConfig();

presto-main/src/main/java/com/facebook/presto/connector/ConnectorManager.java

@@ -62,12 +62,12 @@
 import com.facebook.presto.split.RecordPageSourceProvider;
 import com.facebook.presto.split.SplitManager;
 import com.facebook.presto.sql.analyzer.FeaturesConfig;
+import com.facebook.presto.sql.expressions.ExpressionOptimizerManager;
 import com.facebook.presto.sql.planner.ConnectorPlanOptimizerManager;
 import com.facebook.presto.sql.planner.PartitioningProviderManager;
 import com.facebook.presto.sql.planner.planPrinter.RowExpressionFormatter;
 import com.facebook.presto.sql.relational.ConnectorRowExpressionService;
 import com.facebook.presto.sql.relational.FunctionResolution;
-import com.facebook.presto.sql.relational.RowExpressionOptimizer;
 import com.facebook.presto.transaction.TransactionManager;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -116,6 +116,7 @@ public class ConnectorManager
     private final PageIndexerFactory pageIndexerFactory;
     private final NodeInfo nodeInfo;
     private final TransactionManager transactionManager;
+    private final ExpressionOptimizerManager expressionOptimizerManager;
     private final DomainTranslator domainTranslator;
     private final PredicateCompiler predicateCompiler;
     private final DeterminismEvaluator determinismEvaluator;
@@ -151,6 +152,7 @@ public ConnectorManager(
             PageSorter pageSorter,
             PageIndexerFactory pageIndexerFactory,
             TransactionManager transactionManager,
+            ExpressionOptimizerManager expressionOptimizerManager,
             DomainTranslator domainTranslator,
             PredicateCompiler predicateCompiler,
             DeterminismEvaluator determinismEvaluator,
@@ -176,6 +178,7 @@ public ConnectorManager(
         this.pageIndexerFactory = requireNonNull(pageIndexerFactory, "pageIndexerFactory is null");
         this.nodeInfo = requireNonNull(nodeInfo, "nodeInfo is null");
         this.transactionManager = requireNonNull(transactionManager, "transactionManager is null");
+        this.expressionOptimizerManager = requireNonNull(expressionOptimizerManager, "expressionOptimizerManager is null");
         this.domainTranslator = requireNonNull(domainTranslator, "domainTranslator is null");
         this.predicateCompiler = requireNonNull(predicateCompiler, "predicateCompiler is null");
         this.determinismEvaluator = requireNonNull(determinismEvaluator, "determinismEvaluator is null");
@@ -382,7 +385,7 @@ private Connector createConnector(ConnectorId connectorId, ConnectorFactory fact
                 pageIndexerFactory,
                 new ConnectorRowExpressionService(
                         domainTranslator,
-                        new RowExpressionOptimizer(metadataManager),
+                        expressionOptimizerManager,
                         predicateCompiler,
                         determinismEvaluator,
                         new RowExpressionFormatter(metadataManager.getFunctionAndTypeManager())),

presto-main/src/main/java/com/facebook/presto/cost/ScalarStatsCalculator.java

@@ -23,6 +23,7 @@
 import com.facebook.presto.spi.function.FunctionMetadata;
 import com.facebook.presto.spi.relation.CallExpression;
 import com.facebook.presto.spi.relation.ConstantExpression;
+import com.facebook.presto.spi.relation.ExpressionOptimizerProvider;
 import com.facebook.presto.spi.relation.InputReferenceExpression;
 import com.facebook.presto.spi.relation.LambdaDefinitionExpression;
 import com.facebook.presto.spi.relation.RowExpression;
@@ -78,13 +79,18 @@
 public class ScalarStatsCalculator
 {
     private final Metadata metadata;
-    private final ExpressionOptimizerManager expressionOptimizerManager;
+    private final ExpressionOptimizerProvider expressionOptimizerProvider;
 
     @Inject
     public ScalarStatsCalculator(Metadata metadata, ExpressionOptimizerManager expressionOptimizerManager)
+    {
+        this(metadata, (ExpressionOptimizerProvider) expressionOptimizerManager);
+    }
+
+    public ScalarStatsCalculator(Metadata metadata, ExpressionOptimizerProvider expressionOptimizerProvider)
     {
         this.metadata = requireNonNull(metadata, "metadata can not be null");
-        this.expressionOptimizerManager = requireNonNull(expressionOptimizerManager, "expressionOptimizerManager can not be null");
+        this.expressionOptimizerProvider = requireNonNull(expressionOptimizerProvider, "expressionOptimizerManager can not be null");
     }
 
     @Deprecated
@@ -128,7 +134,7 @@ public VariableStatsEstimate visitCall(CallExpression call, Void context)
                 return computeArithmeticBinaryStatistics(call, context);
             }
 
-            RowExpression value = expressionOptimizerManager.getExpressionOptimizer(session).optimize(call, OPTIMIZED, session);
+            RowExpression value = expressionOptimizerProvider.getExpressionOptimizer(session).optimize(call, OPTIMIZED, session);
 
             if (isNull(value)) {
                 return nullStatsEstimate();

presto-main/src/main/java/com/facebook/presto/sql/planner/PlanOptimizers.java

@@ -350,7 +350,7 @@ public PlanOptimizers(
                 estimatedExchangesCostCalculator,
                 new RewriteConstantArrayContainsToInExpression(metadata.getFunctionAndTypeManager()).rules());
 
-        PlanOptimizer predicatePushDown = new StatsRecordingPlanOptimizer(optimizerStats, new PredicatePushDown(metadata, sqlParser, featuresConfig.isNativeExecutionEnabled()));
+        PlanOptimizer predicatePushDown = new StatsRecordingPlanOptimizer(optimizerStats, new PredicatePushDown(metadata, sqlParser, expressionOptimizerManager, featuresConfig.isNativeExecutionEnabled()));
         PlanOptimizer prefilterForLimitingAggregation = new StatsRecordingPlanOptimizer(optimizerStats, new PrefilterForLimitingAggregation(metadata, statsCalculator));
 
         builder.add(
@@ -731,7 +731,7 @@ public PlanOptimizers(
                         statsCalculator,
                         estimatedExchangesCostCalculator,
                         ImmutableSet.of(new RemoveRedundantIdentityProjections(), new PruneRedundantProjectionAssignments())),
-                new PushdownSubfields(metadata));
+                new PushdownSubfields(metadata, expressionOptimizerManager));
 
         builder.add(predicatePushDown); // Run predicate push down one more time in case we can leverage new information from layouts' effective predicate
         builder.add(simplifyRowExpressionOptimizer); // Should be always run after PredicatePushDown