Add SPI for delegating row expression optimizer

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, "expressionOptimizer 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().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, "expressionOptimizer 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),
+                () -> 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/SystemSessionProperties.java

@@ -325,6 +325,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 DELEGATING_ROW_EXPRESSION_OPTIMIZER_ENABLED = "delegating_row_expression_optimizer_enabled";
 
     // 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";
@@ -1830,6 +1831,11 @@ public SystemSessionProperties(
                         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",
                         queryManagerConfig.getMinColumnarEncodingChannelsToPreferRowWiseEncoding(),
+                        false),
+                booleanProperty(
+                        DELEGATING_ROW_EXPRESSION_OPTIMIZER_ENABLED,
+                        "Enable delegating row optimizer",
+                        featuresConfig.isDelegatingRowExpressionOptimizerEnabled(),
                         false));
     }
 
@@ -3111,4 +3117,9 @@ public static int getMinColumnarEncodingChannelsToPreferRowWiseEncoding(Session
     {
         return session.getSystemProperty(NATIVE_MIN_COLUMNAR_ENCODING_CHANNELS_TO_PREFER_ROW_WISE_ENCODING, Integer.class);
     }
+
+    public static boolean isDelegatingRowExpressionOptimizerEnabled(Session session)
+    {
+        return session.getSystemProperty(DELEGATING_ROW_EXPRESSION_OPTIMIZER_ENABLED, Boolean.class);
+    }
 }

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;
@@ -31,11 +32,11 @@
 import com.facebook.presto.spi.relation.VariableReferenceExpression;
 import com.facebook.presto.sql.analyzer.ExpressionAnalyzer;
 import com.facebook.presto.sql.analyzer.Scope;
+import com.facebook.presto.sql.expressions.ExpressionOptimizerManager;
 import com.facebook.presto.sql.planner.ExpressionInterpreter;
 import com.facebook.presto.sql.planner.NoOpVariableResolver;
 import com.facebook.presto.sql.planner.TypeProvider;
 import com.facebook.presto.sql.relational.FunctionResolution;
-import com.facebook.presto.sql.relational.RowExpressionOptimizer;
 import com.facebook.presto.sql.tree.ArithmeticBinaryExpression;
 import com.facebook.presto.sql.tree.ArithmeticUnaryExpression;
 import com.facebook.presto.sql.tree.AstVisitor;
@@ -78,11 +79,18 @@
 public class ScalarStatsCalculator
 {
     private final Metadata metadata;
+    private final ExpressionOptimizerProvider expressionOptimizerProvider;
 
     @Inject
-    public ScalarStatsCalculator(Metadata metadata)
+    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.expressionOptimizerProvider = requireNonNull(expressionOptimizerProvider, "expressionOptimizerManager can not be null");
     }
 
     @Deprecated
@@ -126,7 +134,7 @@ public VariableStatsEstimate visitCall(CallExpression call, Void context)
                 return computeArithmeticBinaryStatistics(call, context);
             }
 
