[Testing] [Do not review] Test session properties

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

@@ -13,14 +13,19 @@
  */
 package com.facebook.presto.connector;
 
+import com.facebook.presto.metadata.InternalNode;
 import com.facebook.presto.metadata.InternalNodeManager;
 import com.facebook.presto.spi.ConnectorId;
 import com.facebook.presto.spi.Node;
 import com.facebook.presto.spi.NodeManager;
+import com.facebook.presto.spi.PrestoException;
 import com.google.common.collect.ImmutableSet;
 
 import java.util.Set;
 
+import static com.facebook.presto.spi.StandardErrorCode.NO_CPP_SIDECARS;
+import static com.facebook.presto.spi.StandardErrorCode.TOO_MANY_SIDECARS;
+import static com.google.common.collect.Iterables.getOnlyElement;
 import static java.util.Objects.requireNonNull;
 
 public class ConnectorAwareNodeManager
@@ -58,6 +63,19 @@ public Node getCurrentNode()
         return nodeManager.getCurrentNode();
     }
 
+    @Override
+    public Node getSidecarNode()
+    {
+        Set<InternalNode> coordinatorSidecars = nodeManager.getCoordinatorSidecars();
+        if (coordinatorSidecars.isEmpty()) {
+            throw new PrestoException(NO_CPP_SIDECARS, "Expected exactly one coordinator sidecar, but found none");
+        }
+        if (coordinatorSidecars.size() > 1) {
+            throw new PrestoException(TOO_MANY_SIDECARS, "Expected exactly one coordinator sidecar, but found " + coordinatorSidecars.size());
+        }
+        return getOnlyElement(coordinatorSidecars);
+    }
+
     @Override
     public String getEnvironment()
     {

presto-main/src/main/java/com/facebook/presto/metadata/SessionPropertyManager.java

@@ -29,6 +29,7 @@
 import com.facebook.presto.spi.ConnectorId;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.session.PropertyMetadata;
+import com.facebook.presto.spi.session.SystemSessionPropertyProvider;
 import com.facebook.presto.sql.planner.ParameterRewriter;
 import com.facebook.presto.sql.tree.Expression;
 import com.facebook.presto.sql.tree.ExpressionTreeRewriter;
@@ -53,6 +54,7 @@
 import static com.facebook.presto.sql.planner.ExpressionInterpreter.evaluateConstantExpression;
 import static com.google.common.base.MoreObjects.firstNonNull;
 import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.collect.ImmutableList.toImmutableList;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 
@@ -61,21 +63,36 @@ public final class SessionPropertyManager
     private static final JsonCodecFactory JSON_CODEC_FACTORY = new JsonCodecFactory();
     private final ConcurrentMap<String, PropertyMetadata<?>> systemSessionProperties = new ConcurrentHashMap<>();
     private final ConcurrentMap<ConnectorId, Map<String, PropertyMetadata<?>>> connectorSessionProperties = new ConcurrentHashMap<>();
+    private final ConcurrentMap<String, PropertyMetadata<?>> workerSessionProperties = new ConcurrentHashMap<>();
+    private final Map<String, SystemSessionPropertyProvider> systemSessionPropertyProviders;
 
     public SessionPropertyManager()
     {
         this(new SystemSessionProperties());
     }
 
-    @Inject
     public SessionPropertyManager(SystemSessionProperties systemSessionProperties)
     {
         this(systemSessionProperties.getSessionProperties());
     }
 
     public SessionPropertyManager(List<PropertyMetadata<?>> systemSessionProperties)
+    {
+        this(systemSessionProperties, new ConcurrentHashMap<>());
+    }
+
+    public SessionPropertyManager(List<PropertyMetadata<?>> systemSessionProperties, Map<String, SystemSessionPropertyProvider> systemSessionPropertyProviders)
     {
         addSystemSessionProperties(systemSessionProperties);
+        this.systemSessionPropertyProviders = new ConcurrentHashMap<>(systemSessionPropertyProviders);
+    }
+
+    @Inject
+    public SessionPropertyManager(
+            SystemSessionProperties systemSessionProperties,
+            Map<String, SystemSessionPropertyProvider> systemSessionPropertyProviders)
+    {
+        this(systemSessionProperties.getSessionProperties(), systemSessionPropertyProviders);
     }
 
     public void addSystemSessionProperties(List<PropertyMetadata<?>> systemSessionProperties)
@@ -108,7 +125,10 @@ public void removeConnectorSessionProperties(ConnectorId connectorId)
     public Optional<PropertyMetadata<?>> getSystemSessionPropertyMetadata(String name)
     {
         requireNonNull(name, "name is null");
-
+        if (systemSessionProperties.get(name) == null) {
+            updateWorkerSessionProperties();
+            return Optional.ofNullable(workerSessionProperties.get(name));
+        }
         return Optional.ofNullable(systemSessionProperties.get(name));
     }
 
@@ -124,12 +144,30 @@ public Optional<PropertyMetadata<?>> getConnectorSessionPropertyMetadata(Connect
         return Optional.ofNullable(properties.get(propertyName));
     }
 
+    private void updateWorkerSessionProperties()
+    {
+        List<PropertyMetadata<?>> workerSessionPropertiesList = getWorkerSessionProperties();
+        workerSessionPropertiesList.forEach(sessionProperty -> {
+            requireNonNull(sessionProperty, "sessionProperty is null");
+            // TODO: Implement fail fast in case of duplicate entries.
+            workerSessionProperties.put(sessionProperty.getName(), sessionProperty);
+        });
+    }
+
+    private List<PropertyMetadata<?>> getWorkerSessionProperties()
+    {
+        return systemSessionPropertyProviders.values().stream()
+                .flatMap(manager -> manager.getSessionProperties().stream())
+                .collect(toImmutableList());
+    }
+
     public List<SessionPropertyValue> getAllSessionProperties(Session session, Map<String, ConnectorId> catalogs)
     {
         requireNonNull(session, "session is null");
 
         ImmutableList.Builder<SessionPropertyValue> sessionPropertyValues = ImmutableList.builder();
         Map<String, String> systemProperties = session.getSystemProperties();
+        updateWorkerSessionProperties();
         for (PropertyMetadata<?> property : new TreeMap<>(systemSessionProperties).values()) {
             String defaultValue = firstNonNull(property.getDefaultValue(), "").toString();
             String value = systemProperties.getOrDefault(property.getName(), defaultValue);
@@ -165,6 +203,19 @@ public List<SessionPropertyValue> getAllSessionProperties(Session session, Map<S
             }
         }
 
+        for (PropertyMetadata<?> property : new TreeMap<>(workerSessionProperties).values()) {
+            String defaultValue = firstNonNull(property.getDefaultValue(), "").toString();
+            String value = systemProperties.getOrDefault(property.getName(), defaultValue);
+            sessionPropertyValues.add(new SessionPropertyValue(
+                    value,
+                    defaultValue,
+                    property.getName(),
+                    Optional.empty(),
+                    property.getName(),
+                    property.getDescription(),
+                    property.getSqlType().getDisplayName(),
+                    property.isHidden()));
+        }
         return sessionPropertyValues.build();
     }
 

presto-main/src/main/java/com/facebook/presto/nodeManager/PluginNodeManager.java

@@ -0,0 +1,97 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.facebook.presto.nodeManager;
+
+import com.facebook.presto.metadata.InternalNode;
+import com.facebook.presto.metadata.InternalNodeManager;
+import com.facebook.presto.spi.Node;
+import com.facebook.presto.spi.NodeManager;
+import com.facebook.presto.spi.PrestoException;
+import com.google.common.collect.ImmutableSet;
+import com.google.inject.Inject;
+
+import java.util.Set;
+
+import static com.facebook.presto.spi.StandardErrorCode.NO_CPP_SIDECARS;
+import static com.facebook.presto.spi.StandardErrorCode.TOO_MANY_SIDECARS;
+import static com.google.common.collect.ImmutableSet.toImmutableSet;
+import static com.google.common.collect.Iterables.getOnlyElement;
+import static java.util.Objects.requireNonNull;
+
+/**
+ * This class simplifies managing Presto's cluster nodes,
+ * focusing on active workers and coordinators without tying to specific connectors.
+ */
+public class PluginNodeManager
+        implements NodeManager
+{
+    private final InternalNodeManager nodeManager;
+    private final String environment;
+
+    @Inject
+    public PluginNodeManager(InternalNodeManager nodeManager)
+    {
+        this.nodeManager = nodeManager;
+        this.environment = "test";
+    }
+
+    public PluginNodeManager(InternalNodeManager nodeManager, String environment)
+    {
+        this.nodeManager = requireNonNull(nodeManager, "nodeManager is null");
+        this.environment = requireNonNull(environment, "environment is null");
+    }
+
+    @Override
+    public Set<Node> getAllNodes()
+    {
+        return ImmutableSet.<Node>builder()
+                .addAll(getWorkerNodes())
+                .addAll(nodeManager.getCoordinators())
+                .build();
+    }
+
+    @Override
+    public Set<Node> getWorkerNodes()
+    {
+        //Retrieves all active worker nodes, excluding coordinators, resource managers, and catalog servers.
+        return nodeManager.getAllNodes().getActiveNodes().stream()
+                .filter(node -> !node.isResourceManager() && !node.isCoordinator() && !node.isCatalogServer())
+                .collect(toImmutableSet());
+    }
+
+    @Override
+    public Node getCurrentNode()
+    {
+        return nodeManager.getCurrentNode();
+    }
+
+    @Override
+    public Node getSidecarNode()
+    {
+        Set<InternalNode> coordinatorSidecars = nodeManager.getCoordinatorSidecars();
+        if (coordinatorSidecars.isEmpty()) {
+            throw new PrestoException(NO_CPP_SIDECARS, "Expected exactly one coordinator sidecar, but found none");
+        }
+        if (coordinatorSidecars.size() > 1) {
+            throw new PrestoException(TOO_MANY_SIDECARS, "Expected exactly one coordinator sidecar, but found " + coordinatorSidecars.size());
+        }
+        return getOnlyElement(coordinatorSidecars);
+    }
+
+    @Override
+    public String getEnvironment()
+    {
+        return environment;
+    }
+}

presto-main/src/main/java/com/facebook/presto/operator/aggregation/GenericAccumulatorFactory.java

@@ -67,15 +67,15 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 
-import static com.facebook.presto.SystemSessionProperties.getDistinctAggregationLargeBlockSizeThreshold;
-import static com.facebook.presto.SystemSessionProperties.isDedupBasedDistinctAggregationSpillEnabled;
-import static com.facebook.presto.SystemSessionProperties.isDistinctAggregationLargeBlockSpillEnabled;
 import static com.facebook.presto.common.Page.wrapBlocksWithoutCopy;
 import static com.facebook.presto.common.block.ColumnarArray.toColumnarArray;
 import static com.facebook.presto.common.block.ColumnarRow.toColumnarRow;
 import static com.facebook.presto.common.type.BigintType.BIGINT;
 import static com.facebook.presto.common.type.BooleanType.BOOLEAN;
 import static com.facebook.presto.common.type.VarcharType.VARCHAR;
+import static com.facebook.presto.sessionpropertyproviders.JavaWorkerSystemSessionPropertyProvider.getDistinctAggregationLargeBlockSizeThreshold;
+import static com.facebook.presto.sessionpropertyproviders.JavaWorkerSystemSessionPropertyProvider.isDedupBasedDistinctAggregationSpillEnabled;
+import static com.facebook.presto.sessionpropertyproviders.JavaWorkerSystemSessionPropertyProvider.isDistinctAggregationLargeBlockSpillEnabled;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.collect.ImmutableList.toImmutableList;

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

@@ -139,6 +139,7 @@
 import com.facebook.presto.server.thrift.ThriftServerInfoService;
 import com.facebook.presto.server.thrift.ThriftTaskClient;
 import com.facebook.presto.server.thrift.ThriftTaskService;
+import com.facebook.presto.sessionpropertyproviders.JavaWorkerSystemSessionPropertyProvider;
 import com.facebook.presto.spi.ConnectorMetadataUpdateHandle;
 import com.facebook.presto.spi.ConnectorSplit;
 import com.facebook.presto.spi.ConnectorTypeSerde;
@@ -150,6 +151,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.SystemSessionPropertyProvider;
 import com.facebook.presto.spiller.FileSingleStreamSpillerFactory;
 import com.facebook.presto.spiller.GenericPartitioningSpillerFactory;
 import com.facebook.presto.spiller.GenericSpillerFactory;
@@ -182,6 +184,7 @@
 import com.facebook.presto.sql.analyzer.FeaturesConfig.SingleStreamSpillerChoice;
 import com.facebook.presto.sql.analyzer.ForMetadataExtractor;
 import com.facebook.presto.sql.analyzer.FunctionsConfig;
+import com.facebook.presto.sql.analyzer.JavaFeaturesConfig;
 import com.facebook.presto.sql.analyzer.MetadataExtractor;
 import com.facebook.presto.sql.analyzer.MetadataExtractorMBean;
 import com.facebook.presto.sql.analyzer.QueryExplainer;
@@ -223,6 +226,7 @@
 import com.google.inject.Provides;
 import com.google.inject.Scopes;
 import com.google.inject.TypeLiteral;
+import com.google.inject.multibindings.MapBinder;
 import io.airlift.slice.Slice;
 import io.airlift.units.DataSize;
 import io.airlift.units.Duration;
@@ -306,6 +310,7 @@ else if (serverConfig.isCoordinator()) {
 
         configBinder(binder).bindConfig(FeaturesConfig.class);
         configBinder(binder).bindConfig(FunctionsConfig.class);
+        configBinder(binder).bindConfig(JavaFeaturesConfig.class);
 
         binder.bind(PlanChecker.class).in(Scopes.SINGLETON);
 
@@ -785,6 +790,12 @@ public ListeningExecutorService createResourceManagerExecutor(ResourceManagerCon
         //Optional Status Detector
         newOptionalBinder(binder, NodeStatusService.class);
         binder.bind(NodeStatusNotificationManager.class).in(Scopes.SINGLETON);
+
+        //Session property providers
+        MapBinder<String, SystemSessionPropertyProvider> mapBinder = newMapBinder(binder, String.class, SystemSessionPropertyProvider.class);
+        if (!serverConfig.isCoordinatorSidecarEnabled() && !featuresConfig.isNativeExecutionEnabled()) {
+            mapBinder.addBinding("java-worker").to(JavaWorkerSystemSessionPropertyProvider.class);
+        }
     }
 
     @Provides