Ryan19929 commented on code in PR #68727: URL: https://github.com/apache/doris/pull/68727#discussion_r4253916246
########## fe/fe-core/src/main/java/org/apache/doris/common/profile/PlanningDiagnostics.java: ########## @@ -0,0 +1,437 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 org.apache.doris.common.profile; + +import org.apache.doris.common.Config; +import org.apache.doris.common.util.DebugUtil; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.thrift.TUniqueId; + +import com.google.common.annotations.VisibleForTesting; +import com.google.gson.JsonObject; +import org.apache.logging.log4j.CloseableThreadContext; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.ArrayList; +import java.util.EnumMap; +import java.util.List; +import java.util.Locale; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongSupplier; +import java.util.function.Supplier; + +/** + * One planning pass. The planning thread owns timers and the bounded slow-operation buffer. + * The connection checker reads only immutable, volatile-published steps, never planner/table locks. + * Timings are inclusive: nested phases and operations must not be added to their enclosing phase. + */ +public final class PlanningDiagnostics { + private static final Logger LOG = LogManager.getLogger(PlanningDiagnostics.class); + private static final int MAX_SLOW_OPERATIONS = 16; + // Internal MV planning can replace ConnectContext while the calling planner still holds table locks. + private static final ThreadLocal<PlanningDiagnostics> ACTIVE = new ThreadLocal<>(); + + public enum Phase { + PREPROCESS, COLLECT_TABLES, PRELOAD_METADATA, WAIT_CHANGE_VISIBLE, LOCK_TABLES, + ANALYZE, REWRITE, PRE_REWRITE_MV, OPTIMIZE, CHOOSE_PLAN, POST_PROCESS, + TRANSLATE, DISTRIBUTE, RELEASE_RESOURCES; + + public String key() { + return name().toLowerCase(Locale.ROOT); + } + } + + private final ConnectContext context; + private final SummaryProfile summary; + private final long passId; + private final PlanningDiagnostics parent; + private final PlanningDiagnostics root; + private final boolean inheritsQueryIdentity; + private final TUniqueId queryId; + private final long statementId; + private final String threadName; + private volatile long timeoutMs; + private final LongSupplier clock; + private final long started; + private final EnumMap<Phase, Timing> timings = new EnumMap<>(Phase.class); + private final List<Event> slowOperations = new ArrayList<>(); + private final AtomicLong lastReport; + private volatile Step current; + private volatile Class<?> currentJob; + private volatile HeldLocks heldLocks = new HeldLocks(0, 0, ""); + private volatile boolean finished; + // Only the root publishes this pointer, so its registered connection can inspect an internal pass. + private volatile PlanningDiagnostics activePass; + private String failedStep = ""; + private Throwable stepFailure; + private long omittedOperations; + + private PlanningDiagnostics(ConnectContext context) { + this(context, System::nanoTime); + } + + @VisibleForTesting + PlanningDiagnostics(ConnectContext context, LongSupplier clock) { + this.context = context; + this.parent = ACTIVE.get(); + this.root = parent == null ? this : parent.root; + this.lastReport = parent == null ? new AtomicLong(Long.MIN_VALUE) : parent.lastReport; + SummaryProfile contextSummary = SummaryProfile.getSummaryProfile(context); + this.summary = contextSummary != null ? contextSummary : parent == null ? null : parent.summary; Review Comment: [P2] Nested MV planning passes are double-counted in the `planning_*` audit totals When MV rewrite plans an MV definition in-line, `InitMaterializationContextHook` (around :283-291) calls `MTMVPlanUtil.createBasicMvContext(...)` → `MTMVCache.from(...)` → `planWithLock(...)`. That context has no `SummaryProfile`, so the inner pass falls back to `parent.summary` here, and its `finish()` adds the inner phase times to the same totals through `SummaryProfile.addPlanningPass`. The outer pass's rewrite phase already includes all of the inner planning. As a result: - `planning_rewrite` counts the inner rewrite twice; - `planning_analyze` / `planning_optimize` include time spent planning the MV-definition SQL; - `planning_passes` becomes `K + 1` for `K` such MVs. `testNestedAndRepeatedPlanningRestoreParentAndAccumulate` encodes this behavior, which contradicts the "nested phases must not be added to their enclosing phase" note. Consider aggregating only root passes into the audit totals and keeping the nested breakdown in the Profile, or define the per-key semantics explicitly. ########## fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java: ########## @@ -582,6 +584,21 @@ public boolean isWarmup() { @ResetOnRetry(ResetScope.PLAN) private long waitChangeVisibleEndTime = -1L; + @SerializedName("planningPassCount") + private long planningPassCount; Review Comment: [P2] Retry semantics are mixed in one map, and the audit field roughly doubles These new fields have no `@ResetOnRetry`. After a retry, the legacy `plan_times_ms` keys (`analyze`, `rewrite`, ...) describe only the last attempt, while the `planning_*` keys sum all attempts, so the same map carries two meanings. Either reset them with `ResetScope.PLAN`, or document that they are cumulative. Size: every audited statement gains about 19 keys, six of which overlap existing ones (`analyze`, `rewrite`, `optimize`, `lock_tables`, `translate`, `distribute`). That grows every row in `fe.audit.log` and `__internal_schema.audit_log`. Consider limiting the audit field to two or three summary keys (passes / failed passes / last failed) and keeping the per-phase detail in the Profile. ########## fe/fe-core/src/main/java/org/apache/doris/common/profile/PlanningDiagnostics.java: ########## @@ -0,0 +1,437 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 org.apache.doris.common.profile; + +import org.apache.doris.common.Config; +import org.apache.doris.common.util.DebugUtil; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.thrift.TUniqueId; + +import com.google.common.annotations.VisibleForTesting; +import com.google.gson.JsonObject; +import org.apache.logging.log4j.CloseableThreadContext; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.ArrayList; +import java.util.EnumMap; +import java.util.List; +import java.util.Locale; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongSupplier; +import java.util.function.Supplier; + +/** + * One planning pass. The planning thread owns timers and the bounded slow-operation buffer. + * The connection checker reads only immutable, volatile-published steps, never planner/table locks. + * Timings are inclusive: nested phases and operations must not be added to their enclosing phase. + */ +public final class PlanningDiagnostics { + private static final Logger LOG = LogManager.getLogger(PlanningDiagnostics.class); + private static final int MAX_SLOW_OPERATIONS = 16; + // Internal MV planning can replace ConnectContext while the calling planner still holds table locks. + private static final ThreadLocal<PlanningDiagnostics> ACTIVE = new ThreadLocal<>(); + + public enum Phase { + PREPROCESS, COLLECT_TABLES, PRELOAD_METADATA, WAIT_CHANGE_VISIBLE, LOCK_TABLES, + ANALYZE, REWRITE, PRE_REWRITE_MV, OPTIMIZE, CHOOSE_PLAN, POST_PROCESS, + TRANSLATE, DISTRIBUTE, RELEASE_RESOURCES; + + public String key() { + return name().toLowerCase(Locale.ROOT); + } + } + + private final ConnectContext context; + private final SummaryProfile summary; + private final long passId; + private final PlanningDiagnostics parent; + private final PlanningDiagnostics root; + private final boolean inheritsQueryIdentity; + private final TUniqueId queryId; + private final long statementId; + private final String threadName; + private volatile long timeoutMs; + private final LongSupplier clock; + private final long started; + private final EnumMap<Phase, Timing> timings = new EnumMap<>(Phase.class); + private final List<Event> slowOperations = new ArrayList<>(); + private final AtomicLong lastReport; + private volatile Step current; + private volatile Class<?> currentJob; + private volatile HeldLocks heldLocks = new HeldLocks(0, 0, ""); + private volatile boolean finished; + // Only the root publishes this pointer, so its registered connection can inspect an internal pass. + private volatile PlanningDiagnostics activePass; + private String failedStep = ""; + private Throwable stepFailure; + private long omittedOperations; + + private PlanningDiagnostics(ConnectContext context) { + this(context, System::nanoTime); + } + + @VisibleForTesting + PlanningDiagnostics(ConnectContext context, LongSupplier clock) { + this.context = context; + this.parent = ACTIVE.get(); + this.root = parent == null ? this : parent.root; + this.lastReport = parent == null ? new AtomicLong(Long.MIN_VALUE) : parent.lastReport; + SummaryProfile contextSummary = SummaryProfile.getSummaryProfile(context); + this.summary = contextSummary != null ? contextSummary : parent == null ? null : parent.summary; + this.passId = summary == null ? 0 : summary.nextPlanningPassId(); + TUniqueId contextQueryId = context.queryId(); + this.inheritsQueryIdentity = contextQueryId == null && parent != null; + this.queryId = inheritsQueryIdentity ? parent.queryId + : contextQueryId == null ? null : contextQueryId.deepCopy(); + this.statementId = inheritsQueryIdentity ? parent.statementId : context.getStmtId(); + this.threadName = Thread.currentThread().getName(); + this.timeoutMs = inheritsQueryIdentity ? parent.timeoutMs : context.getExecTimeoutS() * 1000L; + this.clock = clock; + this.started = clock.getAsLong(); + } + + /** Includes cleanup and restores an outer pass even when internal planning uses another connection. */ + public static <T> T plan(ConnectContext context, Supplier<T> action) { + return execute(new PlanningDiagnostics(context), action); + } + + @VisibleForTesting + static <T> T execute(PlanningDiagnostics diagnostics, Supplier<T> action) { + ConnectContext context = diagnostics.context; + PlanningDiagnostics previous = context.getPlanningDiagnostics(); + ACTIVE.set(diagnostics); + context.setPlanningDiagnostics(diagnostics); + diagnostics.root.activePass = diagnostics; + boolean success = false; + try { + T result = action.get(); + success = true; + return result; + } catch (RuntimeException | Error failure) { + if (diagnostics.stepFailure != failure) { + diagnostics.failedStep = "between_phases"; + } + throw failure; + } finally { + diagnostics.finished = true; + context.setPlanningDiagnostics(previous); + diagnostics.root.activePass = diagnostics.parent; + if (diagnostics.parent == null) { + ACTIVE.remove(); + } else { + ACTIVE.set(diagnostics.parent); + } + diagnostics.finish(success); + } + } + + public static void runPhase(ConnectContext context, Phase phase, Runnable action) { + phase(context, phase, () -> { + action.run(); + return null; + }); + } + + public static <T> T phase(ConnectContext context, Phase phase, Supplier<T> action) { + PlanningDiagnostics diagnostics = current(context); + return diagnostics == null ? action.get() : diagnostics.measure(phase, "", "", -1, action); + } + + /** A negative operation budget means that timeout handling belongs to the connector. */ + public static <T> T operation(ConnectContext context, String operation, String target, + long budgetMs, Supplier<T> action) { + return operation(context, operation, () -> target, budgetMs, action); + } + + /** Resolve the target only for an active planning pass, on the calling thread. */ + public static <T> T operation(ConnectContext context, String operation, Supplier<String> target, + long budgetMs, Supplier<T> action) { + PlanningDiagnostics diagnostics = current(context); + if (diagnostics == null) { + return action.get(); + } + Step step = diagnostics.current; + return diagnostics.measure(step == null ? Phase.PREPROCESS : step.phase, + operation, target.get(), budgetMs, action); + } + + public static PlanningDiagnostics current(ConnectContext context) { + return context == null ? null : context.getPlanningDiagnostics(); + } + + private <T> T measure(Phase phase, String operation, String target, long budgetMs, Supplier<T> action) { + Step previous = current; + Step step = new Step(phase, operation, target, budgetMs, clock.getAsLong(), + operation.isEmpty() || previous == null ? -1 : previous.phaseStarted); + current = step; + boolean success = false; + try { + T result = action.get(); + success = true; + return result; + } catch (RuntimeException | Error failure) { + // Preserve the innermost failing step of this exception, not an earlier recovered MV failure. + if (stepFailure != failure) { + stepFailure = failure; + failedStep = phase.key() + (operation.isEmpty() ? "" : "/" + operation); + } + throw failure; + } finally { + long elapsed = clock.getAsLong() - step.started; + if (operation.isEmpty()) { + Timing timing = timings.computeIfAbsent(phase, key -> new Timing()); + timing.nanos += elapsed; + timing.calls++; + timing.failures += success ? 0 : 1; + if (phase == Phase.PREPROCESS && success) { + // SET_VAR hints are applied by preprocessing, after this pass was created. + timeoutMs = inheritsQueryIdentity ? parent.timeoutMs : context.getExecTimeoutS() * 1000L; + } + } else { + recordOperation(step, elapsed, success); + } + current = previous; + } + } + + /** Avoid allocating task descriptions for every optimizer job; format the class only on a slow report. */ + public Class<?> setCurrentJob(Class<?> job) { + Class<?> previous = currentJob; + currentJob = job; + return previous; + } + + public LockHold acquiredLock(String table) { + long now = clock.getAsLong(); + HeldLocks previous = heldLocks; + heldLocks = new HeldLocks(previous.count + 1, previous.count == 0 ? now : previous.started, + previous.count == 0 ? table : previous.oldestTable); + return new LockHold(table, now); + } + + /** Closed after unlocking; planner resources release in reverse acquisition order. */ + public final class LockHold implements AutoCloseable { + private final String table; + private final long acquired; + + private LockHold(String table, long acquired) { + this.table = table; + this.acquired = acquired; + } + + @Override + public void close() { + HeldLocks previous = heldLocks; + heldLocks = new HeldLocks(previous.count - 1, previous.started, previous.oldestTable); + recordOperation(new Step(Phase.RELEASE_RESOURCES, "read_lock_hold", table, -1, acquired, -1), + clock.getAsLong() - acquired, true); + } + } + + private void recordOperation(Step step, long nanos, boolean success) { + long threshold = Config.nereids_planning_operation_log_threshold_ms; + if (success && (threshold <= 0 || millis(nanos) < threshold)) { + return; + } + if (slowOperations.size() == MAX_SLOW_OPERATIONS) { + omittedOperations++; + return; + } + JsonObject event = step.toJson(); + event.addProperty("elapsed_ms", millis(nanos)); + event.addProperty("status", success ? "completed" : "failed"); + slowOperations.add(new Event(this, "Planning operation", event)); + } + + /** Invoked by the existing connection timeout checker, including while a planner job is blocked. */ + public void reportIfSlow() { + long threshold = Config.nereids_planning_log_threshold_ms; + PlanningDiagnostics active = root.activePass; + if (root.finished || active == null || threshold <= 0) { + return; + } + // Snapshot published starts before reading the clock. A newer step/lock sampled after the clock + // could otherwise appear to have started in the future when the checker was descheduled. + Step step = active.current; + Class<?> job = active.currentJob; + HeldLocks oldest = active.heldLocks; + int lockCount = oldest.count; + for (PlanningDiagnostics outer = active.parent; outer != null; outer = outer.parent) { + HeldLocks locks = outer.heldLocks; + lockCount += locks.count; + if (locks.count > 0 && (oldest.count == 0 || locks.started < oldest.started)) { + oldest = locks; + } + } + long now = root.clock.getAsLong(); + if (millis(now - root.started) < threshold) { + return; + } + long previous = root.lastReport.get(); + // A minimum interval protects the checker from an accidentally zero/negative dynamic setting. + long interval = Math.max(1000, Config.nereids_planning_log_interval_ms); + if ((previous != Long.MIN_VALUE && millis(now - previous) < interval) + || !root.lastReport.compareAndSet(previous, now)) { + return; + } + JsonObject event = step == null ? new JsonObject() : step.toJson(); + event.addProperty("status", "running"); + event.addProperty("elapsed_ms", millis(now - active.started)); + event.addProperty("root_pass_id", root.passId); + event.addProperty("root_elapsed_ms", millis(now - root.started)); + event.addProperty("phase_elapsed_ms", step == null ? 0 : millis(now - step.phaseStarted)); + event.addProperty("operation_elapsed_ms", step == null || step.operation.isEmpty() + ? 0 : millis(now - step.started)); + event.addProperty("job", job == null ? "" : job.getSimpleName()); + event.addProperty("held_locks", lockCount); + event.addProperty("oldest_lock", lockCount == 0 ? "" : bounded(oldest.oldestTable)); + event.addProperty("oldest_lock_hold_ms", lockCount == 0 ? 0 : millis(now - oldest.started)); + active.log("Slow planning", event); + } + + private void finish(boolean success) { + JsonObject result = new JsonObject(); + result.addProperty("pass_id", passId); + result.addProperty("status", success ? "completed" : "failed"); + result.addProperty("elapsed_ms", millis(clock.getAsLong() - started)); + result.addProperty("failed_step", success ? "" : failedStep); + JsonObject phases = new JsonObject(); + for (Phase phase : Phase.values()) { + Timing timing = timings.get(phase); + JsonObject value = new JsonObject(); + value.addProperty("status", timing == null ? "not_run" : timing.failures > 0 ? "failed" : "completed"); + value.addProperty("elapsed_ms", timing == null ? 0 : millis(timing.nanos)); + value.addProperty("calls", timing == null ? 0 : timing.calls); + phases.add(phase.key(), value); + } + result.add("phases", phases); + result.addProperty("omitted_operations", omittedOperations); + if (summary != null) { + summary.addPlanningPass(result); + } + long threshold = Config.nereids_planning_log_threshold_ms; + boolean report = !success || !slowOperations.isEmpty() + || (threshold > 0 && result.get("elapsed_ms").getAsLong() >= threshold); + // An enclosing MV planning pass may still hold locks. Defer its children's logs as well. + if (parent != null) { + for (Event event : slowOperations) { + parent.defer(event); + } + parent.omittedOperations += omittedOperations; + if (report) { + parent.defer(new Event(this, "Planning finished", result)); + } + } else { + for (Event event : slowOperations) { + event.owner.log(event.message, event.json); + } + if (report) { + log("Planning finished", result); + } + } + } + + private void defer(Event event) { + if (slowOperations.size() == MAX_SLOW_OPERATIONS) { + omittedOperations++; + } else { + slowOperations.add(event); + } + } + + private void log(String message, JsonObject event) { + String formattedId = queryId == null || (queryId.hi == 0 && queryId.lo == 0) + ? "" : DebugUtil.printId(queryId); + event.addProperty("query_id", formattedId); + event.addProperty("pass_id", passId); + event.addProperty("statement_id", statementId); + event.addProperty("planner_thread", threadName); + event.addProperty("query_timeout_ms", timeoutMs); + // Bind only this event; the checker or a deferred event may run under another query's context. + try (CloseableThreadContext.Instance ignored = CloseableThreadContext.put("query_id", formattedId)) { Review Comment: [P2] Coordinate the MDC binding and nested-identity logic with #68694 Problems with the binding here: - The key `"query_id"` is hardcoded. - An empty string is written when there is no id; #68694's `QueryLogContext` removes the key instead. - The `sys_log_enable_query_id` switch introduced by #68694 is not honored. On master no log pattern contains `%X{query_id}`, so if this PR lands alone the binding has no effect. With both PRs merged, the id is printed twice: in the log prefix and in the JSON `query_id` field. Both PRs also implement "nested planning inherits the outer query id" independently (`inheritsQueryIdentity` here, `QueryLogContext.PLANNING_QUERY_ID` there). A `git merge-tree` trial merge shows textual conflicts in `NereidsPlanner.java` (both split `planWithLock` into `planWithLockInternal`) and in `ConnectContext.checkTimeout`. Suggest agreeing on a merge order; whichever PR lands second should reuse `QueryLogContext`'s key and inheritance logic. ########## fe/fe-core/src/main/java/org/apache/doris/common/profile/PlanningDiagnostics.java: ########## @@ -0,0 +1,437 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 org.apache.doris.common.profile; + +import org.apache.doris.common.Config; +import org.apache.doris.common.util.DebugUtil; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.thrift.TUniqueId; + +import com.google.common.annotations.VisibleForTesting; +import com.google.gson.JsonObject; +import org.apache.logging.log4j.CloseableThreadContext; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.ArrayList; +import java.util.EnumMap; +import java.util.List; +import java.util.Locale; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongSupplier; +import java.util.function.Supplier; + +/** + * One planning pass. The planning thread owns timers and the bounded slow-operation buffer. + * The connection checker reads only immutable, volatile-published steps, never planner/table locks. + * Timings are inclusive: nested phases and operations must not be added to their enclosing phase. + */ +public final class PlanningDiagnostics { + private static final Logger LOG = LogManager.getLogger(PlanningDiagnostics.class); + private static final int MAX_SLOW_OPERATIONS = 16; + // Internal MV planning can replace ConnectContext while the calling planner still holds table locks. + private static final ThreadLocal<PlanningDiagnostics> ACTIVE = new ThreadLocal<>(); + + public enum Phase { + PREPROCESS, COLLECT_TABLES, PRELOAD_METADATA, WAIT_CHANGE_VISIBLE, LOCK_TABLES, + ANALYZE, REWRITE, PRE_REWRITE_MV, OPTIMIZE, CHOOSE_PLAN, POST_PROCESS, + TRANSLATE, DISTRIBUTE, RELEASE_RESOURCES; + + public String key() { + return name().toLowerCase(Locale.ROOT); + } + } + + private final ConnectContext context; + private final SummaryProfile summary; + private final long passId; + private final PlanningDiagnostics parent; + private final PlanningDiagnostics root; + private final boolean inheritsQueryIdentity; + private final TUniqueId queryId; + private final long statementId; + private final String threadName; + private volatile long timeoutMs; + private final LongSupplier clock; + private final long started; + private final EnumMap<Phase, Timing> timings = new EnumMap<>(Phase.class); + private final List<Event> slowOperations = new ArrayList<>(); + private final AtomicLong lastReport; + private volatile Step current; + private volatile Class<?> currentJob; + private volatile HeldLocks heldLocks = new HeldLocks(0, 0, ""); + private volatile boolean finished; + // Only the root publishes this pointer, so its registered connection can inspect an internal pass. + private volatile PlanningDiagnostics activePass; + private String failedStep = ""; + private Throwable stepFailure; + private long omittedOperations; + + private PlanningDiagnostics(ConnectContext context) { + this(context, System::nanoTime); + } + + @VisibleForTesting + PlanningDiagnostics(ConnectContext context, LongSupplier clock) { + this.context = context; + this.parent = ACTIVE.get(); + this.root = parent == null ? this : parent.root; + this.lastReport = parent == null ? new AtomicLong(Long.MIN_VALUE) : parent.lastReport; + SummaryProfile contextSummary = SummaryProfile.getSummaryProfile(context); + this.summary = contextSummary != null ? contextSummary : parent == null ? null : parent.summary; + this.passId = summary == null ? 0 : summary.nextPlanningPassId(); + TUniqueId contextQueryId = context.queryId(); + this.inheritsQueryIdentity = contextQueryId == null && parent != null; + this.queryId = inheritsQueryIdentity ? parent.queryId + : contextQueryId == null ? null : contextQueryId.deepCopy(); + this.statementId = inheritsQueryIdentity ? parent.statementId : context.getStmtId(); + this.threadName = Thread.currentThread().getName(); + this.timeoutMs = inheritsQueryIdentity ? parent.timeoutMs : context.getExecTimeoutS() * 1000L; + this.clock = clock; + this.started = clock.getAsLong(); + } + + /** Includes cleanup and restores an outer pass even when internal planning uses another connection. */ + public static <T> T plan(ConnectContext context, Supplier<T> action) { + return execute(new PlanningDiagnostics(context), action); + } + + @VisibleForTesting + static <T> T execute(PlanningDiagnostics diagnostics, Supplier<T> action) { + ConnectContext context = diagnostics.context; + PlanningDiagnostics previous = context.getPlanningDiagnostics(); + ACTIVE.set(diagnostics); + context.setPlanningDiagnostics(diagnostics); + diagnostics.root.activePass = diagnostics; + boolean success = false; + try { + T result = action.get(); + success = true; + return result; + } catch (RuntimeException | Error failure) { + if (diagnostics.stepFailure != failure) { + diagnostics.failedStep = "between_phases"; + } + throw failure; + } finally { + diagnostics.finished = true; + context.setPlanningDiagnostics(previous); + diagnostics.root.activePass = diagnostics.parent; + if (diagnostics.parent == null) { + ACTIVE.remove(); + } else { + ACTIVE.set(diagnostics.parent); + } + diagnostics.finish(success); + } + } + + public static void runPhase(ConnectContext context, Phase phase, Runnable action) { + phase(context, phase, () -> { + action.run(); + return null; + }); + } + + public static <T> T phase(ConnectContext context, Phase phase, Supplier<T> action) { + PlanningDiagnostics diagnostics = current(context); + return diagnostics == null ? action.get() : diagnostics.measure(phase, "", "", -1, action); + } + + /** A negative operation budget means that timeout handling belongs to the connector. */ + public static <T> T operation(ConnectContext context, String operation, String target, + long budgetMs, Supplier<T> action) { + return operation(context, operation, () -> target, budgetMs, action); + } + + /** Resolve the target only for an active planning pass, on the calling thread. */ + public static <T> T operation(ConnectContext context, String operation, Supplier<String> target, + long budgetMs, Supplier<T> action) { + PlanningDiagnostics diagnostics = current(context); + if (diagnostics == null) { + return action.get(); + } + Step step = diagnostics.current; + return diagnostics.measure(step == null ? Phase.PREPROCESS : step.phase, + operation, target.get(), budgetMs, action); + } + + public static PlanningDiagnostics current(ConnectContext context) { + return context == null ? null : context.getPlanningDiagnostics(); + } + + private <T> T measure(Phase phase, String operation, String target, long budgetMs, Supplier<T> action) { + Step previous = current; + Step step = new Step(phase, operation, target, budgetMs, clock.getAsLong(), + operation.isEmpty() || previous == null ? -1 : previous.phaseStarted); + current = step; + boolean success = false; + try { + T result = action.get(); + success = true; + return result; + } catch (RuntimeException | Error failure) { + // Preserve the innermost failing step of this exception, not an earlier recovered MV failure. + if (stepFailure != failure) { + stepFailure = failure; + failedStep = phase.key() + (operation.isEmpty() ? "" : "/" + operation); + } + throw failure; + } finally { + long elapsed = clock.getAsLong() - step.started; + if (operation.isEmpty()) { + Timing timing = timings.computeIfAbsent(phase, key -> new Timing()); + timing.nanos += elapsed; + timing.calls++; + timing.failures += success ? 0 : 1; + if (phase == Phase.PREPROCESS && success) { + // SET_VAR hints are applied by preprocessing, after this pass was created. + timeoutMs = inheritsQueryIdentity ? parent.timeoutMs : context.getExecTimeoutS() * 1000L; + } + } else { + recordOperation(step, elapsed, success); + } + current = previous; + } + } + + /** Avoid allocating task descriptions for every optimizer job; format the class only on a slow report. */ + public Class<?> setCurrentJob(Class<?> job) { + Class<?> previous = currentJob; + currentJob = job; + return previous; + } + + public LockHold acquiredLock(String table) { + long now = clock.getAsLong(); + HeldLocks previous = heldLocks; + heldLocks = new HeldLocks(previous.count + 1, previous.count == 0 ? now : previous.started, + previous.count == 0 ? table : previous.oldestTable); + return new LockHold(table, now); + } + + /** Closed after unlocking; planner resources release in reverse acquisition order. */ + public final class LockHold implements AutoCloseable { + private final String table; + private final long acquired; + + private LockHold(String table, long acquired) { + this.table = table; + this.acquired = acquired; + } + + @Override + public void close() { + HeldLocks previous = heldLocks; + heldLocks = new HeldLocks(previous.count - 1, previous.started, previous.oldestTable); + recordOperation(new Step(Phase.RELEASE_RESOURCES, "read_lock_hold", table, -1, acquired, -1), + clock.getAsLong() - acquired, true); + } + } + + private void recordOperation(Step step, long nanos, boolean success) { + long threshold = Config.nereids_planning_operation_log_threshold_ms; + if (success && (threshold <= 0 || millis(nanos) < threshold)) { + return; + } + if (slowOperations.size() == MAX_SLOW_OPERATIONS) { + omittedOperations++; + return; + } + JsonObject event = step.toJson(); + event.addProperty("elapsed_ms", millis(nanos)); + event.addProperty("status", success ? "completed" : "failed"); + slowOperations.add(new Event(this, "Planning operation", event)); + } + + /** Invoked by the existing connection timeout checker, including while a planner job is blocked. */ + public void reportIfSlow() { + long threshold = Config.nereids_planning_log_threshold_ms; + PlanningDiagnostics active = root.activePass; + if (root.finished || active == null || threshold <= 0) { + return; + } + // Snapshot published starts before reading the clock. A newer step/lock sampled after the clock + // could otherwise appear to have started in the future when the checker was descheduled. + Step step = active.current; + Class<?> job = active.currentJob; + HeldLocks oldest = active.heldLocks; + int lockCount = oldest.count; + for (PlanningDiagnostics outer = active.parent; outer != null; outer = outer.parent) { + HeldLocks locks = outer.heldLocks; + lockCount += locks.count; + if (locks.count > 0 && (oldest.count == 0 || locks.started < oldest.started)) { + oldest = locks; + } + } + long now = root.clock.getAsLong(); + if (millis(now - root.started) < threshold) { + return; + } + long previous = root.lastReport.get(); + // A minimum interval protects the checker from an accidentally zero/negative dynamic setting. + long interval = Math.max(1000, Config.nereids_planning_log_interval_ms); + if ((previous != Long.MIN_VALUE && millis(now - previous) < interval) + || !root.lastReport.compareAndSet(previous, now)) { + return; + } + JsonObject event = step == null ? new JsonObject() : step.toJson(); + event.addProperty("status", "running"); + event.addProperty("elapsed_ms", millis(now - active.started)); + event.addProperty("root_pass_id", root.passId); + event.addProperty("root_elapsed_ms", millis(now - root.started)); + event.addProperty("phase_elapsed_ms", step == null ? 0 : millis(now - step.phaseStarted)); + event.addProperty("operation_elapsed_ms", step == null || step.operation.isEmpty() + ? 0 : millis(now - step.started)); + event.addProperty("job", job == null ? "" : job.getSimpleName()); + event.addProperty("held_locks", lockCount); + event.addProperty("oldest_lock", lockCount == 0 ? "" : bounded(oldest.oldestTable)); + event.addProperty("oldest_lock_hold_ms", lockCount == 0 ? 0 : millis(now - oldest.started)); + active.log("Slow planning", event); + } + + private void finish(boolean success) { Review Comment: [P2] Per-pass JSON construction is paid on every query `finish()` builds about 16 `JsonObject`s for every pass, and `addPlanningPass()` then `deepCopy()`s the result. This happens on every query, even with Profile disabled and nothing to log. With `K + 1` passes per query when MV rewrite plans MV definitions, and with high-QPS `INSERT ... VALUES`, this is avoidable. Consider accumulating per-`Phase` counters in primitive arrays and serializing to JSON only when a report is emitted or the Profile is rendered. ########## fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/scheduler/SimpleJobScheduler.java: ########## @@ -48,7 +49,15 @@ public void executeJobPool(ScheduleContext scheduleContext) { } } Job job = pool.pop(); - job.execute(); + PlanningDiagnostics diagnostics = PlanningDiagnostics.current(context.getConnectContext()); Review Comment: [P3] Hoist the loop-invariant diagnostics lookup `PlanningDiagnostics.current(...)` does not change inside this loop, which is the optimizer's hottest loop. It can be read once above the `while`. ########## fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalTable.java: ########## @@ -182,17 +184,17 @@ public List<Column> getFullSchema() { // NOT getFullSchema(Optional.empty()): an empty snapshot means "this reference has no pin" (=> // latest), whereas the no-arg form means "I have no reference, resolve from the ambient context". // Collapsing the two would strip the ambient resolution from every statement-global caller. - Optional<SchemaCacheValue> schemaCacheValue = getSchemaCacheValue(); - return schemaCacheValue.map(SchemaCacheValue::getSchema).orElse(null); + return PlanningDiagnostics.operation(ConnectContext.get(), "load_schema", this::getNameWithFullQualifiers, -1, Review Comment: [P2] The qualified target name is built on every schema access, not once `getFullSchema()` (and the overload at :196, plus `initSelectedPartitions` at :493) is called many times per query, both directly and through `getColumn(name)` / `getBaseSchema()`. Inside an active pass, every call builds `getNameWithFullQualifiers()`, allocates a `Step`, and reads `nanoTime` twice. The description says targets are resolved once; in practice they are resolved once per call. Consider caching the qualified name or instrumenting at a coarser level, e.g. only the schema-cache load path. ########## fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java: ########## @@ -1593,24 +1611,81 @@ public void setAssignedWeightPerBackend(Map<Backend, Long> assignedWeightPerBack this.assignedWeightPerBackend = assignedWeightPerBackend; } + public synchronized long nextPlanningPassId() { + return ++planningPassSequence; + } + + /** Bounded across retries, CTAS passes and nested MV planning; existing timing fields keep their meaning. */ + public synchronized void addPlanningPass(JsonObject pass) { + planningPassCount++; + planningFailedPassCount += "failed".equals(pass.get("status").getAsString()) ? 1 : 0; + planningElapsedMs += pass.get("elapsed_ms").getAsLong(); + lastPlanningPass = pass.deepCopy(); + for (Phase phase : Phase.values()) { + String key = phase.key(); + long total = planningPhaseTotals.has(key) ? planningPhaseTotals.get(key).getAsLong() : 0; + JsonObject phaseResult = pass.getAsJsonObject("phases").getAsJsonObject(key); + if (!"not_run".equals(phaseResult.get("status").getAsString())) { + planningPhaseTotals.addProperty(key, total + phaseResult.get("elapsed_ms").getAsLong()); + } + if ("failed".equals(phaseResult.get("status").getAsString())) { + long failures = planningPhaseFailures.has(key) ? planningPhaseFailures.get(key).getAsLong() : 0; + planningPhaseFailures.addProperty(key, failures + 1); + } + } + } + + public synchronized JsonObject getPlanningDetails() { + JsonObject result = new JsonObject(); + result.addProperty("passes", planningPassCount); + result.addProperty("failed_passes", planningFailedPassCount); + result.addProperty("elapsed_ms", planningElapsedMs); + result.add("phase_time_ms", planningPhaseTotals.deepCopy()); + result.add("last_pass", lastPlanningPass.deepCopy()); + return result; + } + + /** plan_times_ms is MAP<STRING, INT>. Never add strings or nested objects to this audit field. */ + private synchronized void addPlanningAuditTimes(JsonObject result) { Review Comment: [P3] The plan-timeout error now carries a block of `-1` entries `SimpleJobScheduler` throws `QueryPlanningErrors.planTimeoutError()` while the current pass is still running, and that calls `getPlanTime()`. Because no pass has finished yet, the user-visible "Nereids cost too much time" message gains `planning_passes:0`, `planning_last_failed:-1` and about a dozen `planning_*:-1` entries, which is noise there. Consider omitting the planning keys when `planningPassCount == 0`. ########## fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java: ########## @@ -1326,7 +1327,22 @@ public void cancelQuery(Status cancelReason) { } } + // Published independently of StatementContext's monitor, which is held during table-lock acquisition. + private volatile PlanningDiagnostics planningDiagnostics; + + public PlanningDiagnostics getPlanningDiagnostics() { + return planningDiagnostics; + } + + public void setPlanningDiagnostics(PlanningDiagnostics diagnostics) { + planningDiagnostics = diagnostics; + } + public void checkTimeout(long now) { + PlanningDiagnostics diagnostics = planningDiagnostics; + if (diagnostics != null) { + diagnostics.reportIfSlow(); Review Comment: [P3] `reportIfSlow()` runs before the timeout/kill decision I don't see a throwing path today. If one appears later, `ConnectPoolMgr` catches the exception per connection, and that connection's timeout kill is skipped for the round. Moving the call after the kill logic removes the coupling. ########## fe/fe-core/src/main/java/org/apache/doris/common/profile/PlanningDiagnostics.java: ########## @@ -0,0 +1,437 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 org.apache.doris.common.profile; + +import org.apache.doris.common.Config; +import org.apache.doris.common.util.DebugUtil; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.thrift.TUniqueId; + +import com.google.common.annotations.VisibleForTesting; +import com.google.gson.JsonObject; +import org.apache.logging.log4j.CloseableThreadContext; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +import java.util.ArrayList; +import java.util.EnumMap; +import java.util.List; +import java.util.Locale; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.LongSupplier; +import java.util.function.Supplier; + +/** + * One planning pass. The planning thread owns timers and the bounded slow-operation buffer. + * The connection checker reads only immutable, volatile-published steps, never planner/table locks. + * Timings are inclusive: nested phases and operations must not be added to their enclosing phase. + */ +public final class PlanningDiagnostics { + private static final Logger LOG = LogManager.getLogger(PlanningDiagnostics.class); + private static final int MAX_SLOW_OPERATIONS = 16; + // Internal MV planning can replace ConnectContext while the calling planner still holds table locks. + private static final ThreadLocal<PlanningDiagnostics> ACTIVE = new ThreadLocal<>(); + + public enum Phase { + PREPROCESS, COLLECT_TABLES, PRELOAD_METADATA, WAIT_CHANGE_VISIBLE, LOCK_TABLES, + ANALYZE, REWRITE, PRE_REWRITE_MV, OPTIMIZE, CHOOSE_PLAN, POST_PROCESS, + TRANSLATE, DISTRIBUTE, RELEASE_RESOURCES; + + public String key() { + return name().toLowerCase(Locale.ROOT); + } + } + + private final ConnectContext context; + private final SummaryProfile summary; + private final long passId; + private final PlanningDiagnostics parent; + private final PlanningDiagnostics root; + private final boolean inheritsQueryIdentity; + private final TUniqueId queryId; + private final long statementId; + private final String threadName; + private volatile long timeoutMs; + private final LongSupplier clock; + private final long started; + private final EnumMap<Phase, Timing> timings = new EnumMap<>(Phase.class); + private final List<Event> slowOperations = new ArrayList<>(); + private final AtomicLong lastReport; + private volatile Step current; + private volatile Class<?> currentJob; + private volatile HeldLocks heldLocks = new HeldLocks(0, 0, ""); + private volatile boolean finished; + // Only the root publishes this pointer, so its registered connection can inspect an internal pass. + private volatile PlanningDiagnostics activePass; + private String failedStep = ""; + private Throwable stepFailure; + private long omittedOperations; + + private PlanningDiagnostics(ConnectContext context) { + this(context, System::nanoTime); + } + + @VisibleForTesting + PlanningDiagnostics(ConnectContext context, LongSupplier clock) { + this.context = context; + this.parent = ACTIVE.get(); + this.root = parent == null ? this : parent.root; + this.lastReport = parent == null ? new AtomicLong(Long.MIN_VALUE) : parent.lastReport; + SummaryProfile contextSummary = SummaryProfile.getSummaryProfile(context); + this.summary = contextSummary != null ? contextSummary : parent == null ? null : parent.summary; + this.passId = summary == null ? 0 : summary.nextPlanningPassId(); + TUniqueId contextQueryId = context.queryId(); + this.inheritsQueryIdentity = contextQueryId == null && parent != null; + this.queryId = inheritsQueryIdentity ? parent.queryId + : contextQueryId == null ? null : contextQueryId.deepCopy(); + this.statementId = inheritsQueryIdentity ? parent.statementId : context.getStmtId(); + this.threadName = Thread.currentThread().getName(); + this.timeoutMs = inheritsQueryIdentity ? parent.timeoutMs : context.getExecTimeoutS() * 1000L; + this.clock = clock; + this.started = clock.getAsLong(); + } + + /** Includes cleanup and restores an outer pass even when internal planning uses another connection. */ + public static <T> T plan(ConnectContext context, Supplier<T> action) { + return execute(new PlanningDiagnostics(context), action); + } + + @VisibleForTesting + static <T> T execute(PlanningDiagnostics diagnostics, Supplier<T> action) { + ConnectContext context = diagnostics.context; + PlanningDiagnostics previous = context.getPlanningDiagnostics(); + ACTIVE.set(diagnostics); + context.setPlanningDiagnostics(diagnostics); + diagnostics.root.activePass = diagnostics; + boolean success = false; + try { + T result = action.get(); + success = true; + return result; + } catch (RuntimeException | Error failure) { + if (diagnostics.stepFailure != failure) { + diagnostics.failedStep = "between_phases"; + } + throw failure; + } finally { + diagnostics.finished = true; + context.setPlanningDiagnostics(previous); + diagnostics.root.activePass = diagnostics.parent; + if (diagnostics.parent == null) { + ACTIVE.remove(); + } else { + ACTIVE.set(diagnostics.parent); + } + diagnostics.finish(success); + } + } + + public static void runPhase(ConnectContext context, Phase phase, Runnable action) { + phase(context, phase, () -> { + action.run(); + return null; + }); + } + + public static <T> T phase(ConnectContext context, Phase phase, Supplier<T> action) { + PlanningDiagnostics diagnostics = current(context); + return diagnostics == null ? action.get() : diagnostics.measure(phase, "", "", -1, action); + } + + /** A negative operation budget means that timeout handling belongs to the connector. */ + public static <T> T operation(ConnectContext context, String operation, String target, + long budgetMs, Supplier<T> action) { + return operation(context, operation, () -> target, budgetMs, action); + } + + /** Resolve the target only for an active planning pass, on the calling thread. */ + public static <T> T operation(ConnectContext context, String operation, Supplier<String> target, + long budgetMs, Supplier<T> action) { + PlanningDiagnostics diagnostics = current(context); + if (diagnostics == null) { + return action.get(); + } + Step step = diagnostics.current; + return diagnostics.measure(step == null ? Phase.PREPROCESS : step.phase, + operation, target.get(), budgetMs, action); + } + + public static PlanningDiagnostics current(ConnectContext context) { + return context == null ? null : context.getPlanningDiagnostics(); + } + + private <T> T measure(Phase phase, String operation, String target, long budgetMs, Supplier<T> action) { + Step previous = current; + Step step = new Step(phase, operation, target, budgetMs, clock.getAsLong(), + operation.isEmpty() || previous == null ? -1 : previous.phaseStarted); + current = step; + boolean success = false; + try { + T result = action.get(); + success = true; + return result; + } catch (RuntimeException | Error failure) { + // Preserve the innermost failing step of this exception, not an earlier recovered MV failure. + if (stepFailure != failure) { + stepFailure = failure; + failedStep = phase.key() + (operation.isEmpty() ? "" : "/" + operation); + } + throw failure; + } finally { + long elapsed = clock.getAsLong() - step.started; + if (operation.isEmpty()) { + Timing timing = timings.computeIfAbsent(phase, key -> new Timing()); + timing.nanos += elapsed; + timing.calls++; + timing.failures += success ? 0 : 1; + if (phase == Phase.PREPROCESS && success) { + // SET_VAR hints are applied by preprocessing, after this pass was created. + timeoutMs = inheritsQueryIdentity ? parent.timeoutMs : context.getExecTimeoutS() * 1000L; + } + } else { + recordOperation(step, elapsed, success); + } + current = previous; + } + } + + /** Avoid allocating task descriptions for every optimizer job; format the class only on a slow report. */ + public Class<?> setCurrentJob(Class<?> job) { + Class<?> previous = currentJob; + currentJob = job; + return previous; + } + + public LockHold acquiredLock(String table) { + long now = clock.getAsLong(); + HeldLocks previous = heldLocks; + heldLocks = new HeldLocks(previous.count + 1, previous.count == 0 ? now : previous.started, + previous.count == 0 ? table : previous.oldestTable); + return new LockHold(table, now); + } + + /** Closed after unlocking; planner resources release in reverse acquisition order. */ + public final class LockHold implements AutoCloseable { + private final String table; + private final long acquired; + + private LockHold(String table, long acquired) { + this.table = table; + this.acquired = acquired; + } + + @Override + public void close() { + HeldLocks previous = heldLocks; + heldLocks = new HeldLocks(previous.count - 1, previous.started, previous.oldestTable); + recordOperation(new Step(Phase.RELEASE_RESOURCES, "read_lock_hold", table, -1, acquired, -1), + clock.getAsLong() - acquired, true); + } + } + + private void recordOperation(Step step, long nanos, boolean success) { + long threshold = Config.nereids_planning_operation_log_threshold_ms; + if (success && (threshold <= 0 || millis(nanos) < threshold)) { + return; + } + if (slowOperations.size() == MAX_SLOW_OPERATIONS) { + omittedOperations++; + return; + } + JsonObject event = step.toJson(); + event.addProperty("elapsed_ms", millis(nanos)); + event.addProperty("status", success ? "completed" : "failed"); + slowOperations.add(new Event(this, "Planning operation", event)); + } + + /** Invoked by the existing connection timeout checker, including while a planner job is blocked. */ + public void reportIfSlow() { + long threshold = Config.nereids_planning_log_threshold_ms; + PlanningDiagnostics active = root.activePass; + if (root.finished || active == null || threshold <= 0) { + return; + } + // Snapshot published starts before reading the clock. A newer step/lock sampled after the clock + // could otherwise appear to have started in the future when the checker was descheduled. + Step step = active.current; + Class<?> job = active.currentJob; + HeldLocks oldest = active.heldLocks; + int lockCount = oldest.count; + for (PlanningDiagnostics outer = active.parent; outer != null; outer = outer.parent) { + HeldLocks locks = outer.heldLocks; + lockCount += locks.count; + if (locks.count > 0 && (oldest.count == 0 || locks.started < oldest.started)) { + oldest = locks; + } + } + long now = root.clock.getAsLong(); + if (millis(now - root.started) < threshold) { + return; + } + long previous = root.lastReport.get(); + // A minimum interval protects the checker from an accidentally zero/negative dynamic setting. + long interval = Math.max(1000, Config.nereids_planning_log_interval_ms); + if ((previous != Long.MIN_VALUE && millis(now - previous) < interval) + || !root.lastReport.compareAndSet(previous, now)) { + return; + } + JsonObject event = step == null ? new JsonObject() : step.toJson(); + event.addProperty("status", "running"); + event.addProperty("elapsed_ms", millis(now - active.started)); + event.addProperty("root_pass_id", root.passId); + event.addProperty("root_elapsed_ms", millis(now - root.started)); + event.addProperty("phase_elapsed_ms", step == null ? 0 : millis(now - step.phaseStarted)); + event.addProperty("operation_elapsed_ms", step == null || step.operation.isEmpty() + ? 0 : millis(now - step.started)); + event.addProperty("job", job == null ? "" : job.getSimpleName()); + event.addProperty("held_locks", lockCount); + event.addProperty("oldest_lock", lockCount == 0 ? "" : bounded(oldest.oldestTable)); + event.addProperty("oldest_lock_hold_ms", lockCount == 0 ? 0 : millis(now - oldest.started)); + active.log("Slow planning", event); + } + + private void finish(boolean success) { + JsonObject result = new JsonObject(); + result.addProperty("pass_id", passId); + result.addProperty("status", success ? "completed" : "failed"); + result.addProperty("elapsed_ms", millis(clock.getAsLong() - started)); + result.addProperty("failed_step", success ? "" : failedStep); + JsonObject phases = new JsonObject(); + for (Phase phase : Phase.values()) { + Timing timing = timings.get(phase); + JsonObject value = new JsonObject(); + value.addProperty("status", timing == null ? "not_run" : timing.failures > 0 ? "failed" : "completed"); + value.addProperty("elapsed_ms", timing == null ? 0 : millis(timing.nanos)); + value.addProperty("calls", timing == null ? 0 : timing.calls); + phases.add(phase.key(), value); + } + result.add("phases", phases); + result.addProperty("omitted_operations", omittedOperations); + if (summary != null) { + summary.addPlanningPass(result); + } + long threshold = Config.nereids_planning_log_threshold_ms; + boolean report = !success || !slowOperations.isEmpty() + || (threshold > 0 && result.get("elapsed_ms").getAsLong() >= threshold); + // An enclosing MV planning pass may still hold locks. Defer its children's logs as well. + if (parent != null) { + for (Event event : slowOperations) { + parent.defer(event); + } + parent.omittedOperations += omittedOperations; + if (report) { + parent.defer(new Event(this, "Planning finished", result)); + } + } else { + for (Event event : slowOperations) { + event.owner.log(event.message, event.json); + } + if (report) { + log("Planning finished", result); Review Comment: [P3] The failed-pass summary WARN cannot be disabled and duplicates an existing WARN A failed pass always emits "Planning finished" with a full JSON summary, independent of the thresholds. `StmtExecutor` already logs "Nereids plan query failed" for the same failure, so frequent user errors such as "Unknown column" add another sizeable WARN line each time. Consider gating it behind the threshold, or logging only the failed step. ########## fe/fe-common/src/main/java/org/apache/doris/common/Config.java: ########## @@ -104,6 +104,21 @@ public class Config extends ConfigBase { @ConfField(mutable = true, description = "The threshold of slow query, in milliseconds. If the response time of a " + "query exceeds this threshold, it will be recorded in audit log.") public static long qe_slow_log_ms = 5000; + + @ConfField(mutable = true, description = "Log slow Nereids planning passes after this many milliseconds. " + + "Non-positive values disable slow-pass logs; failed passes are still reported.") + public static volatile long nereids_planning_log_threshold_ms = 5000; Review Comment: [P3] `volatile` is unusual for `Config` fields Other mutable configs in this file are plain `public static` fields. Please keep the new ones consistent, and add a docs PR for the three settings. ########## fe/fe-core/src/test/java/org/apache/doris/common/profile/PlanningDiagnosticsTest.java: ########## @@ -0,0 +1,988 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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 org.apache.doris.common.profile; + +import org.apache.doris.catalog.Column; +import org.apache.doris.catalog.TableIf; +import org.apache.doris.common.Config; +import org.apache.doris.common.FeConstants; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.common.profile.PlanningDiagnostics.Phase; +import org.apache.doris.datasource.ExternalCatalog; +import org.apache.doris.datasource.ExternalDatabase; +import org.apache.doris.datasource.ExternalTable; +import org.apache.doris.datasource.SchemaCacheValue; +import org.apache.doris.datasource.mvcc.MvccSnapshot; +import org.apache.doris.datasource.mvcc.PluginDrivenMvccExternalTable; +import org.apache.doris.mysql.authenticate.TestLogAppender; +import org.apache.doris.nereids.NereidsPlanner; +import org.apache.doris.nereids.StatementContext; +import org.apache.doris.nereids.trees.plans.commands.ExplainCommand.ExplainLevel; +import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan.SelectedPartitions; +import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; +import org.apache.doris.qe.ConnectContext; +import org.apache.doris.qe.OriginStatement; +import org.apache.doris.qe.StmtExecutor; + +import com.google.common.collect.ImmutableList; +import com.google.gson.JsonObject; +import com.google.gson.JsonParser; +import org.apache.logging.log4j.Level; +import org.apache.logging.log4j.ThreadContext; +import org.apache.logging.log4j.core.LogEvent; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; + +public class PlanningDiagnosticsTest { + private static final String QUERY_ID_KEY = "query_id"; + private final AtomicLong clock = new AtomicLong(TimeUnit.SECONDS.toNanos(1)); + private ConnectContext context; + private ConnectContext previousContext; + private SummaryProfile summary; + private long savedThreshold; + private long savedOperationThreshold; + private long savedInterval; + private boolean savedUnitTest; + private String savedQueryId; + + @BeforeEach + public void setUp() { + savedThreshold = Config.nereids_planning_log_threshold_ms; + savedOperationThreshold = Config.nereids_planning_operation_log_threshold_ms; + savedInterval = Config.nereids_planning_log_interval_ms; + savedUnitTest = FeConstants.runningUnitTest; + previousContext = ConnectContext.get(); + savedQueryId = ThreadContext.get(QUERY_ID_KEY); + Config.nereids_planning_log_threshold_ms = 5000; + Config.nereids_planning_operation_log_threshold_ms = 1000; + Config.nereids_planning_log_interval_ms = 30000; + FeConstants.runningUnitTest = true; + context = new ConnectContext(); + context.setQueryId(new org.apache.doris.thrift.TUniqueId(1, 2)); + context.setStmtId(7); + context.setThreadLocalInfo(); + summary = new SummaryProfile(); + StmtExecutor executor = Mockito.mock(StmtExecutor.class); + Mockito.when(executor.getSummaryProfile()).thenReturn(summary); + context.setExecutor(executor); + } + + @AfterEach + public void tearDown() { + ConnectContext.remove(); + if (previousContext != null) { + previousContext.setThreadLocalInfo(); + } + if (savedQueryId == null) { + ThreadContext.remove(QUERY_ID_KEY); + } else { + ThreadContext.put(QUERY_ID_KEY, savedQueryId); + } + Config.nereids_planning_log_threshold_ms = savedThreshold; + Config.nereids_planning_operation_log_threshold_ms = savedOperationThreshold; + Config.nereids_planning_log_interval_ms = savedInterval; + FeConstants.runningUnitTest = savedUnitTest; + } + + @Test + public void testFastPassIsQuietAndAuditContainsTimingsWithoutProfile() { + context.getSessionVariable().enableProfile = false; + summary.addNereidsPreloadExternalMetadataTime(12); + summary.setNereidsCollectTablePartitionFinishTime(100); + summary.setNereidsPreRewriteByMvFinishTime(108); + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + run(() -> { + PlanningDiagnostics.runPhase(context, Phase.CHOOSE_PLAN, () -> advance(7)); + PlanningDiagnostics.runPhase(context, Phase.POST_PROCESS, () -> advance(11)); + }); + Assertions.assertTrue(events(appender).isEmpty()); + } + JsonObject audit = JsonParser.parseString(summary.getPlanTime()).getAsJsonObject(); + Assertions.assertEquals(12, audit.get("preload_external_metadata").getAsLong()); + Assertions.assertEquals(8, audit.get("pre_rewrite_mv").getAsLong()); + Assertions.assertEquals(7, audit.get("planning_choose_plan").getAsLong()); + Assertions.assertEquals(-1, audit.get("planning_analyze").getAsLong()); + audit.entrySet().forEach(entry -> { + Assertions.assertTrue(entry.getValue().isJsonPrimitive()); + Assertions.assertTrue(entry.getValue().getAsJsonPrimitive().isNumber()); + Assertions.assertEquals(entry.getValue().getAsLong(), entry.getValue().getAsInt()); + }); + Assertions.assertEquals(11, phase("post_process").get("elapsed_ms").getAsLong()); + Assertions.assertEquals("not_run", phase("analyze").get("status").getAsString()); + Assertions.assertNull(context.getPlanningDiagnostics()); + } + + @Test + public void testRunningReportRateLimitIdentityAndContextRestoration() { + PlanningDiagnostics diagnostics = new PlanningDiagnostics(context, clock::get); + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + PlanningDiagnostics.execute(diagnostics, () -> { + PlanningDiagnostics.runPhase(context, Phase.OPTIMIZE, () -> { + diagnostics.setCurrentJob(PlanningDiagnosticsTest.class); + advance(5000); + context.queryId().setLo(99); + ThreadContext.put(QUERY_ID_KEY, "checker"); + context.checkTimeout(0); + Assertions.assertEquals("checker", ThreadContext.get(QUERY_ID_KEY)); + context.checkTimeout(0); + Assertions.assertEquals(1, events(appender).size()); + advance(30000); + context.checkTimeout(0); + Assertions.assertEquals(2, events(appender).size()); + assertDiagnosticIdentity(events(appender).get(0), "1-2"); + Assertions.assertTrue(appender.contains(Level.WARN, "\"job\":\"PlanningDiagnosticsTest\"")); + Assertions.assertTrue(appender.contains(Level.WARN, "\"phase\":\"optimize\"")); + }); + return null; + }); + int count = events(appender).size(); + diagnostics.reportIfSlow(); + Assertions.assertEquals(count, events(appender).size()); + } + } + + @Test + public void testMissingAndZeroQueryIdsClearEventIdentityAndRestoreCaller() { + for (org.apache.doris.thrift.TUniqueId queryId : new org.apache.doris.thrift.TUniqueId[] { + null, new org.apache.doris.thrift.TUniqueId(0, 0)}) { + context.setQueryId(queryId); + ThreadContext.put(QUERY_ID_KEY, "caller"); + IllegalStateException failure = new IllegalStateException("planning failure without query identity"); + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + Assertions.assertSame(failure, Assertions.assertThrows(IllegalStateException.class, () -> run(() -> { + Assertions.assertEquals("caller", ThreadContext.get(QUERY_ID_KEY)); + throw failure; + }))); + Assertions.assertEquals(1, events(appender).size()); + assertDiagnosticIdentity(events(appender).get(0), ""); + Assertions.assertEquals("caller", ThreadContext.get(QUERY_ID_KEY)); + Assertions.assertNull(context.getPlanningDiagnostics()); + } + } + } + + @Test + public void testNestedPassWithOwnIdKeepsSnapshotForCheckerAndDeferredEvents() { + ConnectContext child = new ConnectContext(); + child.setQueryId(new org.apache.doris.thrift.TUniqueId(3, 4)); + ThreadContext.put(QUERY_ID_KEY, "caller"); + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + run(() -> PlanningDiagnostics.runPhase(context, Phase.REWRITE, () -> { + PlanningDiagnostics parent = context.getPlanningDiagnostics(); + PlanningDiagnostics.execute(new PlanningDiagnostics(child, clock::get), () -> { + PlanningDiagnostics.runPhase(child, Phase.OPTIMIZE, () -> { + child.queryId().setLo(99); + advance(5000); + context.checkTimeout(0); + Assertions.assertEquals(1, events(appender).size()); + assertDiagnosticIdentity(events(appender).get(0), "3-4"); + Assertions.assertEquals("caller", ThreadContext.get(QUERY_ID_KEY)); + }); + return null; + }); + Assertions.assertSame(parent, context.getPlanningDiagnostics()); + Assertions.assertNull(child.getPlanningDiagnostics()); + Assertions.assertEquals(1, events(appender).size(), "The child completion remains deferred"); + advance(30000); + context.checkTimeout(0); + Assertions.assertEquals(2, events(appender).size()); + assertDiagnosticIdentity(events(appender).get(1), "1-2"); + Assertions.assertEquals("caller", ThreadContext.get(QUERY_ID_KEY)); + })); + Assertions.assertEquals(4, events(appender).size()); + assertDiagnosticIdentity(events(appender).get(2), "3-4"); + assertDiagnosticIdentity(events(appender).get(3), "1-2"); + Assertions.assertEquals(2, summary.getPlanningDetails().get("passes").getAsInt()); + Assertions.assertEquals("caller", ThreadContext.get(QUERY_ID_KEY)); + } + } + + @Test + public void testDynamicDisableAndMinimumInterval() { + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + run(() -> PlanningDiagnostics.runPhase(context, Phase.ANALYZE, () -> { + advance(5000); + Config.nereids_planning_log_threshold_ms = 0; + context.checkTimeout(0); + Assertions.assertTrue(events(appender).isEmpty()); + Config.nereids_planning_log_threshold_ms = 1; + Config.nereids_planning_log_interval_ms = 0; + context.checkTimeout(0); + advance(999); + context.checkTimeout(0); + Assertions.assertEquals(1, events(appender).size()); + advance(1); + context.checkTimeout(0); + Assertions.assertEquals(2, events(appender).size()); + })); + } + } + + @Test + public void testFailurePreservesElapsedAndSkipsLaterStages() { + IllegalStateException expected = new IllegalStateException("original failure"); + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + Assertions.assertSame(expected, Assertions.assertThrows(IllegalStateException.class, + () -> run(() -> PlanningDiagnostics.runPhase(context, Phase.ANALYZE, () -> { + advance(3); + throw expected; + })))); + Assertions.assertTrue(appender.contains(Level.WARN, "\"failed_step\":\"analyze\"")); + } + Assertions.assertEquals(3, phase("analyze").get("elapsed_ms").getAsLong()); + Assertions.assertEquals("failed", phase("analyze").get("status").getAsString()); + Assertions.assertEquals("not_run", phase("optimize").get("status").getAsString()); + Assertions.assertNull(context.getPlanningDiagnostics()); + } + + @Test + public void testNestedAndRepeatedPlanningRestoreParentAndAccumulate() { + run(() -> PlanningDiagnostics.runPhase(context, Phase.REWRITE, () -> { + PlanningDiagnostics parent = context.getPlanningDiagnostics(); + run(() -> PlanningDiagnostics.runPhase(context, Phase.CHOOSE_PLAN, () -> advance(5))); + Assertions.assertSame(parent, context.getPlanningDiagnostics()); + advance(2); + })); + run(() -> PlanningDiagnostics.runPhase(context, Phase.CHOOSE_PLAN, () -> advance(8))); + Assertions.assertEquals(3, summary.getPlanningDetails().get("passes").getAsInt()); + Assertions.assertEquals(13, summary.getPlanningDetails().getAsJsonObject("phase_time_ms") + .get("choose_plan").getAsLong()); + Assertions.assertEquals("not_run", phase("rewrite").get("status").getAsString()); + Assertions.assertNull(context.getPlanningDiagnostics()); + } + + @Test + public void testNestedSlowOperationsAreDeferredUntilOuterPassEnds() { + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + run(() -> { + PlanningDiagnostics.LockHold hold = context.getPlanningDiagnostics().acquiredLock("catalog.db.outer"); + run(() -> PlanningDiagnostics.operation(context, "load_schema", "catalog.db.inner", -1, () -> { + advance(1000); + return null; + })); + Assertions.assertTrue(events(appender).isEmpty()); + hold.close(); + }); + Assertions.assertTrue(appender.contains(Level.WARN, "catalog.db.inner")); + Assertions.assertTrue(appender.contains(Level.WARN, "catalog.db.outer")); + } + } + + @Test + public void testInternalConnectionPlanningDefersLogsAndKeepsQueryIdentity() { + StatementContext outerStatement = new StatementContext(context, new OriginStatement("select 1", 0)); + TableIf table = table(1, "catalog.db.outer"); + outerStatement.getTables().put(ImmutableList.of("outer"), table); + Mockito.when(table.tryReadLock(1, TimeUnit.MINUTES)).thenReturn(true); + // MV cache construction installs a separate connection without an executor or query ID. + ConnectContext internal = new ConnectContext(); + StatementContext internalStatement = new StatementContext(internal, new OriginStatement("select 1", 0)); + NereidsPlanner planner = new NereidsPlanner(internalStatement) { + @Override + protected LogicalPlan preprocess(LogicalPlan plan) { + throw new IllegalArgumentException("internal planning failure"); + } + }; + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + PlanningDiagnostics.plan(context, () -> { + outerStatement.lock(); + try { + internal.setThreadLocalInfo(); + Assertions.assertThrows(IllegalArgumentException.class, () -> planner.planWithLock( + Mockito.mock(LogicalPlan.class), org.apache.doris.nereids.properties.PhysicalProperties.ANY, + ExplainLevel.NONE, false)); + Assertions.assertTrue(events(appender).isEmpty(), "Outer table locks are still held"); + Assertions.assertNull(internal.getPlanningDiagnostics()); + } finally { + context.setThreadLocalInfo(); + outerStatement.releasePlannerResources(); + } + return null; + }); + Mockito.verify(table).readUnlock(); + Assertions.assertTrue(appender.contains(Level.WARN, "\"failed_step\":\"preprocess\"")); + Assertions.assertEquals(2, summary.getPlanningDetails().get("passes").getAsInt()); + Assertions.assertEquals(1, summary.getPlanningDetails().get("failed_passes").getAsInt()); + events(appender).forEach(event -> assertDiagnosticIdentity(event, "1-2")); + } + Assertions.assertNull(context.getPlanningDiagnostics()); + // Reusing the same planner thread must not retain the preceding root or its summary/ID. + internal.setThreadLocalInfo(); + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + Assertions.assertThrows(IllegalArgumentException.class, () -> PlanningDiagnostics.plan(internal, () -> { + throw new IllegalArgumentException("independent planning failure"); + })); + Assertions.assertEquals(1, events(appender).size()); + assertDiagnosticIdentity(events(appender).get(0), ""); + } + Assertions.assertEquals(2, summary.getPlanningDetails().get("passes").getAsInt()); + } + + @Test + public void testRunningReportFollowsInternalPassAndSharesRootRateLimit() { + ConnectContext internal = new ConnectContext(); + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + run(() -> PlanningDiagnostics.runPhase(context, Phase.REWRITE, () -> { + PlanningDiagnostics.LockHold hold = context.getPlanningDiagnostics().acquiredLock("catalog.db.outer"); + try { + advance(5000); + for (int i = 0; i < 2; i++) { + PlanningDiagnostics.execute(new PlanningDiagnostics(internal, clock::get), () -> + PlanningDiagnostics.operation(internal, "load_schema", "catalog.db.inner", -1, () -> { + advance(1); + context.checkTimeout(0); + return null; + })); + } + Assertions.assertEquals(1, events(appender).size(), "Nested passes share the root interval"); + JsonObject event = JsonParser.parseString(events(appender).get(0).getMessage().getFormattedMessage() + .split(": ", 2)[1]).getAsJsonObject(); + Assertions.assertEquals("catalog.db.inner", event.get("target").getAsString()); + Assertions.assertEquals(1, event.get("held_locks").getAsInt()); + Assertions.assertEquals("catalog.db.outer", event.get("oldest_lock").getAsString()); + Assertions.assertEquals(5001, event.get("root_elapsed_ms").getAsLong()); + Assertions.assertEquals(1, event.get("operation_elapsed_ms").getAsLong()); + advance(30000); + context.checkTimeout(0); + Assertions.assertEquals(2, events(appender).size()); + } finally { + hold.close(); + } + })); + } + } + + @Test + public void testRunningReportDoesNotUseStartsPublishedAfterItsTimestamp() throws Exception { + Thread plannerThread = Thread.currentThread(); + CountDownLatch clockRead = new CountDownLatch(1); + CountDownLatch resumeChecker = new CountDownLatch(1); + PlanningDiagnostics diagnostics = new PlanningDiagnostics(context, () -> { + long now = clock.get(); + if (Thread.currentThread() != plannerThread) { + clockRead.countDown(); + try { + Assertions.assertTrue(resumeChecker.await(10, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + return now; + }); + ExecutorService checker = Executors.newSingleThreadExecutor(); + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + PlanningDiagnostics.execute(diagnostics, () -> { + advance(5000); + PlanningDiagnostics.runPhase(context, Phase.REWRITE, () -> { + CompletableFuture<Void> report = CompletableFuture.runAsync(diagnostics::reportIfSlow, checker); + try { + Assertions.assertTrue(clockRead.await(10, TimeUnit.SECONDS)); + advance(10); + PlanningDiagnostics.operation(context, "load_schema", "catalog.db.next", -1, () -> { + PlanningDiagnostics.LockHold hold = diagnostics.acquiredLock("catalog.db.next"); + try { + resumeChecker.countDown(); + report.get(10, TimeUnit.SECONDS); + JsonObject event = JsonParser.parseString(events(appender).get(0).getMessage() + .getFormattedMessage().split(": ", 2)[1]).getAsJsonObject(); + for (String key : new String[] {"phase_elapsed_ms", "operation_elapsed_ms", + "oldest_lock_hold_ms"}) { + Assertions.assertTrue(event.get(key).getAsLong() >= 0, key); + } + } catch (Exception e) { + throw new AssertionError(e); + } finally { + hold.close(); + } + return null; + }); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } finally { + resumeChecker.countDown(); + } + }); + return null; + }); + } finally { + resumeChecker.countDown(); + checker.shutdownNow(); + Assertions.assertTrue(checker.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + public void testRecoveredOperationDoesNotReplaceLaterFailure() { + Assertions.assertThrows(IllegalStateException.class, () -> run(() -> { + try { + PlanningDiagnostics.operation(context, "load_schema", "catalog.db.table", -1, () -> { + throw new IllegalArgumentException("recovered"); + }); + } catch (IllegalArgumentException expected) { + // Materialized-view alternatives can fail while planning continues. + } + PlanningDiagnostics.runPhase(context, Phase.REWRITE, () -> { + throw new IllegalStateException("actual failure"); + }); + })); + Assertions.assertEquals("rewrite", summary.getPlanningDetails().getAsJsonObject("last_pass") + .get("failed_step").getAsString()); + } + + @Test + public void testOperationsAreBoundedDeferredAndCarryQueryIdentity() { + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + run(() -> { + for (int i = 0; i < 20; i++) { + PlanningDiagnostics.operation(context, "load_schema", "catalog.db.table", -1, () -> { + advance(1000); + return null; + }); + } + Assertions.assertTrue(events(appender).isEmpty()); + }); + Assertions.assertEquals(17, events(appender).size()); + events(appender).forEach(event -> assertDiagnosticIdentity(event, "1-2")); + Assertions.assertTrue(appender.contains(Level.WARN, "\"omitted_operations\":4")); + } + } + + @Test + public void testCheckerDoesNotWaitForStatementMonitorAndShowsHeldLock() throws Exception { + StatementContext statement = new StatementContext(context, new OriginStatement("select 1", 0)); + TableIf first = table(1, "catalog.db.first"); + TableIf second = table(2, "catalog.db.second"); + statement.getTables().put(ImmutableList.of("first"), first); + statement.getTables().put(ImmutableList.of("second"), second); + CountDownLatch waiting = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + Mockito.when(first.tryReadLock(1, TimeUnit.MINUTES)).thenReturn(true); + Mockito.when(second.tryReadLock(1, TimeUnit.MINUTES)).thenAnswer(invocation -> { + waiting.countDown(); + Assertions.assertTrue(release.await(10, TimeUnit.SECONDS)); + return true; + }); + ExecutorService worker = Executors.newSingleThreadExecutor(); + try (TestLogAppender appender = TestLogAppender.attach(PlanningDiagnostics.class, Level.WARN)) { + CompletableFuture<Void> planning = CompletableFuture.runAsync(() -> run(() -> { + try { + PlanningDiagnostics.runPhase(context, Phase.LOCK_TABLES, statement::lock); + advance(2000); + } finally { + statement.releasePlannerResources(); + } + }), worker); + try { + Assertions.assertTrue(waiting.await(10, TimeUnit.SECONDS)); + advance(5000); + CompletableFuture.runAsync(() -> context.checkTimeout(0)).get(1, TimeUnit.SECONDS); Review Comment: [P3] Test depends on the common pool and a 1 s timeout `CompletableFuture.runAsync(...)` on the common pool with `.get(1, TimeUnit.SECONDS)` can time out on a loaded CI machine. The other cases use an injected clock/executor and latches; this one could follow the same pattern. ########## fe/fe-core/src/main/java/org/apache/doris/common/profile/SummaryProfile.java: ########## @@ -818,6 +835,7 @@ private void updateExecutionSummaryProfile() { executionSummaryProfile.addInfoString(NEREIDS_PRE_REWRITE_BY_MV_TIME, getPrettyNereidsPreRewriteByMvTime()); executionSummaryProfile.addInfoString(NEREIDS_OPTIMIZE_TIME, getPrettyNereidsOptimizeTime()); executionSummaryProfile.addInfoString(NEREIDS_TRANSLATE_TIME, getPrettyNereidsTranslateTime()); + executionSummaryProfile.addInfoString("Nereids Planning Details", getPlanningDetails().toString()); Review Comment: [P3] Register the new Profile key like the other Nereids entries `"Nereids Planning Details"` is a string literal and is not in `EXECUTION_SUMMARY_KEYS` or the indentation map, so it renders at the end of the execution summary without indentation. Please follow the existing `NEREIDS_*` constant pattern. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
