This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new d2fcb5383ab Fix delayed query cleanup during file metadata scans
(#18679)
d2fcb5383ab is described below
commit d2fcb5383abffc2de168bd9bcefb4004a4c12693
Author: Jackie Tien <[email protected]>
AuthorDate: Sun Sep 20 17:17:53 2026 +0800
Fix delayed query cleanup during file metadata scans (#18679)
---
.../iotdb/db/i18n/DataNodeQueryMessages.java | 3 +
.../iotdb/db/i18n/DataNodeQueryMessages.java | 3 +
.../execution/operator/source/SeriesScanUtil.java | 24 ++
.../source/SeriesScanUtilCancellationTest.java | 399 +++++++++++++++++++++
4 files changed, 429 insertions(+)
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 ef13dccd8ac..ec4f7760f58 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
@@ -21,6 +21,9 @@ package org.apache.iotdb.db.i18n;
public final class DataNodeQueryMessages {
+ public static final String
EXCEPTION_FRAGMENT_INSTANCE_ARG_IS_ALREADY_ARG_B44984B4 =
+ "Fragment instance %s is already %s";
+
public static final String
EXCEPTION_INVALID_THRIFT_MAXIMUM_FRAME_SIZE_ARG_FROM_ARG_A639588B =
"Invalid Thrift maximum frame size %d from %s";
public static final String
MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_BECAUSE_OF_EXCEPTION_ARG_2D8A483D
=
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 f9dde6564ce..c546cfae8c3 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
@@ -21,6 +21,9 @@ package org.apache.iotdb.db.i18n;
public final class DataNodeQueryMessages {
+ public static final String
EXCEPTION_FRAGMENT_INSTANCE_ARG_IS_ALREADY_ARG_B44984B4 =
+ "Fragment instance %s 已处于 %s 状态";
+
public static final String
EXCEPTION_INVALID_THRIFT_MAXIMUM_FRAME_SIZE_ARG_FROM_ARG_A639588B =
"Thrift 最大帧大小 %d 无效，来源：%s";
public static final String
MESSAGE_FAILED_TO_DISPATCH_LOAD_COMMAND_ARG_TO_NODE_ARG_BECAUSE_OF_EXCEPTION_ARG_2D8A483D
=
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
index 1b8ca2b1020..9f54bd63839 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.commons.path.NonAlignedFullPath;
import org.apache.iotdb.db.exception.CorruptedTsFileException;
import org.apache.iotdb.db.i18n.DataNodeQueryMessages;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceState;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet;
import
org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache;
@@ -285,6 +286,7 @@ public class SeriesScanUtil implements Accountable {
// Optional.empty(), it needs to return directly to the checkpoint method
that checks the operator
// execution time slice.
public Optional<Boolean> hasNextFile() throws IOException {
+ checkFragmentInstanceState();
if (runtimeFilterExhausted || !paginationController.hasCurLimit()) {
return Optional.of(false);
}
@@ -2016,8 +2018,10 @@ public class SeriesScanUtil implements Accountable {
}
private Optional<ITimeSeriesMetadata> unpackSeqTsFileResource() throws
IOException {
+ checkFragmentInstanceState();
ITimeSeriesMetadata timeseriesMetadata =
loadTimeSeriesMetadata(orderUtils.getNextSeqFileResource(true), true);
+ checkFragmentInstanceState();
// skip if data type is mismatched which may be caused by delete
if (timeseriesMetadata != null &&
timeseriesMetadata.typeMatch(getTsDataTypeList())) {
timeseriesMetadata.setSeq(true);
@@ -2029,8 +2033,10 @@ public class SeriesScanUtil implements Accountable {
}
private Optional<ITimeSeriesMetadata> unpackUnseqTsFileResource() throws
IOException {
+ checkFragmentInstanceState();
ITimeSeriesMetadata timeseriesMetadata =
loadTimeSeriesMetadata(orderUtils.getNextUnseqFileResource(true),
false);
+ checkFragmentInstanceState();
// skip if data type is mismatched which may be caused by delete
if (timeseriesMetadata != null &&
timeseriesMetadata.typeMatch(getTsDataTypeList())) {
timeseriesMetadata.setSeq(false);
@@ -2041,6 +2047,24 @@ public class SeriesScanUtil implements Accountable {
}
}
+ private void checkFragmentInstanceState() throws IOException {
+ // Compaction also uses this scanner, but its context has no fragment
state machine.
+ if (context.getStateMachine() == null) {
+ return;
+ }
+ FragmentInstanceState state = context.getStateMachine().getState();
+ if (state.isDone()) {
+ // A scan over many overlapping files may stay in one operator call long
after cancellation.
+ // Exit on the driver thread so it can release its lock and finish
resource cleanup.
+ throw new IOException(
+ String.format(
+
DataNodeQueryMessages.EXCEPTION_FRAGMENT_INSTANCE_ARG_IS_ALREADY_ARG_B44984B4,
+ context.getId(),
+ state),
+ context.getFailureCause().orElse(null));
+ }
+ }
+
protected ITimeSeriesMetadata loadTimeSeriesMetadata(TsFileResource
resource, boolean isSeq)
throws IOException {
return FileLoaderUtils.loadTimeSeriesMetadata(
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
new file mode 100644
index 00000000000..a32a83e1e02
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtilCancellationTest.java
@@ -0,0 +1,399 @@
+/*
+ * 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.iotdb.db.queryengine.execution.operator.source;
+
+import org.apache.iotdb.calc.execution.operator.Operator;
+import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager;
+import org.apache.iotdb.commons.exception.QueryTimeoutException;
+import org.apache.iotdb.commons.path.AlignedFullPath;
+import org.apache.iotdb.commons.path.NonAlignedFullPath;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
+import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
+import org.apache.iotdb.db.queryengine.common.QueryId;
+import org.apache.iotdb.db.queryengine.execution.driver.DataDriver;
+import org.apache.iotdb.db.queryengine.execution.driver.DataDriverContext;
+import
org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeManager;
+import org.apache.iotdb.db.queryengine.execution.exchange.sink.ISink;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceExecution;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceState;
+import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceStateMachine;
+import org.apache.iotdb.db.queryengine.execution.schedule.IDriverScheduler;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions;
+import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
+import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource;
+import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
+
+import io.airlift.units.Duration;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.AbstractAlignedTimeSeriesMetadata;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.file.metadata.ITimeSeriesMetadata;
+import org.apache.tsfile.file.metadata.statistics.IntegerStatistics;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntConsumer;
+
+import static org.apache.iotdb.calc.execution.operator.Operator.NOT_BLOCKED;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@RunWith(Parameterized.class)
+public class SeriesScanUtilCancellationTest {
+
+ private final boolean aligned;
+ private final boolean ignoreAllNullRows;
+ private final IDeviceID device;
+ private int loadedFiles;
+ private int previousDataNodeId;
+
+ @Parameterized.Parameters(name = "aligned={0}, ignoreAllNullRows={1}")
+ public static Object[][] parameters() {
+ return new Object[][] {{false, true}, {true, true}, {true, false}};
+ }
+
+ public SeriesScanUtilCancellationTest(boolean aligned, boolean
ignoreAllNullRows) {
+ this.aligned = aligned;
+ this.ignoreAllNullRows = ignoreAllNullRows;
+ this.device =
+ IDeviceID.Factory.DEFAULT_FACTORY.create(
+ ignoreAllNullRows ? new String[] {"root.sg", "d"} : new String[]
{"table", "d"});
+ }
+
+ @Before
+ public void setUp() {
+ previousDataNodeId =
IoTDBDescriptor.getInstance().getConfig().getDataNodeId();
+ // Resource cleanup initializes the query metrics and their coordinator.
+ IoTDBDescriptor.getInstance().getConfig().setDataNodeId(1);
+ }
+
+ @After
+ public void tearDown() {
+
IoTDBDescriptor.getInstance().getConfig().setDataNodeId(previousDataNodeId);
+ }
+
+ @Test
+ public void testTerminalStateBeforeScanDoesNotLoadMetadata() {
+ for (FragmentInstanceState state :
+ new FragmentInstanceState[] {
+ FragmentInstanceState.CANCELLED,
+ FragmentInstanceState.ABORTED,
+ FragmentInstanceState.FINISHED,
+ FragmentInstanceState.FAILED
+ }) {
+ FragmentInstanceContext context = newContext(Runnable::run);
+ SeriesScanUtil scanner = newScanner(context, Ordering.ASC, false, 3,
count -> {});
+ switch (state) {
+ case CANCELLED:
+ context.cancel();
+ break;
+ case ABORTED:
+ context.abort();
+ break;
+ case FINISHED:
+ context.finished();
+ break;
+ case FAILED:
+ context.failed(new QueryTimeoutException());
+ break;
+ default:
+ throw new AssertionError(state);
+ }
+
+ IOException exception = assertThrows(IOException.class,
scanner::hasNextFile);
+ assertSame(context.getFailureCause().orElse(null), exception.getCause());
+ assertEquals(0, loadedFiles);
+ assertEquals(state, context.getStateMachine().getState());
+ }
+ }
+
+ @Test
+ public void testCancellationDuringOverlappingFileScan() {
+ for (Ordering ordering : new Ordering[] {Ordering.ASC, Ordering.DESC}) {
+ for (boolean sequence : new boolean[] {true, false}) {
+ FragmentInstanceContext context = newContext(Runnable::run);
+ SeriesScanUtil scanner =
+ newScanner(
+ context,
+ ordering,
+ sequence,
+ 4,
+ count -> {
+ if (count == 2) {
+ context.cancel();
+ }
+ });
+
+ assertThrows(IOException.class, scanner::hasNextFile);
+ assertEquals(2, loadedFiles);
+ assertEquals(FragmentInstanceState.CANCELLED,
context.getStateMachine().getState());
+ }
+ }
+ }
+
+ @Test
+ public void testTimeoutDuringLastFileReadPreservesCause() {
+ for (boolean sequence : new boolean[] {true, false}) {
+ FragmentInstanceContext context = newContext(Runnable::run);
+ QueryTimeoutException timeout = new QueryTimeoutException();
+ SeriesScanUtil scanner =
+ newScanner(context, Ordering.ASC, sequence, 1, count ->
context.failed(timeout));
+
+ IOException exception = assertThrows(IOException.class,
scanner::hasNextFile);
+ assertSame(timeout, exception.getCause());
+ assertEquals(1, loadedFiles);
+ assertEquals(FragmentInstanceState.FAILED,
context.getStateMachine().getState());
+ }
+ }
+
+ @Test
+ public void testActiveQueryStillLoadsOverlappingFiles() throws IOException {
+ for (boolean flushing : new boolean[] {false, true}) {
+ FragmentInstanceContext context = newContext(Runnable::run);
+ if (flushing) {
+ context.transitionToFlushing();
+ }
+ SeriesScanUtil scanner = newScanner(context, Ordering.ASC, false, 4,
count -> {});
+
+ assertTrue(scanner.hasNextFile().get());
+ assertEquals(4, loadedFiles);
+ assertEquals(
+ flushing ? FragmentInstanceState.FLUSHING :
FragmentInstanceState.RUNNING,
+ context.getStateMachine().getState());
+ }
+ }
+
+ @Test
+ public void testCancellationWithCachedMetadata() throws IOException {
+ FragmentInstanceContext context = newContext(Runnable::run);
+ SeriesScanUtil scanner = newScanner(context, Ordering.ASC, false, 4, count
-> {});
+ assertTrue(scanner.hasNextFile().get());
+ context.cancel();
+
+ assertThrows(IOException.class, scanner::hasNextFile);
+ assertEquals(4, loadedFiles);
+ }
+
+ @Test
+ public void testCancellationDuringMissingMetadataRead() {
+ for (boolean sequence : new boolean[] {true, false}) {
+ FragmentInstanceContext context = newContext(Runnable::run);
+ SeriesScanUtil scanner =
+ newScanner(context, Ordering.ASC, sequence, 1, count ->
context.cancel(), true);
+
+ assertThrows(IOException.class, scanner::hasNextFile);
+ assertEquals(1, loadedFiles);
+ }
+ }
+
+ @Test
+ public void testCompactionContextWithoutStateMachine() throws IOException {
+ FragmentInstanceContext context =
+ FragmentInstanceContext.createFragmentInstanceContextForCompaction(0);
+ SeriesScanUtil scanner = newScanner(context, Ordering.ASC, false, 2, count
-> {});
+
+ assertTrue(scanner.hasNextFile().get());
+ assertEquals(2, loadedFiles);
+ }
+
+ @Test(timeout = 15000)
+ public void testTimeoutUnblocksDriverResourceCleanup() throws Exception {
+ ExecutorService notifications = Executors.newSingleThreadExecutor();
+ FragmentInstanceContext context = newContext(notifications);
+ context.initializeNumOfDrivers(1);
+ try {
+ CountDownLatch closeRequested = new CountDownLatch(1);
+ QueryTimeoutException timeout = new QueryTimeoutException();
+ SeriesScanUtil scanner =
+ newScanner(
+ context,
+ Ordering.ASC,
+ false,
+ 4,
+ count -> {
+ if (count == 2) {
+ context.failed(timeout);
+ try {
+ // The notification thread has requested close while this
thread owns the
+ // driver lock, just as in a query that is still loading
metadata.
+ assertTrue(closeRequested.await(5, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ }
+ });
+ Operator operator = mock(Operator.class);
+ doReturn(NOT_BLOCKED).when(operator).isBlocked();
+ when(operator.hasNextWithTimer()).thenReturn(true);
+ when(operator.nextWithTimer())
+ .thenAnswer(
+ invocation -> {
+ scanner.hasNextFile();
+ return null;
+ });
+ ISink sink = mock(ISink.class);
+ doReturn(NOT_BLOCKED).when(sink).isFull();
+ DataDriverContext driverContext = new DataDriverContext(context, 0);
+ driverContext.setSink(sink);
+ DataDriver driver =
+ new DataDriver(operator, driverContext, 0) {
+ @Override
+ public void close() {
+ super.close();
+ closeRequested.countDown();
+ }
+ };
+ MPPDataExchangeManager exchangeManager =
mock(MPPDataExchangeManager.class);
+ FragmentInstanceExecution.createFragmentInstanceExecution(
+ mock(IDriverScheduler.class),
+ context.getId(),
+ context,
+ Collections.singletonList(driver),
+ sink,
+ context.getStateMachine(),
+ 1000,
+ false,
+ exchangeManager);
+
+ assertSame(
+ timeout,
+ assertThrows(
+ QueryTimeoutException.class,
+ () -> driver.processFor(new Duration(1, TimeUnit.SECONDS))));
+ // This can complete only after the preceding cleanup callback gets past
allDriversClosed.
+ notifications.submit(() -> {}).get(5, TimeUnit.SECONDS);
+ assertEquals(2, loadedFiles);
+ verify(operator).close();
+ verify(context.getMemoryReservationContext()).releaseAllReservedMemory();
+ verify(exchangeManager)
+ .deRegisterFragmentInstanceFromMemoryPool(
+ context.getId().getQueryId().getId(),
context.getId().getFragmentInstanceId(), true);
+ } finally {
+ // Ensure a failed assertion cannot leave the notification thread
waiting on the latch.
+ context.decrementNumOfUnClosedDriver();
+ notifications.shutdownNow();
+ assertTrue(notifications.awaitTermination(5, TimeUnit.SECONDS));
+ }
+ }
+
+ private FragmentInstanceContext newContext(Executor executor) {
+ FragmentInstanceId id =
+ new FragmentInstanceId(new PlanFragmentId(new
QueryId("scan_cancellation"), 0), "0");
+ return FragmentInstanceContext.createFragmentInstanceContext(
+ id, new FragmentInstanceStateMachine(id, executor),
mock(MemoryReservationManager.class));
+ }
+
+ private SeriesScanUtil newScanner(
+ FragmentInstanceContext context,
+ Ordering ordering,
+ boolean sequence,
+ int fileCount,
+ IntConsumer onRead) {
+ return newScanner(context, ordering, sequence, fileCount, onRead, false);
+ }
+
+ private SeriesScanUtil newScanner(
+ FragmentInstanceContext context,
+ Ordering ordering,
+ boolean sequence,
+ int fileCount,
+ IntConsumer onRead,
+ boolean missingMetadata) {
+ context.setIgnoreAllNullRows(ignoreAllNullRows);
+ loadedFiles = 0;
+ List<TsFileResource> resources = new ArrayList<>();
+ for (int i = 0; i < fileCount; i++) {
+ TsFileResource resource = mock(TsFileResource.class);
+ when(resource.getOrderTimeForSeq(device, true)).thenReturn(0L);
+ when(resource.getOrderTimeForSeq(device, false)).thenReturn(100L);
+ when(resource.getOrderTimeForUnseq(device, true)).thenReturn(0L);
+ when(resource.getOrderTimeForUnseq(device, false)).thenReturn(100L);
+ resources.add(resource);
+ }
+ QueryDataSource source =
+ new QueryDataSource(
+ sequence ? resources : Collections.emptyList(),
+ sequence ? Collections.emptyList() : resources);
+ source.setSingleDevice(true);
+ IntegerStatistics statistics = new IntegerStatistics();
+ statistics.update(0L, 0);
+ statistics.update(100L, 100);
+ ITimeSeriesMetadata metadata =
+ aligned ? mock(AbstractAlignedTimeSeriesMetadata.class) :
mock(ITimeSeriesMetadata.class);
+ doReturn(statistics).when(metadata).getStatistics();
+ when(metadata.typeMatch(anyList())).thenReturn(true);
+ MeasurementSchema schema = new MeasurementSchema("s", TSDataType.INT32);
+ SeriesScanOptions.Builder builder = new SeriesScanOptions.Builder();
+ builder.withAllSensors(Collections.singleton("s"));
+ SeriesScanUtil scanner;
+ if (aligned) {
+ scanner =
+ new AlignedSeriesScanUtil(
+ new AlignedFullPath(
+ device, Collections.singletonList("s"),
Collections.singletonList(schema)),
+ ordering,
+ builder.build(),
+ context) {
+ @Override
+ protected AbstractAlignedTimeSeriesMetadata loadTimeSeriesMetadata(
+ TsFileResource resource, boolean isSeq) {
+ onRead.accept(++loadedFiles);
+ return missingMetadata ? null :
(AbstractAlignedTimeSeriesMetadata) metadata;
+ }
+ };
+ } else {
+ scanner =
+ new SeriesScanUtil(
+ new NonAlignedFullPath(device, schema), ordering,
builder.build(), context) {
+ @Override
+ protected ITimeSeriesMetadata loadTimeSeriesMetadata(
+ TsFileResource resource, boolean isSeq) {
+ onRead.accept(++loadedFiles);
+ return missingMetadata ? null : metadata;
+ }
+ };
+ }
+ scanner.initQueryDataSource(source);
+ return scanner;
+ }
+}
