From 306bf5cde243df689bbf4722a0329334bd281aae Mon Sep 17 00:00:00 2001 From: morrySnow Date: Wed, 23 Aug 2023 17:29:52 +0800 Subject: [PATCH] [fix](Nereids) use Stopwatch to do timeout checker 1. avoid thread leak if exception thrown in planning 2. avoid memory release delays since the timer task hold CascadesContext object --- .../java/org/apache/doris/nereids/NereidsPlanner.java | 10 ++-------- .../org/apache/doris/nereids/StatementContext.java | 7 +++++++ .../nereids/jobs/scheduler/SimpleJobScheduler.java | 5 ++++- .../doris/nereids/rules/analysis/AnalyzeCTE.java | 3 ++- 4 files changed, 15 insertions(+), 10 deletions(-) diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java index 9fe20b1509b610..bb1ed834687a83 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/NereidsPlanner.java @@ -70,7 +70,6 @@ import java.util.Collections; import java.util.List; import java.util.Optional; -import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -115,7 +114,9 @@ public void plan(StatementBase queryStmt, org.apache.doris.thrift.TQueryOptions NereidsTracer.logImportantTime("EndParsePlan"); setParsedPlan(parsedPlan); PhysicalProperties requireProperties = buildInitRequireProperties(); + statementContext.getStopwatch().start(); Plan resultPlan = plan(parsedPlan, requireProperties, explainLevel); + statementContext.getStopwatch().stop(); setOptimizedPlan(resultPlan); if (explainLevel.isPlanLevel) { return; @@ -182,7 +183,6 @@ public Plan plan(LogicalPlan plan, PhysicalProperties requireProperties, Explain try (Lock lock = new Lock(plan, cascadesContext)) { // resolve column, table and function - Span queryAnalysisSpan = statementContext.getConnectContext().getTracer() .spanBuilder("query analysis").setParent(Context.current()).startSpan(); @@ -214,11 +214,6 @@ public Plan plan(LogicalPlan plan, PhysicalProperties requireProperties, Explain } } - Optional timeoutExecutor = Optional.empty(); - if (statementContext.getConnectContext().getSessionVariable().enableNereidsTimeout) { - timeoutExecutor = Optional.of(runTimeoutExecutor()); - } - // rule-based optimize rewrite(); if (explainLevel == ExplainLevel.REWRITTEN_PLAN || explainLevel == ExplainLevel.ALL_PLAN) { @@ -252,7 +247,6 @@ public Plan plan(LogicalPlan plan, PhysicalProperties requireProperties, Explain // serialize optimized plan to dumpfile, dumpfile do not have this part means optimize failed MinidumpUtils.serializeOutputToDumpFile(physicalPlan); NereidsTracer.output(statementContext.getConnectContext()); - timeoutExecutor.ifPresent(ExecutorService::shutdown); return physicalPlan; } diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java index 59d23db2923e7c..5f7dcbe0bd7c94 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java @@ -35,6 +35,7 @@ import org.apache.doris.qe.ConnectContext; import org.apache.doris.qe.OriginStatement; +import com.google.common.base.Stopwatch; import com.google.common.base.Supplier; import com.google.common.base.Suppliers; import com.google.common.collect.ImmutableList; @@ -54,6 +55,8 @@ public class StatementContext { private ConnectContext connectContext; + private final Stopwatch stopwatch = Stopwatch.createUnstarted(); + @GuardedBy("this") private final Map> contextCacheMap = Maps.newLinkedHashMap(); @@ -108,6 +111,10 @@ public OriginStatement getOriginStatement() { return originStatement; } + public Stopwatch getStopwatch() { + return stopwatch; + } + public void setMaxNAryInnerJoin(int maxNAryInnerJoin) { if (maxNAryInnerJoin > this.maxNAryInnerJoin) { this.maxNAryInnerJoin = maxNAryInnerJoin; diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/scheduler/SimpleJobScheduler.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/scheduler/SimpleJobScheduler.java index 940f9d572f28e5..c52e2e90f13d60 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/scheduler/SimpleJobScheduler.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/scheduler/SimpleJobScheduler.java @@ -20,6 +20,8 @@ import org.apache.doris.nereids.CascadesContext; import org.apache.doris.nereids.jobs.Job; +import java.util.concurrent.TimeUnit; + /** * Single thread, serial scheduler. */ @@ -29,7 +31,8 @@ public void executeJobPool(ScheduleContext scheduleContext) { JobPool pool = scheduleContext.getJobPool(); while (!pool.isEmpty()) { CascadesContext context = (CascadesContext) scheduleContext; - if (context.isTimeout()) { + if (context.getConnectContext().getSessionVariable().enableNereidsTimeout + && context.getStatementContext().getStopwatch().elapsed(TimeUnit.MILLISECONDS) > 5000) { throw new RuntimeException("Nereids cost too much time ( > 5s )"); } Job job = pool.pop(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/AnalyzeCTE.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/AnalyzeCTE.java index 65d9607a1ce2e2..a465df6da13ef9 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/AnalyzeCTE.java +++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/AnalyzeCTE.java @@ -25,6 +25,7 @@ import org.apache.doris.nereids.rules.RuleType; import org.apache.doris.nereids.trees.expressions.CTEId; import org.apache.doris.nereids.trees.expressions.Slot; +import org.apache.doris.nereids.trees.expressions.StatementScopeIdGenerator; import org.apache.doris.nereids.trees.plans.Plan; import org.apache.doris.nereids.trees.plans.logical.LogicalCTE; import org.apache.doris.nereids.trees.plans.logical.LogicalCTEAnchor; @@ -89,7 +90,7 @@ private Pair>> analyzeCte( innerCascadesCtx.newAnalyzer().analyze(); LogicalPlan analyzedCtePlan = (LogicalPlan) innerCascadesCtx.getRewritePlan(); checkColumnAlias(aliasQuery, analyzedCtePlan.getOutput()); - CTEId cteId = cascadesContext.getStatementContext().getNextCTEId(); + CTEId cteId = StatementScopeIdGenerator.newCTEId(); LogicalSubQueryAlias logicalSubQueryAlias = aliasQuery.withChildren(ImmutableList.of(analyzedCtePlan)); outerCteCtx = new CTEContext(cteId, logicalSubQueryAlias, outerCteCtx);