wenzhenghu commented on code in PR #68727: URL: https://github.com/apache/doris/pull/68727#discussion_r4267301123
########## 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: Addressed in [5d0a0779ce8d](https://github.com/apache/doris/commit/5d0a0779ce8db4a5f372cff13e1e07ba0d7b530a). Completed passes retain a lightweight `PassSnapshot`; phase aggregates use primitive counters. Fast successful completion does not construct the full phase JSON or deep-copy it. JSON is produced for emitted events and Profile rendering/serialization. The adapters preserve the earlier object-shaped persisted Profile format, including empty legacy last-pass values. Legacy-read and snapshot-independence tests passed. I have not measured production allocation/CPU savings, so no quantitative performance claim is made. ########## 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: Addressed in [5d0a0779ce8d](https://github.com/apache/doris/commit/5d0a0779ce8db4a5f372cff13e1e07ba0d7b530a) by caching the qualified target name within each active pass, keyed by table identity. Repeated schema/partition instrumentation reuses that name. Without active diagnostics, it does not resolve the target. The underlying schema/partition calls remain inside their diagnostic scopes; this change caches the display name rather than skipping observation of those operations. Tests cover target capture/cache behavior and no-diagnostics resolution. -- 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]
