[native] Add native session property binding to presto spark module

Author: tanjialiang

presto-spark-base/src/main/java/com/facebook/presto/spark/PrestoSparkModule.java

@@ -110,6 +110,7 @@
 import com.facebook.presto.server.security.PasswordAuthenticatorManager;
 import com.facebook.presto.server.security.PrestoAuthenticatorManager;
 import com.facebook.presto.server.security.SecurityConfig;
+import com.facebook.presto.sessionpropertyproviders.NativeWorkerSessionPropertyProvider;
 import com.facebook.presto.spark.accesscontrol.PrestoSparkAccessControlChecker;
 import com.facebook.presto.spark.accesscontrol.PrestoSparkAuthenticatorProvider;
 import com.facebook.presto.spark.accesscontrol.PrestoSparkCredentialsProvider;
@@ -146,6 +147,7 @@
 import com.facebook.presto.spi.relation.DomainTranslator;
 import com.facebook.presto.spi.relation.PredicateCompiler;
 import com.facebook.presto.spi.relation.VariableReferenceExpression;
+import com.facebook.presto.spi.session.WorkerSessionPropertyProvider;
 import com.facebook.presto.spiller.GenericPartitioningSpillerFactory;
 import com.facebook.presto.spiller.GenericSpillerFactory;
 import com.facebook.presto.spiller.NodeSpillConfig;
@@ -216,6 +218,7 @@
 import com.google.inject.Provides;
 import com.google.inject.Scopes;
 import com.google.inject.TypeLiteral;
+import com.google.inject.multibindings.MapBinder;
 import org.weakref.jmx.MBeanExporter;
 import org.weakref.jmx.testing.TestingMBeanServer;
 
@@ -233,6 +236,7 @@
 import static com.facebook.airlift.json.JsonBinder.jsonBinder;
 import static com.facebook.airlift.json.JsonCodecBinder.jsonCodecBinder;
 import static com.facebook.airlift.json.smile.SmileCodecBinder.smileCodecBinder;
+import static com.google.inject.multibindings.MapBinder.newMapBinder;
 import static com.google.inject.multibindings.Multibinder.newSetBinder;
 import static com.google.inject.multibindings.OptionalBinder.newOptionalBinder;
 import static java.util.Objects.requireNonNull;
@@ -354,6 +358,14 @@ protected void setup(Binder binder)
         binder.bind(AnalyzePropertyManager.class).in(Scopes.SINGLETON);
         binder.bind(QuerySessionSupplier.class).in(Scopes.SINGLETON);
 
+        MapBinder<String, WorkerSessionPropertyProvider> mapBinder =
+                newMapBinder(binder, String.class, WorkerSessionPropertyProvider.class);
+        FeaturesConfig featuresConfig = buildConfigObject(FeaturesConfig.class);
+        if (featuresConfig.isNativeExecutionEnabled()) {
+            mapBinder.addBinding("native-worker").to(NativeWorkerSessionPropertyProvider.class)
+                    .in(Scopes.SINGLETON);
+        }
+
         // expression manager
         binder.bind(ExpressionOptimizerManager.class).in(Scopes.SINGLETON);
 

presto-spark-base/src/main/java/com/facebook/presto/spark/PrestoSparkSessionPropertyManagerProvider.java

@@ -15,13 +15,16 @@
 
 import com.facebook.presto.SystemSessionProperties;
 import com.facebook.presto.metadata.SessionPropertyManager;
+import com.facebook.presto.spi.session.WorkerSessionPropertyProvider;
 import com.facebook.presto.spiller.NodeSpillConfig;
 import com.facebook.presto.sql.analyzer.JavaFeaturesConfig;
 import com.google.common.collect.Streams;
 
 import javax.inject.Inject;
 import javax.inject.Provider;
 
+import java.util.Map;
+
 import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.util.Objects.requireNonNull;
