diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java index 647391dffbc422..235ba022263062 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java @@ -61,6 +61,9 @@ public AbstractJobProcessor(CoordinatorContext coordinatorContext) { protected abstract void doProcessReportExecStatus( TReportExecStatusParams params, SingleFragmentPipelineTask fragmentTask); + // Publish diagnostics that must be visible before an error status releases waiters. + protected void publishReportDiagnosticsBeforeStatus(TReportExecStatusParams params) {} + @Override public final void setPipelineExecutionTask(PipelineExecutionTask pipelineExecutionTask) { Preconditions.checkArgument(pipelineExecutionTask != null, "sqlPipelineTask can not be null"); @@ -134,6 +137,7 @@ public final boolean updateFragmentExecStatus(TReportExecStatusParams params) { DebugUtil.printId(queryId), params.getFragmentId(), DebugUtil.printId(params.getFragmentInstanceId()), params.getBackendId(), status.toString()); + publishReportDiagnosticsBeforeStatus(params); coordinatorContext.updateStatusIfOk(status); } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java index b3005119418039..5b3df949162f8e 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java @@ -2788,6 +2788,15 @@ public boolean updateFragmentExecStatus(TReportExecStatusParams params) { boolean accepted = false; try { Status status = new Status(params.status); + // Publish load diagnostics before an error status can release the completion latch. + if (params.isSetTrackingUrl()) { + LOG.info("query_id={} tracking_url: {}", DebugUtil.printId(queryId), params.getTrackingUrl()); + trackingUrl = params.getTrackingUrl(); + } + if (params.isSetFirstErrorMsg()) { + LOG.info("query_id={} first_error_msg: {}", DebugUtil.printId(queryId), params.getFirstErrorMsg()); + firstErrorMsg = params.getFirstErrorMsg(); + } // for now, abort the query if we see any error except if the error is cancelled // and returned_all_results_ is true. // (UpdateStatus() initiates cancellation, if it hasn't already been initiated) @@ -2814,14 +2823,6 @@ public boolean updateFragmentExecStatus(TReportExecStatusParams params) { if (params.isSetLoadCounters() && loadCounters != null) { updateLoadCounters(params.getLoadCounters()); } - if (params.isSetTrackingUrl()) { - LOG.info("query_id={} tracking_url: {}", DebugUtil.printId(queryId), params.getTrackingUrl()); - trackingUrl = params.getTrackingUrl(); - } - if (params.isSetFirstErrorMsg()) { - LOG.info("query_id={} first_error_msg: {}", DebugUtil.printId(queryId), params.getFirstErrorMsg()); - firstErrorMsg = params.getFirstErrorMsg(); - } // Keep this report's identity local so another report cannot redirect its commit data. long reportTxnId = params.isSetTxnId() ? params.getTxnId() : txnId; if (params.isSetTxnId()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java index d4878ff99a3d6c..01f590ce431a42 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java @@ -159,6 +159,19 @@ public boolean await(long timeout, TimeUnit unit) throws InterruptedException { return latch.get().await(timeout, unit); } + @Override + protected void publishReportDiagnosticsBeforeStatus(TReportExecStatusParams params) { + updateLoadDiagnostics(params); + } + + private void updateLoadDiagnostics(TReportExecStatusParams params) { + if (params.isSetTrackingUrl()) { + loadContext.updateTrackingUrl(params.getTrackingUrl()); + } + if (params.isSetFirstErrorMsg()) { + loadContext.updateFirstErrorMsg(params.getFirstErrorMsg()); + } + } @Override protected void doProcessReportExecStatus(TReportExecStatusParams params, SingleFragmentPipelineTask fragmentTask) { @@ -224,12 +237,7 @@ private void acceptFinalReport(TReportExecStatusParams params) { if (params.isSetLoadCounters()) { loadContext.updateLoadCounters(params.getLoadCounters()); } - if (params.isSetTrackingUrl()) { - loadContext.updateTrackingUrl(params.getTrackingUrl()); - } - if (params.isSetFirstErrorMsg()) { - loadContext.updateFirstErrorMsg(params.getFirstErrorMsg()); - } + updateLoadDiagnostics(params); if (params.isSetTxnId()) { loadContext.updateTransactionId(params.getTxnId()); } diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorLoadDiagnosticsTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorLoadDiagnosticsTest.java new file mode 100644 index 00000000000000..bbf924f506f5b9 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/CoordinatorLoadDiagnosticsTest.java @@ -0,0 +1,116 @@ +// 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.qe; + +import org.apache.doris.analysis.DescriptorTable; +import org.apache.doris.common.MarkedCountDownLatch; +import org.apache.doris.common.Pair; +import org.apache.doris.common.Status; +import org.apache.doris.planner.PlanFragment; +import org.apache.doris.planner.PlanFragmentId; +import org.apache.doris.thrift.TReportExecStatusParams; +import org.apache.doris.thrift.TStatus; +import org.apache.doris.thrift.TStatusCode; +import org.apache.doris.thrift.TUniqueId; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + +class CoordinatorLoadDiagnosticsTest { + private static final int FRAGMENT_ID = 7; + private static final long BACKEND_ID = 9; + private static final String TRACKING_URL = "http://127.0.0.1/error-log"; + private static final String FIRST_ERROR_MSG = "invalid integer. Src line: invalid"; + + @Test + void publishesDiagnosticsBeforeFailureReleasesWaiters() throws Exception { + AtomicBoolean cancelled = new AtomicBoolean(); + Coordinator coordinator = new Coordinator(-1L, new TUniqueId(12345, 6), new DescriptorTable(), + Collections.singletonList(fragment()), Collections.emptyList(), "UTC", false, false) { + @Override + protected void cancelInternal(Status cancelReason) { + super.cancelInternal(cancelReason); + // Inspect the result at the exact point where a load task can return from join(), + // before the report handler resumes. No scheduling delay is needed to expose the race. + Assertions.assertTrue(join(1)); + Assertions.assertEquals(TStatusCode.DATA_QUALITY_ERROR, getExecStatus().getErrorCode()); + Assertions.assertEquals(TRACKING_URL, getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR_MSG, getFirstErrorMsg()); + cancelled.set(true); + } + }; + prepareReport(coordinator); + + Assertions.assertTrue(coordinator.updateFragmentExecStatus(report(TStatusCode.DATA_QUALITY_ERROR))); + + Assertions.assertTrue(cancelled.get()); + } + + @Test + void retainsDiagnosticsOnNormalCompletion() throws Exception { + Coordinator coordinator = new Coordinator(-1L, new TUniqueId(12345, 7), new DescriptorTable(), + Collections.singletonList(fragment()), Collections.emptyList(), "UTC", false, false); + prepareReport(coordinator); + + Assertions.assertTrue(coordinator.updateFragmentExecStatus(report(TStatusCode.OK))); + + Assertions.assertTrue(coordinator.join(1)); + Assertions.assertTrue(coordinator.getExecStatus().ok()); + Assertions.assertEquals(TRACKING_URL, coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR_MSG, coordinator.getFirstErrorMsg()); + } + + private static PlanFragment fragment() { + PlanFragment fragment = Mockito.mock(PlanFragment.class); + Mockito.when(fragment.getFragmentId()).thenReturn(new PlanFragmentId(FRAGMENT_ID)); + return fragment; + } + + private static TReportExecStatusParams report(TStatusCode statusCode) { + return new TReportExecStatusParams() + .setFragmentId(FRAGMENT_ID) + .setBackendId(BACKEND_ID) + .setDone(true) + .setStatus(new TStatus(statusCode)) + .setTrackingUrl(TRACKING_URL) + .setFirstErrorMsg(FIRST_ERROR_MSG); + } + + @SuppressWarnings("unchecked") + private static void prepareReport(Coordinator coordinator) throws Exception { + Coordinator.PipelineExecContext context = Mockito.mock(Coordinator.PipelineExecContext.class); + Mockito.when(context.updatePipelineStatus(Mockito.any())).thenReturn(true); + Field contextsField = Coordinator.class.getDeclaredField("pipelineExecContexts"); + contextsField.setAccessible(true); + Map, Coordinator.PipelineExecContext> contexts = + (Map, Coordinator.PipelineExecContext>) contextsField.get(coordinator); + contexts.put(Pair.of(FRAGMENT_ID, BACKEND_ID), context); + + MarkedCountDownLatch latch = new MarkedCountDownLatch<>(1); + latch.addMark(FRAGMENT_ID, BACKEND_ID); + Field latchField = Coordinator.class.getDeclaredField("fragmentsDoneLatch"); + latchField.setAccessible(true); + latchField.set(coordinator, latch); + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/LoadProcessorTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/LoadProcessorTest.java new file mode 100644 index 00000000000000..3839270ee5f431 --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/LoadProcessorTest.java @@ -0,0 +1,171 @@ +// 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.qe.runtime; + +import org.apache.doris.analysis.DescriptorTable; +import org.apache.doris.catalog.Env; +import org.apache.doris.common.Status; +import org.apache.doris.common.jmockit.Deencapsulation; +import org.apache.doris.load.loadv2.LoadManager; +import org.apache.doris.load.loadv2.ProgressManager; +import org.apache.doris.nereids.trees.plans.distribute.PipelineDistributedPlan; +import org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker; +import org.apache.doris.nereids.trees.plans.distribute.worker.job.UnassignedJob; +import org.apache.doris.planner.PlanFragment; +import org.apache.doris.planner.PlanFragmentId; +import org.apache.doris.qe.CoordinatorContext; +import org.apache.doris.qe.NereidsCoordinator; +import org.apache.doris.rpc.BackendServiceProxy; +import org.apache.doris.system.Backend; +import org.apache.doris.thrift.TReportExecStatusParams; +import org.apache.doris.thrift.TStatus; +import org.apache.doris.thrift.TStatusCode; +import org.apache.doris.thrift.TUniqueId; + +import com.google.protobuf.ByteString; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +import java.util.Collections; +import java.util.concurrent.TimeUnit; + +class LoadProcessorTest { + private static final long BACKEND_ID = 9; + private static final int FRAGMENT_ID = 7; + private static final String TRACKING_URL = "http://127.0.0.1/error-log"; + private static final String FIRST_ERROR = "Cannot convert value to integer"; + + @Test + void publishesDiagnosticsBeforeCancellationReleasesLoadWaiters() throws Exception { + Fixture fixture = createFixture(); + Mockito.doAnswer(invocation -> { + // LoadProcessor.cancel invokes remote cancellation before releasing its real latch. + Assertions.assertFalse(fixture.processor.isDone()); + Assertions.assertEquals(TRACKING_URL, fixture.coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR, fixture.coordinator.getFirstErrorMsg()); + Assertions.assertEquals(TStatusCode.DATA_QUALITY_ERROR, + fixture.coordinator.getExecStatus().getErrorCode()); + return null; + }).when(fixture.fragmentsTask).cancelExecute(Mockito.any(Status.class)); + + Assertions.assertFalse(fixture.processor.await(0, TimeUnit.MILLISECONDS)); + fixture.coordinator.updateFragmentExecStatus(report(TStatusCode.DATA_QUALITY_ERROR) + .setTrackingUrl(TRACKING_URL) + .setFirstErrorMsg(FIRST_ERROR)); + + Mockito.verify(fixture.fragmentsTask).cancelExecute(Mockito.any(Status.class)); + Assertions.assertTrue(fixture.processor.await(0, TimeUnit.MILLISECONDS)); + Assertions.assertTrue(fixture.coordinator.join(1)); + } + + @Test + void successfulFinalReportPreservesDiagnostics() throws Exception { + Fixture fixture = createFixture(); + + fixture.coordinator.updateFragmentExecStatus(report(TStatusCode.OK) + .setTrackingUrl(TRACKING_URL) + .setFirstErrorMsg(FIRST_ERROR)); + + Assertions.assertTrue(fixture.coordinator.join(1)); + Assertions.assertTrue(fixture.coordinator.getExecStatus().ok()); + Assertions.assertEquals(TRACKING_URL, fixture.coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR, fixture.coordinator.getFirstErrorMsg()); + Mockito.verify(fixture.fragmentsTask, Mockito.never()).cancelExecute(Mockito.any(Status.class)); + } + + @Test + void failedReportWithoutDiagnosticsPreservesEarlierDiagnostics() throws Exception { + Fixture fixture = createFixture(); + fixture.processor.loadContext.updateTrackingUrl(TRACKING_URL); + fixture.processor.loadContext.updateFirstErrorMsg(FIRST_ERROR); + Mockito.doAnswer(invocation -> { + Assertions.assertFalse(fixture.processor.isDone()); + Assertions.assertEquals(TRACKING_URL, fixture.coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR, fixture.coordinator.getFirstErrorMsg()); + return null; + }).when(fixture.fragmentsTask).cancelExecute(Mockito.any(Status.class)); + + fixture.coordinator.updateFragmentExecStatus(report(TStatusCode.DATA_QUALITY_ERROR)); + + Mockito.verify(fixture.fragmentsTask).cancelExecute(Mockito.any(Status.class)); + Assertions.assertTrue(fixture.coordinator.join(1)); + Assertions.assertEquals(TRACKING_URL, fixture.coordinator.getTrackingUrl()); + Assertions.assertEquals(FIRST_ERROR, fixture.coordinator.getFirstErrorMsg()); + } + + private static TReportExecStatusParams report(TStatusCode status) { + return new TReportExecStatusParams() + .setBackendId(BACKEND_ID) + .setFragmentId(FRAGMENT_ID) + .setDone(true) + .setStatus(new TStatus(status)); + } + + private static Fixture createFixture() { + PlanFragment fragment = Mockito.mock(PlanFragment.class); + Mockito.when(fragment.getFragmentId()).thenReturn(new PlanFragmentId(FRAGMENT_ID)); + UnassignedJob fragmentJob = Mockito.mock(UnassignedJob.class); + Mockito.when(fragmentJob.getFragment()).thenReturn(fragment); + PipelineDistributedPlan distributedPlan = Mockito.mock(PipelineDistributedPlan.class); + Mockito.when(distributedPlan.getFragmentJob()).thenReturn(fragmentJob); + Mockito.when(distributedPlan.getInstanceJobs()).thenReturn(Collections.emptyList()); + + Env env = Mockito.mock(Env.class); + Mockito.when(env.getLoadManager()).thenReturn(Mockito.mock(LoadManager.class)); + Mockito.when(env.getProgressManager()).thenReturn(Mockito.mock(ProgressManager.class)); + NereidsCoordinator coordinator = Mockito.mock(NereidsCoordinator.class, Mockito.CALLS_REAL_METHODS); + CoordinatorContext context; + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentEnv).thenReturn(env); + context = CoordinatorContext.buildForLoad(coordinator, -1, new TUniqueId(1, 2), + Collections.singletonList(fragment), Collections.singletonList(distributedPlan), + Collections.emptyList(), new DescriptorTable(), "UTC", true, false); + } + Deencapsulation.setField(coordinator, "coordinatorContext", context); + LoadProcessor processor = context.asLoadProcessor(); + + Backend backend = Mockito.mock(Backend.class); + Mockito.when(backend.getId()).thenReturn(BACKEND_ID); + BackendWorker worker = Mockito.mock(BackendWorker.class); + Mockito.when(worker.id()).thenReturn(BACKEND_ID); + BackendServiceProxy backendProxy = Mockito.mock(BackendServiceProxy.class); + SingleFragmentPipelineTask fragmentTask = new SingleFragmentPipelineTask( + backend, FRAGMENT_ID, Collections.singleton(new TUniqueId(3, 4))); + MultiFragmentsPipelineTask fragmentsTask = Mockito.spy(new MultiFragmentsPipelineTask( + context, backend, backendProxy, ByteString.EMPTY, + Collections.singletonMap(FRAGMENT_ID, fragmentTask))); + Mockito.doNothing().when(fragmentsTask).cancelExecute(Mockito.any(Status.class)); + processor.setPipelineExecutionTask(new PipelineExecutionTask(context, backendProxy, + Collections.singletonMap(worker, fragmentsTask))); + return new Fixture(coordinator, processor, fragmentsTask); + } + + private static class Fixture { + final NereidsCoordinator coordinator; + final LoadProcessor processor; + final MultiFragmentsPipelineTask fragmentsTask; + + Fixture(NereidsCoordinator coordinator, LoadProcessor processor, MultiFragmentsPipelineTask fragmentsTask) { + this.coordinator = coordinator; + this.processor = processor; + this.fragmentsTask = fragmentsTask; + } + } +}