Separate out worker session properties and introduce JavaFeaturesConfig

Co-authored-by: Joe Abraham <[email protected]>

presto-druid/src/test/java/com/facebook/presto/druid/TestDruidQueryBase.java

@@ -20,7 +20,6 @@
 import com.facebook.presto.metadata.FunctionAndTypeManager;
 import com.facebook.presto.metadata.Metadata;
 import com.facebook.presto.metadata.MetadataManager;
-import com.facebook.presto.metadata.SessionPropertyManager;
 import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ConnectorId;
 import com.facebook.presto.spi.ConnectorSession;
@@ -62,6 +61,7 @@
 import static com.facebook.presto.druid.DruidQueryGeneratorContext.Origin.DERIVED;
 import static com.facebook.presto.druid.DruidQueryGeneratorContext.Origin.TABLE_COLUMN;
 import static com.facebook.presto.metadata.FunctionAndTypeManager.createTestFunctionAndTypeManager;
+import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
 import static com.facebook.presto.spi.plan.LimitNode.Step.FINAL;
 import static com.facebook.presto.sql.analyzer.ExpressionAnalyzer.getExpressionTypes;
 import static com.facebook.presto.testing.TestingConnectorSession.SESSION;
@@ -109,7 +109,7 @@ protected static class SessionHolder
         public SessionHolder()
         {
             connectorSession = SESSION;
-            session = TestingSession.testSessionBuilder(new SessionPropertyManager(new SystemSessionProperties().getSessionProperties())).build();
+            session = TestingSession.testSessionBuilder(createTestingSessionPropertyManager(new SystemSessionProperties().getSessionProperties())).build();
         }
 
         public ConnectorSession getConnectorSession()

presto-iceberg/src/test/java/com/facebook/presto/iceberg/TestIcebergFileWriter.java

@@ -63,6 +63,7 @@
 import static com.facebook.presto.iceberg.IcebergAbstractMetadata.toIcebergSchema;
 import static com.facebook.presto.iceberg.IcebergQueryRunner.ICEBERG_CATALOG;
 import static com.facebook.presto.iceberg.IcebergSessionProperties.dataSizeSessionProperty;
+import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
 import static com.facebook.presto.testing.TestingSession.testSessionBuilder;
 import static com.google.common.io.Files.createTempDir;
 import static org.apache.iceberg.parquet.ParquetSchemaUtil.convert;
