wenzhenghu commented on code in PR #68727:
URL: https://github.com/apache/doris/pull/68727#discussion_r4267307918
##########
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:
I kept `volatile` deliberately for these three dynamically updated settings.
They are read by planning threads and the background connection checker without
a shared lock with the configuration writer; `volatile` supplies the visibility
contract for those reads. Matching the plain-field style would remove that
guarantee, so I did not adopt that stylistic change.
Defaults remain `5000 / 1000 / 30000` ms. The PR description and field
descriptions explain the triggers, disabling semantics, and tuning guidance; a
separate configuration-reference documentation PR is still pending.
##########
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:
Addressed in
[5d0a0779ce8d](https://github.com/apache/doris/commit/5d0a0779ce8db4a5f372cff13e1e07ba0d7b530a).
The lock/checker test now uses a dedicated single-thread checker executor,
latch coordination, a 10-second wait, and executor cleanup rather than the
common pool with a 1-second bound.
The latch still ensures the planner holds the statement monitor while the
checker must complete independently. The diagnostic test class passed all 33
cases in the independent FE run.
--
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]
