Skip to content

Support customizable ways of launching Presto queries in Verifier #14823

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
caithagoras merged 10 commits into from
Jul 30, 2020
Merged

Support customizable ways of launching Presto queries in Verifier #14823

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
d7018f4
Style improvments in DeterminismAnalyzer
caithagoras0 Jul 10, 2020
34a6906
Make control.http-port optional
caithagoras0 Jul 9, 2020
dbb1796
Make the entire JDBC QueryStats available in verification results
caithagoras0 Jul 10, 2020
93fb5c5
Support customizable ways of launching Presto queries in Verifier
caithagoras0 Jul 10, 2020
8e9b86f
Allow additional data to be emitted in QueryStatsEvent
caithagoras0 Jul 10, 2020
a38463c
Use a single metadata-timeout and checksum-timeout
caithagoras0 Jul 10, 2020
8f3754c
Move the logic of mangling session properties into a utility method
caithagoras0 Jul 17, 2020
84ad4a4
Fix binding failure when failure resolver is disabled
caithagoras0 Jul 20, 2020
b31e88a
Move Query#toStatementStats to a utility class
caithagoras0 Jul 23, 2020
2f85b49
Introduce additional methods in SqlExceptionClassifier interface
caithagoras0 Jul 28, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 1 addition & 91 deletions presto-main/src/main/java/com/facebook/presto/server/protocol/Query.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,6 @@
import com.facebook.presto.client.FailureInfo;
import com.facebook.presto.client.QueryError;
import com.facebook.presto.client.QueryResults;
import com.facebook.presto.client.StageStats;
import com.facebook.presto.client.StatementStats;
import com.facebook.presto.common.Page;
import com.facebook.presto.common.block.BlockEncodingSerde;
import com.facebook.presto.common.type.BooleanType;
Expand All @@ -30,11 +28,7 @@
import com.facebook.presto.execution.QueryInfo;
import com.facebook.presto.execution.QueryManager;
import com.facebook.presto.execution.QueryState;
import com.facebook.presto.execution.QueryStats;
import com.facebook.presto.execution.StageExecutionInfo;
import com.facebook.presto.execution.StageExecutionStats;
import com.facebook.presto.execution.StageInfo;
import com.facebook.presto.execution.TaskInfo;
import com.facebook.presto.execution.buffer.PagesSerdeFactory;
import com.facebook.presto.operator.ExchangeClient;
import com.facebook.presto.spi.ErrorCode;
Expand All @@ -61,7 +55,6 @@
import javax.ws.rs.core.UriInfo;

import java.net.URI;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
Expand All @@ -76,6 +69,7 @@
import static com.facebook.presto.SystemSessionProperties.getTargetResultSize;
import static com.facebook.presto.SystemSessionProperties.isExchangeCompressionEnabled;
import static com.facebook.presto.execution.QueryState.FAILED;
import static com.facebook.presto.server.protocol.QueryResourceUtil.toStatementStats;
import static com.facebook.presto.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
import static com.facebook.presto.util.Failures.toFailure;
import static com.google.common.base.MoreObjects.firstNonNull;
Expand Down Expand Up @@ -523,90 +517,6 @@ private synchronized URI createNextResultsUri(String scheme, UriInfo uriInfo, lo
return uri.build();
}

private static StatementStats toStatementStats(QueryInfo queryInfo)
{
QueryStats queryStats = queryInfo.getQueryStats();
StageInfo outputStage = queryInfo.getOutputStage().orElse(null);

return StatementStats.builder()
.setState(queryInfo.getState().toString())
.setQueued(queryInfo.getState() == QueryState.QUEUED)
.setScheduled(queryInfo.isScheduled())
.setNodes(globalUniqueNodes(outputStage).size())
.setTotalSplits(queryStats.getTotalDrivers())
.setQueuedSplits(queryStats.getQueuedDrivers())
.setRunningSplits(queryStats.getRunningDrivers() + queryStats.getBlockedDrivers())
.setCompletedSplits(queryStats.getCompletedDrivers())
.setCpuTimeMillis(queryStats.getTotalCpuTime().toMillis())
.setWallTimeMillis(queryStats.getTotalScheduledTime().toMillis())
.setQueuedTimeMillis(queryStats.getQueuedTime().toMillis())
.setElapsedTimeMillis(queryStats.getElapsedTime().toMillis())
.setProcessedRows(queryStats.getRawInputPositions())
.setProcessedBytes(queryStats.getRawInputDataSize().toBytes())
.setPeakMemoryBytes(queryStats.getPeakUserMemoryReservation().toBytes())
.setPeakTotalMemoryBytes(queryStats.getPeakTotalMemoryReservation().toBytes())
.setPeakTaskTotalMemoryBytes(queryStats.getPeakTaskTotalMemory().toBytes())
.setSpilledBytes(queryStats.getSpilledDataSize().toBytes())
.setRootStage(toStageStats(outputStage))
.build();
}