@@ -31,14 +34,24 @@ public class PrestoSparkSessionPropertyManagerProvider
 {
     private final SystemSessionProperties systemSessionProperties;
     private final PrestoSparkSessionProperties prestoSparkSessionProperties;
+    private final Map<String, WorkerSessionPropertyProvider> workerSessionPropertyProviders;
     private final JavaFeaturesConfig javaFeaturesConfig;
     private final NodeSpillConfig nodeSpillConfig;
 
     @Inject
-    public PrestoSparkSessionPropertyManagerProvider(SystemSessionProperties systemSessionProperties, PrestoSparkSessionProperties prestoSparkSessionProperties, JavaFeaturesConfig javaFeaturesConfig, NodeSpillConfig nodeSpillConfig)
+    public PrestoSparkSessionPropertyManagerProvider(
+            SystemSessionProperties systemSessionProperties,
+            PrestoSparkSessionProperties prestoSparkSessionProperties,
+            Map<String, WorkerSessionPropertyProvider> workerSessionPropertyProviders,
+            JavaFeaturesConfig javaFeaturesConfig,
+            NodeSpillConfig nodeSpillConfig)
     {
-        this.systemSessionProperties = requireNonNull(systemSessionProperties, "systemSessionProperties is null");
-        this.prestoSparkSessionProperties = requireNonNull(prestoSparkSessionProperties, "prestoSparkSessionProperties is null");
+        this.systemSessionProperties = requireNonNull(systemSessionProperties,
+            "systemSessionProperties is null");
+        this.prestoSparkSessionProperties = requireNonNull(prestoSparkSessionProperties,
+            "prestoSparkSessionProperties is null");
+        this.workerSessionPropertyProviders = requireNonNull(workerSessionPropertyProviders,
+            "workerSessionPropertyProviders is null");
         this.javaFeaturesConfig = requireNonNull(javaFeaturesConfig, "javaFeaturesConfig is null");
         this.nodeSpillConfig = requireNonNull(nodeSpillConfig, "nodeSpillConfig is null");
     }
@@ -47,11 +60,13 @@ public PrestoSparkSessionPropertyManagerProvider(SystemSessionProperties systemS
     public SessionPropertyManager get()
     {
         return createTestingSessionPropertyManager(
-                Streams.concat(
-                        systemSessionProperties.getSessionProperties().stream(),
-                        prestoSparkSessionProperties.getSessionProperties().stream()
-                ).collect(toImmutableList()),
-                javaFeaturesConfig,
-                nodeSpillConfig);
+            Streams.concat(
+                systemSessionProperties.getSessionProperties().stream(),
+                prestoSparkSessionProperties.getSessionProperties().stream(),
+                workerSessionPropertyProviders.values().stream()
+                    .flatMap(provider -> provider.getSessionProperties().stream())
+            ).collect(toImmutableList()),
+            javaFeaturesConfig,
+            nodeSpillConfig);
     }
 }

presto-spark-base/src/test/java/com/facebook/presto/spark/planner/optimizers/TestPickJoinSides.java

@@ -53,6 +53,7 @@
 
 import java.util.Arrays;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
 
 import static com.facebook.presto.SystemSessionProperties.JOIN_MAX_BROADCAST_TABLE_SIZE;
 import static com.facebook.presto.SystemSessionProperties.TASK_CONCURRENCY;
@@ -89,7 +90,7 @@ public void setUp()
         tester = new RuleTester(
                 ImmutableList.of(),
                 ImmutableMap.of(),
-                new PrestoSparkSessionPropertyManagerProvider(new SystemSessionProperties(), new PrestoSparkSessionProperties(), new JavaFeaturesConfig(), new NodeSpillConfig()).get(),
+                new PrestoSparkSessionPropertyManagerProvider(new SystemSessionProperties(), new PrestoSparkSessionProperties(), new ConcurrentHashMap<>(), new JavaFeaturesConfig(), new NodeSpillConfig()).get(),
                 Optional.of(NODES_COUNT),
                 new TpchConnectorFactory(1));
     }