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-clickhouse/src/main/java/com/facebook/presto/plugin/clickhouse/ClickHouseConnector.java

@@ -113,8 +113,7 @@ public ConnectorPlanOptimizerProvider getConnectorPlanOptimizerProvider()
                 clickHouseClient,
                 functionManager,
                 functionResolution,
-                rowExpressionService.getDeterminismEvaluator(),
-                rowExpressionService.getExpressionOptimizer(),
+                rowExpressionService,
                 clickhouseQueryGenerator);
     }
 

presto-clickhouse/src/main/java/com/facebook/presto/plugin/clickhouse/optimization/ClickHouseComputePushdown.java

@@ -35,8 +35,8 @@
 import com.facebook.presto.spi.plan.PlanVisitor;
 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.RowExpression;
+import com.facebook.presto.spi.relation.RowExpressionService;
 import com.facebook.presto.spi.relation.VariableReferenceExpression;
 import com.google.common.collect.ImmutableList;
 
@@ -57,7 +57,7 @@
 public class ClickHouseComputePushdown
         implements ConnectorPlanOptimizer
 {
-    private final ExpressionOptimizer expressionOptimizer;
+    private final RowExpressionService rowExpressionService;
     private final ClickHouseFilterToSqlTranslator clickHouseFilterToSqlTranslator;
     private final LogicalRowExpressions logicalRowExpressions;
     private final ClickHouseQueryGenerator clickhouseQueryGenerator;
@@ -67,7 +67,7 @@ public ClickHouseComputePushdown(
             FunctionMetadataManager functionMetadataManager,
             StandardFunctionResolution functionResolution,
             DeterminismEvaluator determinismEvaluator,
-            ExpressionOptimizer expressionOptimizer,
+            RowExpressionService rowExpressionService,
             String identifierQuote,
             Set<Class<?>> functionTranslators,
             ClickHouseQueryGenerator clickhouseQueryGenerator)
@@ -78,7 +78,7 @@ public ClickHouseComputePushdown(
         requireNonNull(determinismEvaluator, "determinismEvaluator is null");
         requireNonNull(functionResolution, "functionResolution is null");
 
-        this.expressionOptimizer = requireNonNull(expressionOptimizer, "expressionOptimizer is null");
+        this.rowExpressionService = requireNonNull(rowExpressionService, "rowExpressionService is null");
         this.clickHouseFilterToSqlTranslator = new ClickHouseFilterToSqlTranslator(
                 functionMetadataManager,
                 buildFunctionTranslator(functionTranslators),
@@ -256,7 +256,7 @@ public PlanNode visitFilter(FilterNode node, Void context)
             TableHandle oldTableHandle = oldTableScanNode.getTable();
             ClickHouseTableHandle oldConnectorTable = (ClickHouseTableHandle) oldTableHandle.getConnectorHandle();
 
-            RowExpression predicate = expressionOptimizer.optimize(node.getPredicate(), OPTIMIZED, session);
+            RowExpression predicate = rowExpressionService.getExpressionOptimizer(session).optimize(node.getPredicate(), OPTIMIZED, session);
             predicate = logicalRowExpressions.convertToConjunctiveNormalForm(predicate);
             TranslatedExpression<ClickHouseExpression> clickHouseExpression = translateWith(
                     predicate,

presto-clickhouse/src/main/java/com/facebook/presto/plugin/clickhouse/optimization/ClickHousePlanOptimizerProvider.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.RowExpressionService;
 import com.google.common.collect.ImmutableSet;
 import com.google.inject.Inject;
 
@@ -33,8 +33,8 @@ public class ClickHousePlanOptimizerProvider
 {
     private final FunctionMetadataManager functionManager;
     private final StandardFunctionResolution functionResolution;
+    private final RowExpressionService rowExpressionService;
     private final DeterminismEvaluator determinismEvaluator;
-    private final ExpressionOptimizer expressionOptimizer;
     private final String identifierQuote;
     private final ClickHouseQueryGenerator clickhouseQueryGenerator;
 
@@ -43,16 +43,15 @@ public ClickHousePlanOptimizerProvider(
             ClickHouseClient clickHouseClient,
             FunctionMetadataManager functionManager,
             StandardFunctionResolution functionResolution,
-            DeterminismEvaluator determinismEvaluator,
-            ExpressionOptimizer expressionOptimizer,
+            RowExpressionService rowExpressionService,
             ClickHouseQueryGenerator clickhouseQueryGenerator)
     {
         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.rowExpressionService = requireNonNull(rowExpressionService, "rowExpressionService is null");
         this.identifierQuote = clickHouseClient.getIdentifierQuote();
         this.clickhouseQueryGenerator = clickhouseQueryGenerator;
+        this.determinismEvaluator = rowExpressionService.getDeterminismEvaluator();
     }
 
     @Override
@@ -68,7 +67,7 @@ public Set<ConnectorPlanOptimizer> getPhysicalPlanOptimizers()
                 functionManager,
                 functionResolution,
                 determinismEvaluator,
-                expressionOptimizer,
+                rowExpressionService,
                 identifierQuote,
                 getFunctionTranslators(),
                 clickhouseQueryGenerator));

presto-hive-common/src/main/java/com/facebook/presto/hive/MetadataUtils.java

@@ -85,7 +85,7 @@ public static RowExpression getSubfieldPredicate(
             StandardFunctionResolution functionResolution,
             RowExpressionService rowExpressionService)
     {
-        SubfieldExtractor subfieldExtractor = new SubfieldExtractor(functionResolution, rowExpressionService.getExpressionOptimizer(), session);
+        SubfieldExtractor subfieldExtractor = new SubfieldExtractor(functionResolution, rowExpressionService.getExpressionOptimizer(session), session);
 
         return rowExpressionService.getDomainTranslator().toPredicate(
                 layoutHandle.getDomainPredicate()

presto-hive-common/src/main/java/com/facebook/presto/hive/rule/BaseSubfieldExtractionRewriter.java

@@ -217,7 +217,7 @@ public ConnectorPushdownFilterResult pushdownFilter(
         ExtractionResult<Subfield> decomposedFilter = rowExpressionService.getDomainTranslator()
                 .fromPredicate(session, filter, new SubfieldExtractor(
                         functionResolution,
-                        rowExpressionService.getExpressionOptimizer(),
+                        rowExpressionService.getExpressionOptimizer(session),
                         session).toColumnExtractor());
 
         if (currentLayoutHandle.isPresent()) {
@@ -231,7 +231,7 @@ public ConnectorPushdownFilterResult pushdownFilter(
             return new ConnectorPushdownFilterResult(EMPTY_TABLE_LAYOUT, FALSE_CONSTANT);
         }
 
-        RowExpression optimizedRemainingExpression = rowExpressionService.getExpressionOptimizer()
+        RowExpression optimizedRemainingExpression = rowExpressionService.getExpressionOptimizer(session)
                 .optimize(decomposedFilter.getRemainingExpression(), OPTIMIZED, session);
         if (optimizedRemainingExpression instanceof ConstantExpression) {
             ConstantExpression constantExpression = (ConstantExpression) optimizedRemainingExpression;
@@ -438,7 +438,7 @@ private boolean isCandidate(Map<ColumnHandle, NullableValue> bindings)
             // spurious query failures for partitions that would otherwise be filtered out.
             RowExpression optimized;
             try {
-                optimized = evaluator.getExpressionOptimizer().optimize(expression, OPTIMIZED, session, variableResolver);
+                optimized = evaluator.getExpressionOptimizer(session).optimize(expression, OPTIMIZED, session, variableResolver);
             }
             catch (PrestoException e) {
                 propagateIfUnhandled(e);

presto-hive/src/main/java/com/facebook/presto/hive/FilteringPageSource.java

@@ -118,7 +118,7 @@ public FilteringPageSource(
                         columnHandle -> new VariableReferenceExpression(Optional.empty(), columnHandle.getName(), columnHandle.getHiveType().getType(typeManager)),
                         columnHandle -> new InputReferenceExpression(Optional.empty(), columnHandle.getHiveColumnIndex(), columnHandle.getHiveType().getType(typeManager))));
 
-        RowExpression optimizedRemainingPredicate = rowExpressionService.getExpressionOptimizer().optimize(remainingPredicate, OPTIMIZED, session);
+        RowExpression optimizedRemainingPredicate = rowExpressionService.getExpressionOptimizer(session).optimize(remainingPredicate, OPTIMIZED, session);
         if (TRUE_CONSTANT.equals(optimizedRemainingPredicate)) {
             this.filterFunction = null;
         }

presto-hive/src/main/java/com/facebook/presto/hive/HiveMetadata.java

@@ -888,7 +888,7 @@ public TableStatistics getTableStatistics(ConnectorSession session, ConnectorTab
                 .transform(subfield -> isEntireColumn(subfield) ? subfield.getRootName() : null)
                 .transform(allColumns::get)));
 
-        SubfieldExtractor subfieldExtractor = new SubfieldExtractor(functionResolution, rowExpressionService.getExpressionOptimizer(), session);
+        SubfieldExtractor subfieldExtractor = new SubfieldExtractor(functionResolution, rowExpressionService.getExpressionOptimizer(session), session);
 
         RowExpression domainPredicate = rowExpressionService.getDomainTranslator().toPredicate(
                 hiveLayoutHandle.getDomainPredicate()

presto-hive/src/main/java/com/facebook/presto/hive/HivePageSourceProvider.java

@@ -129,7 +129,7 @@ public HivePageSourceProvider(
         this.optimizedRowExpressionCache = CacheBuilder.newBuilder()
                 .recordStats()
                 .maximumSize(10_000)
-                .build(CacheLoader.from(cacheKey -> rowExpressionService.getExpressionOptimizer().optimize(cacheKey.rowExpression, OPTIMIZED, cacheKey.session)));
+                .build(CacheLoader.from(cacheKey -> rowExpressionService.getExpressionOptimizer(cacheKey.session).optimize(cacheKey.rowExpression, OPTIMIZED, cacheKey.session)));
     }
 
     @Override

presto-hive/src/main/java/com/facebook/presto/hive/orc/OrcSelectivePageSourceFactory.java

@@ -425,7 +425,7 @@ private static Map<Integer, List<Subfield>> collectRequiredSubfields(List<HiveCo
                 .forEach(column -> outputSubfields.put(column.getHiveColumnIndex(), new HashSet<>(column.getRequiredSubfields())));
 
         Map<Integer, Set<Subfield>> predicateSubfields = new HashMap<>();
-        SubfieldExtractor subfieldExtractor = new SubfieldExtractor(functionResolution, rowExpressionService.getExpressionOptimizer(), session);
+        SubfieldExtractor subfieldExtractor = new SubfieldExtractor(functionResolution, rowExpressionService.getExpressionOptimizer(session), session);
         remainingPredicate.accept(
                 new RequiredSubfieldsExtractor(subfieldExtractor),
                 subfield -> predicateSubfields.computeIfAbsent(columnIndices.get(subfield.getRootName()), v -> new HashSet<>()).add(subfield));

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

@@ -524,7 +524,7 @@ private static RowType toRowType(List<ColumnMetadata> columns)
                             }).collect(toList()))
                     .build();
 
-    private static final SubfieldExtractor SUBFIELD_EXTRACTOR = new SubfieldExtractor(FUNCTION_RESOLUTION, ROW_EXPRESSION_SERVICE.getExpressionOptimizer(), SESSION);
+    private static final SubfieldExtractor SUBFIELD_EXTRACTOR = new SubfieldExtractor(FUNCTION_RESOLUTION, ROW_EXPRESSION_SERVICE.getExpressionOptimizer(SESSION), SESSION);
 
     private static final TypeProvider TYPE_PROVIDER_AFTER = TypeProvider.copyOf(MISMATCH_SCHEMA_TABLE_AFTER.stream()
             .collect(toImmutableMap(ColumnMetadata::getName, ColumnMetadata::getType)));

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

@@ -126,7 +126,7 @@ public DomainTranslator getDomainTranslator()
         }
 
         @Override
-        public ExpressionOptimizer getExpressionOptimizer()
+        public ExpressionOptimizer getExpressionOptimizer(ConnectorSession session)
         {
             return new RowExpressionOptimizer(METADATA);
         }
@@ -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-iceberg/src/main/java/com/facebook/presto/iceberg/optimizer/IcebergMetadataOptimizer.java

@@ -356,7 +356,7 @@ else if (scalarFunctionName.equals("least")) {
                         throw new PrestoException(StandardErrorCode.NOT_SUPPORTED, "unsupported function: " + scalarFunctionName);
                     }
 
-                    RowExpression reducedValue = rowExpressionService.getExpressionOptimizer().optimize(
+                    RowExpression reducedValue = rowExpressionService.getExpressionOptimizer(connectorSession).optimize(
                             new CallExpression(
                                     Optional.empty(),
                                     scalarFunctionName,