private static StageStats toStageStats(StageInfo stageInfo)
{
if (stageInfo == null) {
return null;
}

StageExecutionInfo currentStageExecutionInfo = stageInfo.getLatestAttemptExecutionInfo();
StageExecutionStats stageExecutionStats = currentStageExecutionInfo.getStats();

ImmutableList.Builder<StageStats> subStages = ImmutableList.builder();
for (StageInfo subStage : stageInfo.getSubStages()) {
subStages.add(toStageStats(subStage));
}

Set<String> uniqueNodes = new HashSet<>();
for (TaskInfo task : currentStageExecutionInfo.getTasks()) {
// todo add nodeId to TaskInfo
URI uri = task.getTaskStatus().getSelf();
uniqueNodes.add(uri.getHost() + ":" + uri.getPort());
}

return StageStats.builder()
.setStageId(String.valueOf(stageInfo.getStageId().getId()))
.setState(currentStageExecutionInfo.getState().toString())
.setDone(currentStageExecutionInfo.getState().isDone())
.setNodes(uniqueNodes.size())
.setTotalSplits(stageExecutionStats.getTotalDrivers())
.setQueuedSplits(stageExecutionStats.getQueuedDrivers())
.setRunningSplits(stageExecutionStats.getRunningDrivers() + stageExecutionStats.getBlockedDrivers())
.setCompletedSplits(stageExecutionStats.getCompletedDrivers())
.setCpuTimeMillis(stageExecutionStats.getTotalCpuTime().toMillis())
.setWallTimeMillis(stageExecutionStats.getTotalScheduledTime().toMillis())
.setProcessedRows(stageExecutionStats.getRawInputPositions())
.setProcessedBytes(stageExecutionStats.getRawInputDataSize().toBytes())
.setSubStages(subStages.build())
.build();
}

private static Set<String> globalUniqueNodes(StageInfo stageInfo)
{
if (stageInfo == null) {
return ImmutableSet.of();
}
ImmutableSet.Builder<String> nodes = ImmutableSet.builder();
for (TaskInfo task : stageInfo.getLatestAttemptExecutionInfo().getTasks()) {
// todo add nodeId to TaskInfo
URI uri = task.getTaskStatus().getSelf();
nodes.add(uri.getHost() + ":" + uri.getPort());
}

for (StageInfo subStage : stageInfo.getSubStages()) {
nodes.addAll(globalUniqueNodes(subStage));
}
return nodes.build();
}

private static URI findCancelableLeafStage(QueryInfo queryInfo)
{
// if query is running, find the leaf-most running stage
Expand Down
98 changes: 98 additions & 0 deletions presto-main/src/main/java/com/facebook/presto/server/protocol/QueryResourceUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,12 +14,26 @@
package com.facebook.presto.server.protocol;

import com.facebook.presto.client.QueryResults;
import com.facebook.presto.client.StageStats;
import com.facebook.presto.client.StatementStats;
import com.facebook.presto.execution.QueryInfo;
import com.facebook.presto.execution.QueryState;
import com.facebook.presto.execution.QueryStats;
import com.facebook.presto.execution.StageExecutionInfo;
import com.facebook.presto.execution.StageExecutionStats;
import com.facebook.presto.execution.StageInfo;
import com.facebook.presto.execution.TaskInfo;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;

import javax.ws.rs.core.Response;

import java.io.UnsupportedEncodingException;
import java.net.URI;
import java.net.URLEncoder;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import static com.facebook.presto.client.PrestoHeaders.PRESTO_ADDED_PREPARE;
import static com.facebook.presto.client.PrestoHeaders.PRESTO_CLEAR_SESSION;
Expand Down Expand Up @@ -79,6 +93,34 @@ public static Response toResponse(Query query, QueryResults queryResults)
return response.build();
}