@@ -78,7 +79,7 @@ public class TestIcebergFileWriter
     public void setup() throws Exception
     {
         ConnectorId connectorId = new ConnectorId("iceberg");
-        SessionPropertyManager sessionPropertyManager = new SessionPropertyManager();
+        SessionPropertyManager sessionPropertyManager = createTestingSessionPropertyManager();
 
         sessionPropertyManager.addConnectorSessionProperties(
                 connectorId,

presto-main/src/main/java/com/facebook/presto/connector/system/SystemConnectorSessionUtil.java

@@ -16,13 +16,14 @@
 import com.facebook.presto.Session;
 import com.facebook.presto.SystemSessionProperties;
 import com.facebook.presto.common.transaction.TransactionId;
-import com.facebook.presto.metadata.SessionPropertyManager;
 import com.facebook.presto.spi.ConnectorSession;
 import com.facebook.presto.spi.QueryId;
 import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
 import com.facebook.presto.spi.security.ConnectorIdentity;
 import com.facebook.presto.spi.security.Identity;
 
+import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
+
 public final class SystemConnectorSessionUtil
 {
     private static final SystemSessionProperties SYSTEM_SESSION_PROPERTIES = new SystemSessionProperties();
@@ -35,7 +36,7 @@ public static Session toSession(ConnectorTransactionHandle transactionHandle, Co
         TransactionId transactionId = ((GlobalSystemTransactionHandle) transactionHandle).getTransactionId();
         ConnectorIdentity connectorIdentity = session.getIdentity();
         Identity identity = new Identity(connectorIdentity.getUser(), connectorIdentity.getPrincipal(), connectorIdentity.getExtraCredentials());
-        return Session.builder(new SessionPropertyManager(SYSTEM_SESSION_PROPERTIES))
+        return Session.builder(createTestingSessionPropertyManager(SYSTEM_SESSION_PROPERTIES))
                 .setQueryId(new QueryId(session.getQueryId()))
                 .setTransactionId(transactionId)
                 .setCatalog("catalog")

presto-main/src/main/java/com/facebook/presto/dispatcher/DispatchManager.java

@@ -24,7 +24,6 @@
 import com.facebook.presto.execution.QueryTracker;
 import com.facebook.presto.execution.resourceGroups.ResourceGroupManager;
 import com.facebook.presto.execution.warnings.WarningCollectorFactory;
-import com.facebook.presto.metadata.SessionPropertyManager;
 import com.facebook.presto.resourcemanager.ClusterQueryTrackerService;
 import com.facebook.presto.resourcemanager.ClusterStatusSender;
 import com.facebook.presto.server.BasicQueryInfo;
@@ -56,6 +55,7 @@
 
 import static com.facebook.presto.Session.SessionBuilder;
 import static com.facebook.presto.SystemSessionProperties.getAnalyzerType;
+import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
 import static com.facebook.presto.spi.StandardErrorCode.QUERY_TEXT_TOO_LARGE;
 import static com.facebook.presto.util.AnalyzerUtil.createAnalyzerOptions;
 import static com.google.common.base.Preconditions.checkArgument;
@@ -330,7 +330,7 @@ private <C> void createQueryInternal(QueryId queryId, String slug, int retryCoun
         catch (Throwable throwable) {
             // creation must never fail, so register a failed query in this case
             if (session == null) {
-                session = Session.builder(new SessionPropertyManager())
+                session = Session.builder(createTestingSessionPropertyManager())
                         .setQueryId(queryId)
                         .setIdentity(sessionContext.getIdentity())
                         .setSource(sessionContext.getSource())

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

@@ -119,6 +119,7 @@
 import static com.facebook.presto.metadata.MetadataUtil.getOptionalCatalogMetadata;
 import static com.facebook.presto.metadata.MetadataUtil.getOptionalTableHandle;
 import static com.facebook.presto.metadata.MetadataUtil.toSchemaTableName;
+import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
 import static com.facebook.presto.metadata.TableLayout.fromConnectorLayout;
 import static com.facebook.presto.spi.Constraint.alwaysTrue;
 import static com.facebook.presto.spi.StandardErrorCode.INVALID_VIEW;
@@ -240,7 +241,7 @@ public static MetadataManager createTestMetadataManager(TransactionManager trans
         return new MetadataManager(
                 new FunctionAndTypeManager(transactionManager, blockEncodingManager, featuresConfig, functionsConfig, new HandleResolver(), ImmutableSet.of()),
                 blockEncodingManager,
-                new SessionPropertyManager(),
+                createTestingSessionPropertyManager(),
                 new SchemaPropertyManager(),
                 new TablePropertyManager(),
                 new ColumnPropertyManager(),

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

@@ -25,16 +25,27 @@
 import com.facebook.presto.common.type.IntegerType;
 import com.facebook.presto.common.type.MapType;
 import com.facebook.presto.common.type.Type;
+import com.facebook.presto.common.type.TypeManager;
 import com.facebook.presto.common.type.VarcharType;
+import com.facebook.presto.sessionpropertyproviders.JavaWorkerSessionPropertyProvider;
 import com.facebook.presto.spi.ConnectorId;
+import com.facebook.presto.spi.NodeManager;
 import com.facebook.presto.spi.PrestoException;
 import com.facebook.presto.spi.session.PropertyMetadata;
+import com.facebook.presto.spi.session.SessionPropertyContext;
+import com.facebook.presto.spi.session.WorkerSessionPropertyProvider;
+import com.facebook.presto.spi.session.WorkerSessionPropertyProviderFactory;
+import com.facebook.presto.spiller.NodeSpillConfig;
+import com.facebook.presto.sql.analyzer.FeaturesConfig;
+import com.facebook.presto.sql.analyzer.JavaFeaturesConfig;
 import com.facebook.presto.sql.planner.ParameterRewriter;
 import com.facebook.presto.sql.tree.Expression;
 import com.facebook.presto.sql.tree.ExpressionTreeRewriter;
 import com.facebook.presto.sql.tree.NodeRef;
 import com.facebook.presto.sql.tree.Parameter;
+import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 
 import javax.annotation.Nullable;
@@ -47,35 +58,100 @@
 import java.util.TreeMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.function.Supplier;
 
 import static com.facebook.presto.common.type.TypeUtils.writeNativeValue;
 import static com.facebook.presto.spi.StandardErrorCode.INVALID_SESSION_PROPERTY;
 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;
+import static java.util.concurrent.TimeUnit.HOURS;
 
 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 Map<String, WorkerSessionPropertyProvider> workerSessionPropertyProviders;
+    private final Map<String, WorkerSessionPropertyProviderFactory> workerSessionPropertyProviderFactories = new ConcurrentHashMap<>();
+    private final Supplier<Map<String, PropertyMetadata<?>>> memoizedWorkerSessionProperties;
+    private final Optional<NodeManager> nodeManager;
+    private final Optional<TypeManager> functionAndTypeManager;
 
-    public SessionPropertyManager()
+    @Inject
+    public SessionPropertyManager(
+            SystemSessionProperties systemSessionProperties,
+            Map<String, WorkerSessionPropertyProvider> workerSessionPropertyProviders,
+            FunctionAndTypeManager functionAndTypeManager,
+            NodeManager nodeManager)
     {
-        this(new SystemSessionProperties());
+        this(systemSessionProperties.getSessionProperties(), workerSessionPropertyProviders, Optional.ofNullable(functionAndTypeManager), Optional.ofNullable(nodeManager));
     }
 
-    @Inject
-    public SessionPropertyManager(SystemSessionProperties systemSessionProperties)
+    public SessionPropertyManager(
+            List<PropertyMetadata<?>> sessionProperties,
+            Map<String, WorkerSessionPropertyProvider> workerSessionPropertyProviders,
+            Optional<TypeManager> functionAndTypeManager,
+            Optional<NodeManager> nodeManager)
+    {
+        this.nodeManager = requireNonNull(nodeManager, "nodeManager is null");
+        this.functionAndTypeManager = requireNonNull(functionAndTypeManager, "functionAndTypeManager is null");
+        this.memoizedWorkerSessionProperties = Suppliers.memoizeWithExpiration(this::getWorkerSessionProperties,
+                1, HOURS);
+        this.workerSessionPropertyProviders = new ConcurrentHashMap<>(workerSessionPropertyProviders);
+        addSystemSessionProperties(sessionProperties);
+    }
+
+    public static SessionPropertyManager createTestingSessionPropertyManager()
+    {
+        return createTestingSessionPropertyManager(new SystemSessionProperties().getSessionProperties(), new JavaFeaturesConfig(), new NodeSpillConfig());
+    }
+
+    public static SessionPropertyManager createTestingSessionPropertyManager(SystemSessionProperties systemSessionProperties)
+    {
+        return createTestingSessionPropertyManager(systemSessionProperties.getSessionProperties(), new JavaFeaturesConfig(), new NodeSpillConfig());
+    }
+
+    public static SessionPropertyManager createTestingSessionPropertyManager(List<PropertyMetadata<?>> sessionProperties)
+    {
+        return createTestingSessionPropertyManager(sessionProperties, new JavaFeaturesConfig(), new NodeSpillConfig());
+    }
+
+    public static SessionPropertyManager createTestingSessionPropertyManager(
+            List<PropertyMetadata<?>> sessionProperties,
+            JavaFeaturesConfig javaFeaturesConfig,
+            NodeSpillConfig nodeSpillConfig)
     {
-        this(systemSessionProperties.getSessionProperties());
+        return new SessionPropertyManager(
+                sessionProperties,
+                ImmutableMap.of(
+                        "java-worker",
+                        new JavaWorkerSessionPropertyProvider(
+                                new FeaturesConfig(),
+                                javaFeaturesConfig,
+                                nodeSpillConfig)),
+                Optional.empty(),
+                Optional.empty());
+    }
+
+    public void loadSessionPropertyProvider(String sessionPropertyProviderName)
+    {
+        WorkerSessionPropertyProviderFactory factory = workerSessionPropertyProviderFactories.get(sessionPropertyProviderName);
+        checkState(factory != null, "No factory for session property provider : " + sessionPropertyProviderName);
+        WorkerSessionPropertyProvider sessionPropertyProvider = factory.create(new SessionPropertyContext(functionAndTypeManager, nodeManager));
+        if (workerSessionPropertyProviders.putIfAbsent(sessionPropertyProviderName, sessionPropertyProvider) != null) {
+            throw new IllegalArgumentException("System session property provider is already registered for property provider : " + sessionPropertyProviderName);
+        }
     }
 
-    public SessionPropertyManager(List<PropertyMetadata<?>> systemSessionProperties)
+    public void loadSessionPropertyProviders()
     {
-        addSystemSessionProperties(systemSessionProperties);
+        for (String sessionPropertyProviderName : workerSessionPropertyProviderFactories.keySet()) {
+            loadSessionPropertyProvider(sessionPropertyProviderName);
+        }
     }
 
     public void addSystemSessionProperties(List<PropertyMetadata<?>> systemSessionProperties)
@@ -108,7 +184,9 @@ public void removeConnectorSessionProperties(ConnectorId connectorId)
     public Optional<PropertyMetadata<?>> getSystemSessionPropertyMetadata(String name)
     {
         requireNonNull(name, "name is null");
-
+        if (systemSessionProperties.get(name) == null) {
+            return Optional.ofNullable(memoizedWorkerSessionProperties.get().get(name));
+        }
         return Optional.ofNullable(systemSessionProperties.get(name));
     }
 
@@ -124,6 +202,20 @@ public Optional<PropertyMetadata<?>> getConnectorSessionPropertyMetadata(Connect
         return Optional.ofNullable(properties.get(propertyName));
     }
 
+    private Map<String, PropertyMetadata<?>> getWorkerSessionProperties()
+    {
+        List<PropertyMetadata<?>> workerSessionPropertiesList = workerSessionPropertyProviders.values().stream()
+                .flatMap(manager -> manager.getSessionProperties().stream())
+                .collect(toImmutableList());
+        Map<String, PropertyMetadata<?>> workerSessionProperties = new ConcurrentHashMap<>();
+        workerSessionPropertiesList.forEach(sessionProperty -> {
+            requireNonNull(sessionProperty, "sessionProperty is null");
+            // TODO: Implement fail fast in case of duplicate entries.
+            workerSessionProperties.put(sessionProperty.getName(), sessionProperty);
+        });
+        return workerSessionProperties;
+    }
+
     public List<SessionPropertyValue> getAllSessionProperties(Session session, Map<String, ConnectorId> catalogs)
     {
         requireNonNull(session, "session is null");
@@ -165,6 +257,19 @@ public List<SessionPropertyValue> getAllSessionProperties(Session session, Map<S
             }
         }
 
+        for (PropertyMetadata<?> property : new TreeMap<>(memoizedWorkerSessionProperties.get()).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/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.JavaWorkerSessionPropertyProvider.getDistinctAggregationLargeBlockSizeThreshold;
+import static com.facebook.presto.sessionpropertyproviders.JavaWorkerSessionPropertyProvider.isDedupBasedDistinctAggregationSpillEnabled;
+import static com.facebook.presto.sessionpropertyproviders.JavaWorkerSessionPropertyProvider.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/PrestoServer.java

@@ -39,6 +39,7 @@
 import com.facebook.presto.execution.warnings.WarningCollectorModule;
 import com.facebook.presto.metadata.Catalog;
 import com.facebook.presto.metadata.CatalogManager;
+import com.facebook.presto.metadata.SessionPropertyManager;
 import com.facebook.presto.metadata.StaticCatalogStore;
 import com.facebook.presto.metadata.StaticFunctionNamespaceStore;
 import com.facebook.presto.security.AccessControlManager;
@@ -179,6 +180,7 @@ public void run()
             injector.getInstance(NodeStatusNotificationManager.class).loadNodeStatusNotificationProvider();
             injector.getInstance(GracefulShutdownHandler.class).loadNodeStatusNotification();
             injector.getInstance(PlanCheckerProviderManager.class).loadPlanCheckerProviders();
+            injector.getInstance(SessionPropertyManager.class).loadSessionPropertyProviders();
             startAssociatedProcesses(injector);
 
             injector.getInstance(Announcer.class).start();