-            RowExpression value = new RowExpressionOptimizer(metadata).optimize(call, OPTIMIZED, session);
+            RowExpression value = expressionOptimizerProvider.getExpressionOptimizer().optimize(call, OPTIMIZED, session);
 
             if (isNull(value)) {
                 return nullStatsEstimate();

presto-main/src/main/java/com/facebook/presto/server/PluginManager.java

@@ -43,13 +43,15 @@
 import com.facebook.presto.spi.security.SystemAccessControlFactory;
 import com.facebook.presto.spi.session.SessionPropertyConfigurationManagerFactory;
 import com.facebook.presto.spi.session.WorkerSessionPropertyProviderFactory;
+import com.facebook.presto.spi.sql.planner.ExpressionOptimizerFactory;
 import com.facebook.presto.spi.statistics.HistoryBasedPlanStatisticsProvider;
 import com.facebook.presto.spi.storage.TempStorageFactory;
 import com.facebook.presto.spi.tracing.TracerProvider;
 import com.facebook.presto.spi.ttl.ClusterTtlProviderFactory;
 import com.facebook.presto.spi.ttl.NodeTtlFetcherFactory;
 import com.facebook.presto.sql.analyzer.AnalyzerProviderManager;
 import com.facebook.presto.sql.analyzer.QueryPreparerProviderManager;
+import com.facebook.presto.sql.expressions.ExpressionOptimizerManager;
 import com.facebook.presto.sql.planner.sanity.PlanCheckerProviderManager;
 import com.facebook.presto.storage.TempStorageManager;
 import com.facebook.presto.tracing.TracerProviderManager;
@@ -135,6 +137,7 @@ public class PluginManager
     private final QueryPreparerProviderManager queryPreparerProviderManager;
     private final NodeStatusNotificationManager nodeStatusNotificationManager;
     private final PlanCheckerProviderManager planCheckerProviderManager;
+    private final ExpressionOptimizerManager expressionOptimizerManager;
 
     @Inject
     public PluginManager(
@@ -157,7 +160,8 @@ public PluginManager(
             HistoryBasedPlanStatisticsManager historyBasedPlanStatisticsManager,
             TracerProviderManager tracerProviderManager,
             NodeStatusNotificationManager nodeStatusNotificationManager,
-            PlanCheckerProviderManager planCheckerProviderManager)
+            PlanCheckerProviderManager planCheckerProviderManager,
+            ExpressionOptimizerManager expressionOptimizerManager)
     {
         requireNonNull(nodeInfo, "nodeInfo is null");
         requireNonNull(config, "config is null");
@@ -190,6 +194,7 @@ public PluginManager(
         this.queryPreparerProviderManager = requireNonNull(queryPreparerProviderManager, "queryPreparerProviderManager is null");
         this.nodeStatusNotificationManager = requireNonNull(nodeStatusNotificationManager, "nodeStatusNotificationManager is null");
         this.planCheckerProviderManager = requireNonNull(planCheckerProviderManager, "planCheckerProviderManager is null");
+        this.expressionOptimizerManager = requireNonNull(expressionOptimizerManager, "expressionManager is null");
     }
 
     public void loadPlugins()
@@ -372,6 +377,11 @@ public void installCoordinatorPlugin(CoordinatorPlugin plugin)
             log.info("Registering system session property provider factory %s", providerFactory.getName());
             metadata.getSessionPropertyManager().addSessionPropertyProviderFactory(providerFactory);
         }
+
+        for (ExpressionOptimizerFactory batchRowExpressionInterpreterProvider : plugin.getRowExpressionInterpreterServiceFactories()) {
+            log.info("Registering batch row expression interpreter provider %s", batchRowExpressionInterpreterProvider.getName());
+            expressionOptimizerManager.addExpressionOptimizerFactory(batchRowExpressionInterpreterProvider);
+        }
     }
 
     private URLClassLoader buildClassLoader(String plugin)

presto-main/src/main/java/com/facebook/presto/server/PrestoServer.java

@@ -47,6 +47,7 @@
 import com.facebook.presto.server.security.PasswordAuthenticatorManager;
 import com.facebook.presto.server.security.ServerSecurityModule;
 import com.facebook.presto.sql.analyzer.FeaturesConfig;
+import com.facebook.presto.sql.expressions.ExpressionOptimizerManager;
 import com.facebook.presto.sql.parser.SqlParserOptions;
 import com.facebook.presto.sql.planner.sanity.PlanCheckerProviderManager;
 import com.facebook.presto.storage.TempStorageManager;
@@ -181,6 +182,7 @@ public void run()
             injector.getInstance(GracefulShutdownHandler.class).loadNodeStatusNotification();
             injector.getInstance(PlanCheckerProviderManager.class).loadPlanCheckerProviders();
             injector.getInstance(SessionPropertyManager.class).loadSessionPropertyProviders();
+            injector.getInstance(ExpressionOptimizerManager.class).loadExpressions();
             startAssociatedProcesses(injector);
 
             injector.getInstance(Announcer.class).start();

presto-main/src/main/java/com/facebook/presto/server/ServerMainModule.java

@@ -194,6 +194,7 @@
 import com.facebook.presto.sql.analyzer.MetadataExtractorMBean;
 import com.facebook.presto.sql.analyzer.QueryExplainer;
 import com.facebook.presto.sql.analyzer.QueryPreparerProviderManager;
+import com.facebook.presto.sql.expressions.ExpressionOptimizerManager;
 import com.facebook.presto.sql.gen.ExpressionCompiler;
 import com.facebook.presto.sql.gen.JoinCompiler;
 import com.facebook.presto.sql.gen.JoinFilterFunctionCompiler;
@@ -359,6 +360,9 @@ else if (serverConfig.isCoordinator()) {
         binder.bind(SystemSessionProperties.class).in(Scopes.SINGLETON);
         binder.bind(SessionPropertyDefaults.class).in(Scopes.SINGLETON);
 
+        // expression manager
+        binder.bind(ExpressionOptimizerManager.class).in(Scopes.SINGLETON);
+
         // schema properties
         binder.bind(SchemaPropertyManager.class).in(Scopes.SINGLETON);