public static StatementStats toStatementStats(QueryInfo queryInfo)
{
QueryStats queryStats = queryInfo.getQueryStats();
StageInfo outputStage = queryInfo.getOutputStage().orElse(null);

return StatementStats.builder()
.setState(queryInfo.getState().toString())
.setQueued(queryInfo.getState() == QueryState.QUEUED)
.setScheduled(queryInfo.isScheduled())
.setNodes(globalUniqueNodes(outputStage).size())
.setTotalSplits(queryStats.getTotalDrivers())
.setQueuedSplits(queryStats.getQueuedDrivers())
.setRunningSplits(queryStats.getRunningDrivers() + queryStats.getBlockedDrivers())
.setCompletedSplits(queryStats.getCompletedDrivers())
.setCpuTimeMillis(queryStats.getTotalCpuTime().toMillis())
.setWallTimeMillis(queryStats.getTotalScheduledTime().toMillis())
.setQueuedTimeMillis(queryStats.getQueuedTime().toMillis())
.setElapsedTimeMillis(queryStats.getElapsedTime().toMillis())
.setProcessedRows(queryStats.getRawInputPositions())
.setProcessedBytes(queryStats.getRawInputDataSize().toBytes())
.setPeakMemoryBytes(queryStats.getPeakUserMemoryReservation().toBytes())
.setPeakTotalMemoryBytes(queryStats.getPeakTotalMemoryReservation().toBytes())
.setPeakTaskTotalMemoryBytes(queryStats.getPeakTaskTotalMemory().toBytes())
.setSpilledBytes(queryStats.getSpilledDataSize().toBytes())
.setRootStage(toStageStats(outputStage))
.build();
}

private static String urlEncode(String value)
{
try {
Expand All @@ -88,4 +130,60 @@ private static String urlEncode(String value)
throw new AssertionError(e);
}
}

private static Set<String> globalUniqueNodes(StageInfo stageInfo)
{
if (stageInfo == null) {
return ImmutableSet.of();
}
ImmutableSet.Builder<String> nodes = ImmutableSet.builder();
for (TaskInfo task : stageInfo.getLatestAttemptExecutionInfo().getTasks()) {
// todo add nodeId to TaskInfo
URI uri = task.getTaskStatus().getSelf();
nodes.add(uri.getHost() + ":" + uri.getPort());
}

for (StageInfo subStage : stageInfo.getSubStages()) {
nodes.addAll(globalUniqueNodes(subStage));
}
return nodes.build();
}

private static StageStats toStageStats(StageInfo stageInfo)
{
if (stageInfo == null) {
return null;
}

StageExecutionInfo currentStageExecutionInfo = stageInfo.getLatestAttemptExecutionInfo();
StageExecutionStats stageExecutionStats = currentStageExecutionInfo.getStats();

ImmutableList.Builder<StageStats> subStages = ImmutableList.builder();
for (StageInfo subStage : stageInfo.getSubStages()) {
subStages.add(toStageStats(subStage));
}

Set<String> uniqueNodes = new HashSet<>();
for (TaskInfo task : currentStageExecutionInfo.getTasks()) {
// todo add nodeId to TaskInfo
URI uri = task.getTaskStatus().getSelf();
uniqueNodes.add(uri.getHost() + ":" + uri.getPort());
}

return StageStats.builder()
.setStageId(String.valueOf(stageInfo.getStageId().getId()))
.setState(currentStageExecutionInfo.getState().toString())
.setDone(currentStageExecutionInfo.getState().isDone())
.setNodes(uniqueNodes.size())
.setTotalSplits(stageExecutionStats.getTotalDrivers())
.setQueuedSplits(stageExecutionStats.getQueuedDrivers())
.setRunningSplits(stageExecutionStats.getRunningDrivers() + stageExecutionStats.getBlockedDrivers())
.setCompletedSplits(stageExecutionStats.getCompletedDrivers())
.setCpuTimeMillis(stageExecutionStats.getTotalCpuTime().toMillis())
.setWallTimeMillis(stageExecutionStats.getTotalScheduledTime().toMillis())
.setProcessedRows(stageExecutionStats.getRawInputPositions())
.setProcessedBytes(stageExecutionStats.getRawInputDataSize().toBytes())
.setSubStages(subStages.build())
.build();
}
}
6 changes: 6 additions & 0 deletions presto-verifier/src/main/java/com/facebook/presto/verifier/PrestoVerifyCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,6 +55,12 @@ public Set<String> getCustomEventClientTypes()
return ImmutableSet.of();
}

