This is an automated email from the ASF dual-hosted git repository. Wei-hao-Li pushed a commit to branch fixKillQ in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0fc2deb9badb0c38bc45bac7ecdf415e5af48e2c Author: Weihao Li <[email protected]> AuthorDate: Wed Oct 28 00:28:40 2026 +0800 fix Signed-off-by: Weihao Li <[email protected]> --- .../iotdb/db/i18n/DataNodeQueryMessages.java | 3 ++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 2 + .../queryengine/plan/execution/QueryExecution.java | 16 +++++++- .../plan/execution/QueryExecutionRetryTest.java | 47 ++++++++++++++++++++++ 4 files changed, 67 insertions(+), 1 deletion(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index be040f13e21..c326a7d0fd2 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -1682,6 +1682,9 @@ public final class DataNodeQueryMessages { "[CleanUpQuery]]"; public static final String RELEASE_QUERY_RESOURCE_STATE = "[ReleaseQueryResource] state is: {}"; + public static final String IGNORE_EXCEPTION_AFTER_QUERY_DONE = + "[IgnoreException] Query {} is already in terminal state {}, the exception caused by " + + "concurrent resource cleanup is ignored"; public static final String SKIP_EXECUTE = "[SkipExecute]"; public static final String SKIP_EXECUTE_AFTER_LOGICAL_PLAN = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java index e0c51972501..a406b246c4e 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java @@ -1662,6 +1662,8 @@ public final class DataNodeQueryMessages { "[清理查询]"; public static final String RELEASE_QUERY_RESOURCE_STATE = "[释放查询资源] 状态为: {}"; + public static final String IGNORE_EXCEPTION_AFTER_QUERY_DONE = + "[忽略异常] 查询 {} 已处于终态 {}，忽略并发清理资源导致的异常"; public static final String SKIP_EXECUTE = "[跳过执行]"; public static final String SKIP_EXECUTE_AFTER_LOGICAL_PLAN = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java index b8f753d70a0..11b9eb7114b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecution.java @@ -179,7 +179,21 @@ public class QueryExecution implements IQueryExecution { public void start() { try { startInternal(); - } catch (RuntimeException | Error e) { + } catch (RuntimeException e) { + finishAnalysisAttemptForCurrentState(); + if (stateMachine.getState().isDone()) { + // The query has already reached a terminal state (e.g. killed by others), and the status + // recorded by the state machine is the authoritative result. An exception thrown here is + // usually caused by resources being released concurrently, so it must not override it. + LOGGER.warn( + DataNodeQueryMessages.IGNORE_EXCEPTION_AFTER_QUERY_DONE, + context.getQueryId(), + stateMachine.getState(), + e); + return; + } + throw e; + } catch (Error e) { finishAnalysisAttemptForCurrentState(); throw e; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecutionRetryTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecutionRetryTest.java index 1d717d0f7d9..8761346df5e 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecutionRetryTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/execution/QueryExecutionRetryTest.java @@ -37,7 +37,9 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; +import java.io.UncheckedIOException; import java.lang.reflect.Field; +import java.nio.file.NoSuchFileException; import java.time.ZoneId; import java.util.ArrayList; import java.util.Arrays; @@ -49,6 +51,7 @@ import java.util.concurrent.ScheduledExecutorService; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -107,6 +110,50 @@ public class QueryExecutionRetryTest { assertFalse(planner.getAnalysisEvents().contains("commit")); } + @Test + public void testPlanningFailureIsNotSwallowedWhileNotDone() { + RecordingPlanner planner = + new RecordingPlanner(false) { + @Override + public LogicalQueryPlan doLogicalPlan(IAnalysis analysis, MPPQueryContext context) { + throw new IllegalStateException("planning failed"); + } + }; + QueryExecution execution = + new QueryExecution(planner, createQueryContext("plan_fails"), stateMachineExecutor); + + IllegalStateException thrown = assertThrows(IllegalStateException.class, execution::start); + assertEquals("planning failed", thrown.getMessage()); + } + + @Test + public void testCleanupFailureAfterQueryCanceledIsIgnoredDuringStart() { + QueryExecution[] self = new QueryExecution[1]; + RecordingPlanner planner = + new RecordingPlanner(false) { + @Override + public LogicalQueryPlan doLogicalPlan(IAnalysis analysis, MPPQueryContext context) { + // Simulate KILL QUERY arriving while the planner is still fetching device schema. The + // cancel releases the spill directory, so the next segment write fails with + // NoSuchFileException, which must not override the recorded QUERY_WAS_KILLED status. + self[0].cancel(); + throw new UncheckedIOException( + new NoSuchFileException( + "/tmp/device-entry/cancel_during_start/1/raw/segment-000002.tmp")); + } + }; + QueryExecution execution = + new QueryExecution( + planner, createQueryContext("cancel_during_start"), stateMachineExecutor); + self[0] = execution; + + execution.start(); + + assertEquals(Arrays.asList("begin", "analyze", "rollback"), planner.getAnalysisEvents()); + ExecutionResult result = execution.getStatus(); + assertEquals(TSStatusCode.QUERY_WAS_KILLED.getStatusCode(), result.status.code); + } + private static MPPQueryContext createQueryContext(String queryId) { MPPQueryContext context = new MPPQueryContext(
