Fix delayed query cleanup during file metadata scans (#18679)

Author: JackieTien97

iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -21,6 +21,9 @@
 
 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 =

iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java

@@ -21,6 +21,9 @@
 
 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 =

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java

@@ -25,6 +25,7 @@
 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 @@ protected DescPriorityMergeReader getDescPriorityMergeReader() {
   // 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 @@ private void unpackAllOverlappedTsFilesToTimeSeriesMetadata(long endpointTime)
   }
 
   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 @@ private Optional<ITimeSeriesMetadata> unpackSeqTsFileResource() throws IOExcepti
   }
 
   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 @@ private Optional<ITimeSeriesMetadata> unpackUnseqTsFileResource() throws IOExcep
     }
   }
 
+  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(