Skip to content
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.

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

Collect query semantic analysis query statistic #1084

Merged
merged 6 commits into from Aug 17, 2019
Merged
Changes from all commits
Commits
File filter...
Filter file types
Jump to…
Jump to file or symbol
Failed to load files and symbols.

Always

Just for now

@@ -49,8 +49,7 @@ on a Presto installation.
"endTime" : "2013-12-29T16:17:32.152-05:00",
"elapsedTime" : "125.00ms",
"queuedTime" : "1.31ms",
"analysisTime" : "4.84ms",
"distributedPlanningTime" : "353.00us",
"analysisTime" : "4.84ms",
"totalTasks" : 2,
"runningTasks" : 0,
"completedTasks" : 2,
@@ -63,7 +63,7 @@

.column("queued_time_ms", BIGINT)
.column("analysis_time_ms", BIGINT)
.column("distributed_planning_time_ms", BIGINT)
.column("planning_time_ms", BIGINT)

.column("created", TIMESTAMP)
.column("started", TIMESTAMP)
@@ -119,7 +119,7 @@ public RecordCursor cursor(ConnectorTransactionHandle transactionHandle, Connect

toMillis(queryStats.getQueuedTime()),
toMillis(queryStats.getAnalysisTime()),
toMillis(queryStats.getDistributedPlanningTime()),
toMillis(queryStats.getPlanningTime()),

toTimeStamp(queryStats.getCreateTime()),
toTimeStamp(queryStats.getExecutionStartTime()),
@@ -148,7 +148,6 @@ public void queryImmediateFailureEvent(BasicQueryInfo queryInfo, ExecutionFailur
ofMillis(queryInfo.getQueryStats().getQueuedTime().toMillis()),
Optional.empty(),
Optional.empty(),
Optional.empty(),
0,
0,
0,
@@ -229,7 +228,6 @@ private QueryStatistics createQueryStatistics(QueryInfo queryInfo)
ofMillis(queryStats.getQueuedTime().toMillis()),
Optional.of(ofMillis(queryStats.getResourceWaitingTime().toMillis())),
Optional.of(ofMillis(queryStats.getAnalysisTime().toMillis())),
Optional.of(ofMillis(queryStats.getDistributedPlanningTime().toMillis())),
queryStats.getPeakUserMemoryReservation().toBytes(),
queryStats.getPeakTotalMemoryReservation().toBytes(),
queryStats.getPeakTaskUserMemory().toBytes(),
@@ -387,7 +385,7 @@ private static void logQueryTimeline(QueryInfo queryInfo)
}

// planning duration -- start to end of planning
long planning = queryStats.getTotalPlanningTime().toMillis();
long planning = queryStats.getPlanningTime().toMillis();

// Time spent waiting for required no. of worker nodes to be present
long waiting = queryStats.getResourceWaitingTime().toMillis();
@@ -535,7 +535,6 @@ private QueryStats getQueryStats(Optional<StageInfo> rootStage)
queryStateTimer.getDispatchingTime(),
queryStateTimer.getExecutionTime(),
queryStateTimer.getAnalysisTime(),
queryStateTimer.getDistributedPlanningTime(),
queryStateTimer.getPlanningTime(),
queryStateTimer.getFinishingTime(),

@@ -897,24 +896,14 @@ public void recordHeartbeat()

public void beginAnalysis()
{
queryStateTimer.beginAnalyzing();
queryStateTimer.beginAnalysis();
}

public void endAnalysis()
{
queryStateTimer.endAnalysis();
}

public void beginDistributedPlanning()
{
queryStateTimer.beginDistributedPlanning();
}

public void endDistributedPlanning()
{
queryStateTimer.endDistributedPlanning();
}

public DateTime getCreateTime()
{
return queryStateTimer.getCreateTime();
@@ -1033,8 +1022,7 @@ private static QueryStats pruneQueryStats(QueryStats queryStats)
queryStats.getDispatchingTime(),
queryStats.getExecutionTime(),
queryStats.getAnalysisTime(),
queryStats.getDistributedPlanningTime(),
queryStats.getTotalPlanningTime(),
queryStats.getPlanningTime(),
queryStats.getFinishingTime(),
queryStats.getTotalTasks(),
queryStats.getRunningTasks(),
@@ -49,9 +49,6 @@
private final AtomicReference<Long> beginAnalysisNanos = new AtomicReference<>();
private final AtomicReference<Duration> analysisTime = new AtomicReference<>();

private final AtomicReference<Long> beginDistributedPlanningNanos = new AtomicReference<>();
private final AtomicReference<Duration> distributedPlanningTime = new AtomicReference<>();

private final AtomicReference<Long> lastHeartbeatNanos;

public QueryStateTimer(Ticker ticker)
@@ -149,7 +146,7 @@ private void endQuery(long now)
// Additional timings
//

public void beginAnalyzing()
public void beginAnalysis()
{
beginAnalysisNanos.compareAndSet(null, tickerNanos());
}
@@ -159,16 +156,6 @@ public void endAnalysis()
analysisTime.compareAndSet(null, nanosSince(beginAnalysisNanos, tickerNanos()));
}

public void beginDistributedPlanning()
{
beginDistributedPlanningNanos.compareAndSet(null, tickerNanos());
}

public void endDistributedPlanning()
{
distributedPlanningTime.compareAndSet(null, nanosSince(beginDistributedPlanningNanos, tickerNanos()));
}

public void recordHeartbeat()
{
lastHeartbeatNanos.set(tickerNanos());
@@ -242,11 +229,6 @@ public Duration getAnalysisTime()
return getDuration(analysisTime, beginAnalysisNanos);
}

public Duration getDistributedPlanningTime()
{
return getDuration(distributedPlanningTime, beginDistributedPlanningNanos);
}

public DateTime getLastHeartbeat()
{
return toDateTime(lastHeartbeatNanos.get());
@@ -50,8 +50,7 @@
private final Duration dispatchingTime;
private final Duration executionTime;
private final Duration analysisTime;
private final Duration distributedPlanningTime;
private final Duration totalPlanningTime;
private final Duration planningTime;
private final Duration finishingTime;

private final int totalTasks;
@@ -116,8 +115,7 @@ public QueryStats(
@JsonProperty("dispatchingTime") Duration dispatchingTime,
@JsonProperty("executionTime") Duration executionTime,
@JsonProperty("analysisTime") Duration analysisTime,
@JsonProperty("distributedPlanningTime") Duration distributedPlanningTime,
@JsonProperty("totalPlanningTime") Duration totalPlanningTime,
@JsonProperty("planningTime") Duration planningTime,
@JsonProperty("finishingTime") Duration finishingTime,

@JsonProperty("totalTasks") int totalTasks,
@@ -180,8 +178,7 @@ public QueryStats(
this.dispatchingTime = requireNonNull(dispatchingTime, "dispatchingTime is null");
this.executionTime = requireNonNull(executionTime, "executionTime is null");
this.analysisTime = requireNonNull(analysisTime, "analysisTime is null");
this.distributedPlanningTime = requireNonNull(distributedPlanningTime, "distributedPlanningTime is null");
this.totalPlanningTime = requireNonNull(totalPlanningTime, "totalPlanningTime is null");
this.planningTime = requireNonNull(planningTime, "planningTime is null");
this.finishingTime = requireNonNull(finishingTime, "finishingTime is null");

checkArgument(totalTasks >= 0, "totalTasks is negative");
@@ -308,15 +305,9 @@ public Duration getAnalysisTime()
}

@JsonProperty
public Duration getDistributedPlanningTime()
public Duration getPlanningTime()
{
return distributedPlanningTime;
}

@JsonProperty
public Duration getTotalPlanningTime()
{
return totalPlanningTime;
return planningTime;
}

@JsonProperty
@@ -173,18 +173,7 @@ private SqlQueryExecution(
this.stateMachine = requireNonNull(stateMachine, "stateMachine is null");

// analyze query
requireNonNull(preparedQuery, "preparedQuery is null");
Analyzer analyzer = new Analyzer(
stateMachine.getSession(),
metadata,
sqlParser,
accessControl,
Optional.of(queryExplainer),
preparedQuery.getParameters(),
warningCollector);
this.analysis = analyzer.analyze(preparedQuery.getStatement());

stateMachine.setUpdateType(analysis.getUpdateType());
this.analysis = analyze(preparedQuery, stateMachine, metadata, accessControl, sqlParser, queryExplainer, warningCollector);

// when the query finishes cache the final query info, and clear the reference to the output stage
AtomicReference<SqlQueryScheduler> queryScheduler = this.queryScheduler;
@@ -204,6 +193,35 @@ private SqlQueryExecution(
}
}

private Analysis analyze(
PreparedQuery preparedQuery,
QueryStateMachine stateMachine,
Metadata metadata,
AccessControl accessControl,
SqlParser sqlParser,
QueryExplainer queryExplainer,
WarningCollector warningCollector)
{
stateMachine.beginAnalysis();

requireNonNull(preparedQuery, "preparedQuery is null");
Analyzer analyzer = new Analyzer(
stateMachine.getSession(),
metadata,
sqlParser,
accessControl,
Optional.of(queryExplainer),
preparedQuery.getParameters(),
warningCollector);
Analysis analysis = analyzer.analyze(preparedQuery.getStatement());

stateMachine.setUpdateType(analysis.getUpdateType());

stateMachine.endAnalysis();

return analysis;
}

@Override
public String getSlug()
{
@@ -307,21 +325,17 @@ public void start()
{
try (SetThreadName ignored = new SetThreadName("Query-%s", stateMachine.getQueryId())) {
try {
// transition to planning

This comment has been minimized.

Copy link
@dain

dain Aug 15, 2019

Member

Typo in commit message coments => comments

if (!stateMachine.transitionToPlanning()) {
// query already started or finished
return;
}

// analyze query
PlanRoot plan = analyzeQuery();
PlanRoot plan = planQuery();

metadata.beginQuery(getSession(), plan.getTableHandles());

// plan distribution of query
planDistribution(plan);

// transition to starting
if (!stateMachine.transitionToStarting()) {
// query already started or finished
return;
@@ -361,21 +375,18 @@ public void addFinalQueryInfoListener(StateChangeListener<QueryInfo> stateChange
stateMachine.addQueryInfoStateChangeListener(stateChangeListener);
}

private PlanRoot analyzeQuery()
private PlanRoot planQuery()
{
try {
return doAnalyzeQuery();
return doPlanQuery();
}
catch (StackOverflowError e) {
throw new PrestoException(NOT_SUPPORTED, "statement is too large (stack overflow during analysis)", e);
}
}

private PlanRoot doAnalyzeQuery()
private PlanRoot doPlanQuery()
{
// time analysis phase
stateMachine.beginAnalysis();

// plan query
PlanNodeIdAllocator idAllocator = new PlanNodeIdAllocator();
LogicalPlanner logicalPlanner = new LogicalPlanner(stateMachine.getSession(), planOptimizers, idAllocator, metadata, new TypeAnalyzer(sqlParser, metadata), statsCalculator, costCalculator, stateMachine.getWarningCollector());
@@ -393,9 +404,6 @@ private PlanRoot doAnalyzeQuery()
// fragment the plan
SubPlan fragmentedPlan = planFragmenter.createSubPlans(stateMachine.getSession(), plan, false, stateMachine.getWarningCollector());

// record analysis time
stateMachine.endAnalysis();

boolean explainAnalyze = analysis.getStatement() instanceof Explain && ((Explain) analysis.getStatement()).isAnalyze();
return new PlanRoot(fragmentedPlan, !explainAnalyze, extractTableHandles(analysis));
}
@@ -418,13 +426,9 @@ private PlanRoot doAnalyzeQuery()

private void planDistribution(PlanRoot plan)
{
// time distribution planning
stateMachine.beginDistributedPlanning();

// plan the execution on the active nodes
DistributedExecutionPlanner distributedPlanner = new DistributedExecutionPlanner(splitManager, metadata);
StageExecutionPlan outputStageExecutionPlan = distributedPlanner.plan(plan.getRoot(), stateMachine.getSession());
stateMachine.endDistributedPlanning();

// ensure split sources are closed
stateMachine.addStateChangeListener(state -> {
@@ -182,7 +182,6 @@ private static QueryInfo toFullQueryInfo(DispatchQuery query)
ZERO_MILLIS,
ZERO_MILLIS,
ZERO_MILLIS,
ZERO_MILLIS,
0,
0,
0,

Large diffs are not rendered by default.

@@ -1242,12 +1242,20 @@ export class QueryDetail extends React.Component {
{query.queryStats.queuedTime}
</td>
</tr>
<tr>
<td className="info-title">
Analysis Time
</td>
<td className="info-text">
{query.queryStats.analysisTime}
</td>
</tr>
<tr>
<td className="info-title">
Planning Time
</td>
<td className="info-text">
{query.queryStats.totalPlanningTime}
{query.queryStats.planningTime}
</td>
</tr>
<tr>
@@ -176,7 +176,7 @@ private void assertAllTimeSpentInQueueing(QueryState expectedState, Consumer<Que
assertEquals(queryStats.getQueuedTime(), new Duration(7, MILLISECONDS));
assertEquals(queryStats.getResourceWaitingTime(), new Duration(0, MILLISECONDS));
assertEquals(queryStats.getDispatchingTime(), new Duration(0, MILLISECONDS));
assertEquals(queryStats.getTotalPlanningTime(), new Duration(0, MILLISECONDS));
assertEquals(queryStats.getPlanningTime(), new Duration(0, MILLISECONDS));
assertEquals(queryStats.getExecutionTime(), new Duration(0, MILLISECONDS));
assertEquals(queryStats.getFinishingTime(), new Duration(0, MILLISECONDS));
}
@@ -336,7 +336,7 @@ public void testPlanningTimeDuration()
assertEquals(queryStats.getQueuedTime().toMillis(), 25);
assertEquals(queryStats.getResourceWaitingTime().toMillis(), 50);
assertEquals(queryStats.getDispatchingTime().toMillis(), 100);
assertEquals(queryStats.getTotalPlanningTime().toMillis(), 200);
assertEquals(queryStats.getPlanningTime().toMillis(), 200);
// there is no way to induce finishing time without a transaction and connector
assertEquals(queryStats.getFinishingTime().toMillis(), 0);
// query execution time is starts when query transitions to planning
@@ -444,7 +444,7 @@ private static void assertState(QueryStateMachine stateMachine, QueryState expec
assertNotNull(queryStats.getResourceWaitingTime());
assertNotNull(queryStats.getDispatchingTime());
assertNotNull(queryStats.getExecutionTime());
assertNotNull(queryStats.getTotalPlanningTime());
assertNotNull(queryStats.getPlanningTime());
assertNotNull(queryStats.getFinishingTime());

assertNotNull(queryStats.getCreateTime());
ProTip! Use n and p to navigate between commits in a pull request.
You can’t perform that action at this time.