Skip to content

Commit cee51f5

Browse files
pdabre12Joe-Abraham
authored and
tdcmeehan
committed
Separate out worker session properties and introduce JavaFeaturesConfig
Co-authored-by: Joe Abraham <joe.abraham@ibm.com>
1 parent b936b6d commit cee51f5

File tree

52 files changed

+1148
-627
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

52 files changed

+1148
-627
lines changed

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020
import com.facebook.presto.metadata.FunctionAndTypeManager;
2121
import com.facebook.presto.metadata.Metadata;
2222
import com.facebook.presto.metadata.MetadataManager;
23-
import com.facebook.presto.metadata.SessionPropertyManager;
2423
import com.facebook.presto.spi.ColumnHandle;
2524
import com.facebook.presto.spi.ConnectorId;
2625
import com.facebook.presto.spi.ConnectorSession;
@@ -62,6 +61,7 @@
6261
import static com.facebook.presto.druid.DruidQueryGeneratorContext.Origin.DERIVED;
6362
import static com.facebook.presto.druid.DruidQueryGeneratorContext.Origin.TABLE_COLUMN;
6463
import static com.facebook.presto.metadata.FunctionAndTypeManager.createTestFunctionAndTypeManager;
64+
import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
6565
import static com.facebook.presto.spi.plan.LimitNode.Step.FINAL;
6666
import static com.facebook.presto.sql.analyzer.ExpressionAnalyzer.getExpressionTypes;
6767
import static com.facebook.presto.testing.TestingConnectorSession.SESSION;
@@ -109,7 +109,7 @@ protected static class SessionHolder
109109
public SessionHolder()
110110
{
111111
connectorSession = SESSION;
112-
session = TestingSession.testSessionBuilder(new SessionPropertyManager(new SystemSessionProperties().getSessionProperties())).build();
112+
session = TestingSession.testSessionBuilder(createTestingSessionPropertyManager(new SystemSessionProperties().getSessionProperties())).build();
113113
}
114114

