Skip to content

Commit

Permalink
Separate out java worker properties and introduce JavaFeaturesConfig
Browse files Browse the repository at this point in the history
  • Loading branch information
@pdabre12
pdabre12 authored and Pratik Joseph Dabre committed Sep 17, 2024
1 parent 0785ce8 commit 308dea6
Show file tree
Hide file tree
Showing 35 changed files with 694 additions and 497 deletions.
281 changes: 38 additions & 243 deletions presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java
View file

Large diffs are not rendered by default.

55 changes: 53 additions & 2 deletions presto-main/src/main/java/com/facebook/presto/metadata/SessionPropertyManager.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand All @@ -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)
Expand Down Expand Up @@ -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));
}

Expand All @@ -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);
Expand Down Expand Up @@ -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();
}

Expand Down
6 changes: 3 additions & 3 deletions ...ain/src/main/java/com/facebook/presto/operator/aggregation/GenericAccumulatorFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
11 changes: 11 additions & 0 deletions presto-main/src/main/java/com/facebook/presto/server/ServerMainModule.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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
Expand Down
Loading

0 comments on commit 308dea6

Please sign in to comment.