@Override
public Set<String> getCustomQueryActionTypes()
{
return ImmutableSet.of();
}

@Override
public List<Class<? extends Predicate<SourceQuery>>> getCustomQueryFilterClasses()
{
Expand Down
31 changes: 31 additions & 0 deletions presto-verifier/src/main/java/com/facebook/presto/verifier/annotation/ForHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
/*
* 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.verifier.annotation;

import javax.inject.Qualifier;

import java.lang.annotation.Retention;
import java.lang.annotation.Target;

import static java.lang.annotation.ElementType.FIELD;
import static java.lang.annotation.ElementType.METHOD;
import static java.lang.annotation.ElementType.PARAMETER;
import static java.lang.annotation.RetentionPolicy.RUNTIME;

@Retention(RUNTIME)
@Target({FIELD, PARAMETER, METHOD})
@Qualifier
public @interface ForHelper
{
}
13 changes: 11 additions & 2 deletions presto-verifier/src/main/java/com/facebook/presto/verifier/event/QueryInfo.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,13 +43,14 @@ public class QueryInfo
private final Double wallTimeSecs;
private final Long peakTotalMemoryBytes;
private final Long peakTaskTotalMemoryBytes;
private final QueryStatsEvent queryStats;

public QueryInfo(
String catalog,
String schema,
String originalQuery)
{
this(catalog, schema, originalQuery, Optional.empty(), ImmutableList.of(), ImmutableList.of(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty());
this(catalog, schema, originalQuery, Optional.empty(), ImmutableList.of(), ImmutableList.of(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty(), Optional.empty());
}

public QueryInfo(
Expand All @@ -67,7 +68,8 @@ public QueryInfo(
Optional<Double> cpuTimeSecs,
Optional<Double> wallTimeSecs,
Optional<Long> peakTotalMemoryBytes,
Optional<Long> peakTaskTotalMemoryBytes)
Optional<Long> peakTaskTotalMemoryBytes,
Copy link
Copy Markdown
Contributor

@wenleix wenleix Jul 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

wow. why not using OptionalLong/OptionalDouble? -- this is irrelevant to this PR though

Copy link
Copy Markdown
Contributor Author

@caithagoras caithagoras Jul 23, 2020

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since we have everything in the QueryStatsEvent, we'll eventually deprecate those fields.

Optional<QueryStatsEvent> queryStats)
{
this.catalog = requireNonNull(catalog, "catalog is null");
this.schema = requireNonNull(schema, "schema is null");
Expand All @@ -84,6 +86,7 @@ public QueryInfo(
this.wallTimeSecs = wallTimeSecs.orElse(null);
this.peakTotalMemoryBytes = peakTotalMemoryBytes.orElse(null);
this.peakTaskTotalMemoryBytes = peakTaskTotalMemoryBytes.orElse(null);
this.queryStats = queryStats.orElse(null);
}

@EventField
Expand Down Expand Up @@ -175,4 +178,10 @@ public Long getPeakTaskTotalMemoryBytes()
{
return peakTaskTotalMemoryBytes;
}

@EventField
public QueryStatsEvent getQueryStats()
{
return queryStats;
}
}
Loading