115115
public ConnectorSession getConnectorSession()

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,7 @@
6363
import static com.facebook.presto.iceberg.IcebergAbstractMetadata.toIcebergSchema;
6464
import static com.facebook.presto.iceberg.IcebergQueryRunner.ICEBERG_CATALOG;
6565
import static com.facebook.presto.iceberg.IcebergSessionProperties.dataSizeSessionProperty;
66+
import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
6667
import static com.facebook.presto.testing.TestingSession.testSessionBuilder;
6768
import static com.google.common.io.Files.createTempDir;
6869
import static org.apache.iceberg.parquet.ParquetSchemaUtil.convert;
@@ -78,7 +79,7 @@ public class TestIcebergFileWriter
7879
public void setup() throws Exception
7980
{
8081
ConnectorId connectorId = new ConnectorId("iceberg");
81-
SessionPropertyManager sessionPropertyManager = new SessionPropertyManager();
82+
SessionPropertyManager sessionPropertyManager = createTestingSessionPropertyManager();
8283

8384
sessionPropertyManager.addConnectorSessionProperties(
8485
connectorId,

‎presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

Lines changed: 14 additions & 347 deletions
Large diffs are not rendered by default.

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

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,13 +16,14 @@
1616
import com.facebook.presto.Session;
1717
import com.facebook.presto.SystemSessionProperties;
1818
import com.facebook.presto.common.transaction.TransactionId;
19-
import com.facebook.presto.metadata.SessionPropertyManager;
2019
import com.facebook.presto.spi.ConnectorSession;
2120
import com.facebook.presto.spi.QueryId;
2221
import com.facebook.presto.spi.connector.ConnectorTransactionHandle;
2322
import com.facebook.presto.spi.security.ConnectorIdentity;
2423
import com.facebook.presto.spi.security.Identity;
2524

25+
import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
26+
2627
public final class SystemConnectorSessionUtil
2728
{
2829
private static final SystemSessionProperties SYSTEM_SESSION_PROPERTIES = new SystemSessionProperties();
@@ -35,7 +36,7 @@ public static Session toSession(ConnectorTransactionHandle transactionHandle, Co
3536
TransactionId transactionId = ((GlobalSystemTransactionHandle) transactionHandle).getTransactionId();
3637
ConnectorIdentity connectorIdentity = session.getIdentity();
3738
Identity identity = new Identity(connectorIdentity.getUser(), connectorIdentity.getPrincipal(), connectorIdentity.getExtraCredentials());
38-
return Session.builder(new SessionPropertyManager(SYSTEM_SESSION_PROPERTIES))
39+
return Session.builder(createTestingSessionPropertyManager(SYSTEM_SESSION_PROPERTIES))
3940
.setQueryId(new QueryId(session.getQueryId()))
4041
.setTransactionId(transactionId)
4142
.setCatalog("catalog")

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

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@
2424
import com.facebook.presto.execution.QueryTracker;
2525
import com.facebook.presto.execution.resourceGroups.ResourceGroupManager;
2626
import com.facebook.presto.execution.warnings.WarningCollectorFactory;
27-
import com.facebook.presto.metadata.SessionPropertyManager;
2827
import com.facebook.presto.resourcemanager.ClusterQueryTrackerService;
2928
import com.facebook.presto.resourcemanager.ClusterStatusSender;
3029
import com.facebook.presto.server.BasicQueryInfo;
@@ -56,6 +55,7 @@
5655

5756
import static com.facebook.presto.Session.SessionBuilder;
5857
import static com.facebook.presto.SystemSessionProperties.getAnalyzerType;
58+
import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
5959
import static com.facebook.presto.spi.StandardErrorCode.QUERY_TEXT_TOO_LARGE;
6060
import static com.facebook.presto.util.AnalyzerUtil.createAnalyzerOptions;
6161
import static com.google.common.base.Preconditions.checkArgument;
@@ -330,7 +330,7 @@ private <C> void createQueryInternal(QueryId queryId, String slug, int retryCoun
330330
catch (Throwable throwable) {
331331
// creation must never fail, so register a failed query in this case
332332
if (session == null) {
333-
session = Session.builder(new SessionPropertyManager())
333+
session = Session.builder(createTestingSessionPropertyManager())
334334
.setQueryId(queryId)
335335
.setIdentity(sessionContext.getIdentity())
336336
.setSource(sessionContext.getSource())

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,7 @@
119119
import static com.facebook.presto.metadata.MetadataUtil.getOptionalCatalogMetadata;
120120
import static com.facebook.presto.metadata.MetadataUtil.getOptionalTableHandle;
121121
import static com.facebook.presto.metadata.MetadataUtil.toSchemaTableName;
122+
import static com.facebook.presto.metadata.SessionPropertyManager.createTestingSessionPropertyManager;
122123
import static com.facebook.presto.metadata.TableLayout.fromConnectorLayout;
123124
import static com.facebook.presto.spi.Constraint.alwaysTrue;
124125
import static com.facebook.presto.spi.StandardErrorCode.INVALID_VIEW;
@@ -240,7 +241,7 @@ public static MetadataManager createTestMetadataManager(TransactionManager trans
240241
return new MetadataManager(
241242
new FunctionAndTypeManager(transactionManager, blockEncodingManager, featuresConfig, functionsConfig, new HandleResolver(), ImmutableSet.of()),
242243
blockEncodingManager,
243-
new SessionPropertyManager(),
244+
createTestingSessionPropertyManager(),
244245
new SchemaPropertyManager(),
245246
new TablePropertyManager(),
246247
new ColumnPropertyManager(),

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

Lines changed: 113 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -25,16 +25,27 @@
2525
import com.facebook.presto.common.type.IntegerType;
2626
import com.facebook.presto.common.type.MapType;
2727
import com.facebook.presto.common.type.Type;
28+
import com.facebook.presto.common.type.TypeManager;
2829
import com.facebook.presto.common.type.VarcharType;
30+
import com.facebook.presto.sessionpropertyproviders.JavaWorkerSessionPropertyProvider;
2931
import com.facebook.presto.spi.ConnectorId;
32+
import com.facebook.presto.spi.NodeManager;
3033
import com.facebook.presto.spi.PrestoException;
3134
import com.facebook.presto.spi.session.PropertyMetadata;
35+
import com.facebook.presto.spi.session.SessionPropertyContext;
36+
import com.facebook.presto.spi.session.WorkerSessionPropertyProvider;
37+
import com.facebook.presto.spi.session.WorkerSessionPropertyProviderFactory;
38+
import com.facebook.presto.spiller.NodeSpillConfig;
39+
import com.facebook.presto.sql.analyzer.FeaturesConfig;
40+
import com.facebook.presto.sql.analyzer.JavaFeaturesConfig;
3241
import com.facebook.presto.sql.planner.ParameterRewriter;
3342
import com.facebook.presto.sql.tree.Expression;
3443
import com.facebook.presto.sql.tree.ExpressionTreeRewriter;
3544
import com.facebook.presto.sql.tree.NodeRef;
3645
import com.facebook.presto.sql.tree.Parameter;
46+
import com.google.common.base.Suppliers;
3747
import com.google.common.collect.ImmutableList;
48+
import com.google.common.collect.ImmutableMap;
3849
import com.google.common.collect.Maps;
3950

4051
import javax.annotation.Nullable;
@@ -47,35 +58,100 @@
4758
import java.util.TreeMap;
4859
import java.util.concurrent.ConcurrentHashMap;
4960
import java.util.concurrent.ConcurrentMap;
61+
import java.util.function.Supplier;
5062

5163
import static com.facebook.presto.common.type.TypeUtils.writeNativeValue;
5264
import static com.facebook.presto.spi.StandardErrorCode.INVALID_SESSION_PROPERTY;
5365
import static com.facebook.presto.sql.planner.ExpressionInterpreter.evaluateConstantExpression;
5466
import static com.google.common.base.MoreObjects.firstNonNull;
5567
import static com.google.common.base.Preconditions.checkState;
68+
import static com.google.common.collect.ImmutableList.toImmutableList;
5669
import static java.lang.String.format;
5770
import static java.util.Objects.requireNonNull;
71+
import static java.util.concurrent.TimeUnit.HOURS;
5872

5973
public final class SessionPropertyManager
6074
{
6175
private static final JsonCodecFactory JSON_CODEC_FACTORY = new JsonCodecFactory();
6276
private final ConcurrentMap<String, PropertyMetadata<?>> systemSessionProperties = new ConcurrentHashMap<>();
6377
private final ConcurrentMap<ConnectorId, Map<String, PropertyMetadata<?>>> connectorSessionProperties = new ConcurrentHashMap<>();
78+
private final Map<String, WorkerSessionPropertyProvider> workerSessionPropertyProviders;
79+
private final Map<String, WorkerSessionPropertyProviderFactory> workerSessionPropertyProviderFactories = new ConcurrentHashMap<>();
80+
private final Supplier<Map<String, PropertyMetadata<?>>> memoizedWorkerSessionProperties;
81+
private final Optional<NodeManager> nodeManager;
82+
private final Optional<TypeManager> functionAndTypeManager;
6483

65-
public SessionPropertyManager()
84+
@Inject
85+
public SessionPropertyManager(
86+
SystemSessionProperties systemSessionProperties,
87+
Map<String, WorkerSessionPropertyProvider> workerSessionPropertyProviders,
88+
FunctionAndTypeManager functionAndTypeManager,
89+
NodeManager nodeManager)
6690
{
67-
this(new SystemSessionProperties());
91+
this(systemSessionProperties.getSessionProperties(), workerSessionPropertyProviders, Optional.ofNullable(functionAndTypeManager), Optional.ofNullable(nodeManager));
6892
}
6993

70-
@Inject
71-
public SessionPropertyManager(SystemSessionProperties systemSessionProperties)
94+
public SessionPropertyManager(
95+
List<PropertyMetadata<?>> sessionProperties,
96+
Map<String, WorkerSessionPropertyProvider> workerSessionPropertyProviders,
97+
Optional<TypeManager> functionAndTypeManager,
98+
Optional<NodeManager> nodeManager)
99+
{
100+
this.nodeManager = requireNonNull(nodeManager, "nodeManager is null");
101+
this.functionAndTypeManager = requireNonNull(functionAndTypeManager, "functionAndTypeManager is null");
102+
this.memoizedWorkerSessionProperties = Suppliers.memoizeWithExpiration(this::getWorkerSessionProperties,
103+
1, HOURS);
104+
this.workerSessionPropertyProviders = new ConcurrentHashMap<>(workerSessionPropertyProviders);
105+
addSystemSessionProperties(sessionProperties);
106+
}
107+
108+
public static SessionPropertyManager createTestingSessionPropertyManager()
109+
{
110+
return createTestingSessionPropertyManager(new SystemSessionProperties().getSessionProperties(), new JavaFeaturesConfig(), new NodeSpillConfig());
111+
}
112+
113+
public static SessionPropertyManager createTestingSessionPropertyManager(SystemSessionProperties systemSessionProperties)
114+
{
115+
return createTestingSessionPropertyManager(systemSessionProperties.getSessionProperties(), new JavaFeaturesConfig(), new NodeSpillConfig());
116+
}
117+
118+
public static SessionPropertyManager createTestingSessionPropertyManager(List<PropertyMetadata<?>> sessionProperties)
119+
{
120+
return createTestingSessionPropertyManager(sessionProperties, new JavaFeaturesConfig(), new NodeSpillConfig());
121+
}
122+
123+
public static SessionPropertyManager createTestingSessionPropertyManager(
124+
List<PropertyMetadata<?>> sessionProperties,
125+
JavaFeaturesConfig javaFeaturesConfig,
126+
NodeSpillConfig nodeSpillConfig)
72127
{
73-
this(systemSessionProperties.getSessionProperties());
128+
return new SessionPropertyManager(
129+
sessionProperties,
130+
ImmutableMap.of(
131+
"java-worker",
132+
new JavaWorkerSessionPropertyProvider(
133+
new FeaturesConfig(),
134+
javaFeaturesConfig,
135+
nodeSpillConfig)),
136+
Optional.empty(),
137+
Optional.empty());
138+
}
139+
140+
public void loadSessionPropertyProvider(String sessionPropertyProviderName)
141+
{
142+
WorkerSessionPropertyProviderFactory factory = workerSessionPropertyProviderFactories.get(sessionPropertyProviderName);
143+
checkState(factory != null, "No factory for session property provider : " + sessionPropertyProviderName);
144+
WorkerSessionPropertyProvider sessionPropertyProvider = factory.create(new SessionPropertyContext(functionAndTypeManager, nodeManager));
145+
if (workerSessionPropertyProviders.putIfAbsent(sessionPropertyProviderName, sessionPropertyProvider) != null) {
146+
throw new IllegalArgumentException("System session property provider is already registered for property provider : " + sessionPropertyProviderName);
147+
}
74148
}
75149

76-
public SessionPropertyManager(List<PropertyMetadata<?>> systemSessionProperties)
150+
public void loadSessionPropertyProviders()
77151
{
78-
addSystemSessionProperties(systemSessionProperties);
152+
for (String sessionPropertyProviderName : workerSessionPropertyProviderFactories.keySet()) {
153+
loadSessionPropertyProvider(sessionPropertyProviderName);
154+
}
79155
}
80156

81157
public void addSystemSessionProperties(List<PropertyMetadata<?>> systemSessionProperties)
@@ -108,7 +184,9 @@ public void removeConnectorSessionProperties(ConnectorId connectorId)
108184
public Optional<PropertyMetadata<?>> getSystemSessionPropertyMetadata(String name)
109185
{
110186
requireNonNull(name, "name is null");
111-
187+
if (systemSessionProperties.get(name) == null) {
188+
return Optional.ofNullable(memoizedWorkerSessionProperties.get().get(name));
189+
}
112190
return Optional.ofNullable(systemSessionProperties.get(name));
113191
}
114192

@@ -124,6 +202,20 @@ public Optional<PropertyMetadata<?>> getConnectorSessionPropertyMetadata(Connect
124202
return Optional.ofNullable(properties.get(propertyName));
125203
}
126204

205+
private Map<String, PropertyMetadata<?>> getWorkerSessionProperties()
206+
{
207+
List<PropertyMetadata<?>> workerSessionPropertiesList = workerSessionPropertyProviders.values().stream()
208+
.flatMap(manager -> manager.getSessionProperties().stream())
209+
.collect(toImmutableList());
210+
Map<String, PropertyMetadata<?>> workerSessionProperties = new ConcurrentHashMap<>();
211+
workerSessionPropertiesList.forEach(sessionProperty -> {
212+
requireNonNull(sessionProperty, "sessionProperty is null");
213+
// TODO: Implement fail fast in case of duplicate entries.
214+
workerSessionProperties.put(sessionProperty.getName(), sessionProperty);
215+
});
216+
return workerSessionProperties;
217+
}
218+
127219
public List<SessionPropertyValue> getAllSessionProperties(Session session, Map<String, ConnectorId> catalogs)
128220
{
129221
requireNonNull(session, "session is null");
@@ -165,6 +257,19 @@ public List<SessionPropertyValue> getAllSessionProperties(Session session, Map<S
165257
}
166258
}
167259

260+
for (PropertyMetadata<?> property : new TreeMap<>(memoizedWorkerSessionProperties.get()).values()) {
261+
String defaultValue = firstNonNull(property.getDefaultValue(), "").toString();
262+
String value = systemProperties.getOrDefault(property.getName(), defaultValue);
263+
sessionPropertyValues.add(new SessionPropertyValue(
264+
value,
265+
defaultValue,
266+
property.getName(),
267+
Optional.empty(),
268+
property.getName(),
269+
property.getDescription(),
270+
property.getSqlType().getDisplayName(),
271+
property.isHidden()));
272+
}
168273
return sessionPropertyValues.build();
169274
}
170275

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -67,15 +67,15 @@
6767
import java.util.Optional;
6868
import java.util.stream.Collectors;
6969

70-
import static com.facebook.presto.SystemSessionProperties.getDistinctAggregationLargeBlockSizeThreshold;
71-
import static com.facebook.presto.SystemSessionProperties.isDedupBasedDistinctAggregationSpillEnabled;
72-
import static com.facebook.presto.SystemSessionProperties.isDistinctAggregationLargeBlockSpillEnabled;
7370
import static com.facebook.presto.common.Page.wrapBlocksWithoutCopy;
7471
import static com.facebook.presto.common.block.ColumnarArray.toColumnarArray;
7572
import static com.facebook.presto.common.block.ColumnarRow.toColumnarRow;
7673
import static com.facebook.presto.common.type.BigintType.BIGINT;
7774
import static com.facebook.presto.common.type.BooleanType.BOOLEAN;
7875
import static com.facebook.presto.common.type.VarcharType.VARCHAR;
76+
import static com.facebook.presto.sessionpropertyproviders.JavaWorkerSessionPropertyProvider.getDistinctAggregationLargeBlockSizeThreshold;
77+
import static com.facebook.presto.sessionpropertyproviders.JavaWorkerSessionPropertyProvider.isDedupBasedDistinctAggregationSpillEnabled;
78+
import static com.facebook.presto.sessionpropertyproviders.JavaWorkerSessionPropertyProvider.isDistinctAggregationLargeBlockSpillEnabled;
7979
import static com.google.common.base.Preconditions.checkArgument;
8080
import static com.google.common.base.Preconditions.checkState;
8181
import static com.google.common.collect.ImmutableList.toImmutableList;

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import com.facebook.presto.execution.warnings.WarningCollectorModule;
4040
import com.facebook.presto.metadata.Catalog;
4141
import com.facebook.presto.metadata.CatalogManager;
42+
import com.facebook.presto.metadata.SessionPropertyManager;
4243
import com.facebook.presto.metadata.StaticCatalogStore;
4344
import com.facebook.presto.metadata.StaticFunctionNamespaceStore;
4445
import com.facebook.presto.security.AccessControlManager;
@@ -179,6 +180,7 @@ public void run()
179180
injector.getInstance(NodeStatusNotificationManager.class).loadNodeStatusNotificationProvider();
180181
injector.getInstance(GracefulShutdownHandler.class).loadNodeStatusNotification();
181182
injector.getInstance(PlanCheckerProviderManager.class).loadPlanCheckerProviders();
183+
injector.getInstance(SessionPropertyManager.class).loadSessionPropertyProviders();
182184
startAssociatedProcesses(injector);
183185

184186
injector.getInstance(Announcer.class).start();

0 commit comments

Comments
 (0)