[To dev/1.3] Prevent deleted values from reappearing after a local LOAD crash

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java

@@ -29,6 +29,16 @@ public class MppJVMConfig implements JVMConfig {
   private int maxHeapSize;
   private int maxDirectMemorySize;
   private String timezone;
+  private int debugPort;
+
+  /** Enables loopback-only JDWP for tests that need to observe an exact node execution stage. */
+  public void setDebugPort(int debugPort) {
+    this.debugPort = debugPort;
+  }
+
+  public int getDebugPort() {
+    return debugPort;
+  }
 
   @Override
   public JVMConfig setInitHeapSize(int initSize) {
@@ -94,6 +104,7 @@ public void override(@Nullable MppJVMConfig config) {
     this.setMaxHeapSize(config.getMaxHeapSize());
     this.setMaxDirectMemorySize(config.getMaxDirectMemorySize());
     this.setTimezone(config.getTimezone());
+    this.setDebugPort(config.getDebugPort());
   }
 
   public static Builder builder() {

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java

@@ -416,6 +416,11 @@ public void start() {
       File stdoutFile = new File(getLogPath());
       List<String> startCmd = new ArrayList<>();
       startCmd.add(JAVA_CMD);
+      if (jvmConfig.getDebugPort() > 0) {
+        startCmd.add(
+            "-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=127.0.0.1:"
+                + jvmConfig.getDebugPort());
+      }
       if (!SystemUtils.IS_JAVA_1_8) {
         startCmd.add("--add-opens=java.base/java.util.concurrent=ALL-UNNAMED");
         startCmd.add("--add-opens=java.base/java.lang=ALL-UNNAMED");

integration-test/src/test/java/org/apache/iotdb/db/it/utils/NodeDebugger.java

@@ -0,0 +1,232 @@
+/*
+ * 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.it.utils;
+
+import com.sun.jdi.Bootstrap;
+import com.sun.jdi.ObjectReference;
+import com.sun.jdi.ReferenceType;
+import com.sun.jdi.StackFrame;
+import com.sun.jdi.StringReference;
+import com.sun.jdi.ThreadReference;
+import com.sun.jdi.VMDisconnectedException;
+import com.sun.jdi.Value;
+import com.sun.jdi.VirtualMachine;
+import com.sun.jdi.connect.AttachingConnector;
+import com.sun.jdi.connect.Connector;
+import com.sun.jdi.event.BreakpointEvent;
+import com.sun.jdi.event.ClassPrepareEvent;
+import com.sun.jdi.event.Event;
+import com.sun.jdi.event.EventSet;
+import com.sun.jdi.event.MethodEntryEvent;
+import com.sun.jdi.event.MethodExitEvent;
+import com.sun.jdi.event.VMStartEvent;
+import com.sun.jdi.request.BreakpointRequest;
+import com.sun.jdi.request.ClassPrepareRequest;
+import com.sun.jdi.request.EventRequest;
+import com.sun.jdi.request.MethodEntryRequest;
+import com.sun.jdi.request.MethodExitRequest;
+import org.junit.Assert;
+
+import java.net.ConnectException;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
+
+/**
+ * Pauses only the node thread executing a selected method. Other threads (including the active
+ * scanner) keep running. All control stays in the test JVM; no product hooks or agent bytecode are
+ * installed. Closing the debugger releases a paused thread even when an assertion fails.
+ */
+public final class NodeDebugger implements AutoCloseable {
+  private final VirtualMachine vm;
+  private final String className;
+  private final String methodName;
+  private final boolean observeMethodEntries;
+  private final Set<ReferenceType> breakpointTypes = new HashSet<>();
+  private EventSet suspended;
+  private ThreadReference pausedThread;
+
+  public NodeDebugger(int port, String className, String methodName) throws Exception {
+    this(port, className, methodName, false);
+  }
+
+  public NodeDebugger(int port, String className, String methodName, boolean observeMethodEntries)
+      throws Exception {
+    this.className = className;
+    this.methodName = methodName;
+    this.observeMethodEntries = observeMethodEntries;
+    final AttachingConnector connector =
+        Bootstrap.virtualMachineManager().attachingConnectors().stream()
+            .filter(candidate -> candidate.name().equals("com.sun.jdi.SocketAttach"))
+            .findFirst()
+            .orElseThrow(AssertionError::new);
+    final Map<String, Connector.Argument> arguments = connector.defaultArguments();
+    arguments.get("hostname").setValue("127.0.0.1");
+    arguments.get("port").setValue(Integer.toString(port));
+    arguments.get("timeout").setValue("10000");
+    final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+    VirtualMachine connected;
+    while (true) {
+      try {
+        connected = connector.attach(arguments);
+        break;
+      } catch (ConnectException e) {
+        if (System.nanoTime() >= deadline) {
+          throw e;
+        }
+        // ProcessBuilder.start returns before the suspended VM has opened its JDWP listener.
+        TimeUnit.MILLISECONDS.sleep(100);
+      }
+    }
+    vm = connected;
+    try {
+      if (observeMethodEntries) {
+        installMethodEntryRequest();
+      } else {
+        final ClassPrepareRequest prepare = vm.eventRequestManager().createClassPrepareRequest();
+        prepare.addClassFilter(className);
+        prepare.setSuspendPolicy(EventRequest.SUSPEND_EVENT_THREAD);
+        prepare.enable();
+        for (ReferenceType type : vm.classesByName(className)) {
+          if (type.isPrepared()) {
+            installBreakpoint(type);
+          }
+        }
+      }
+    } catch (Exception | AssertionError e) {
+      vm.dispose();
+      throw e;
+    }
+  }
+
+  private void installMethodEntryRequest() {
+    final MethodEntryRequest entry = vm.eventRequestManager().createMethodEntryRequest();
+    entry.addClassFilter(className);
+    entry.setSuspendPolicy(EventRequest.SUSPEND_EVENT_THREAD);
+    entry.enable();
+  }
+
+  /** Releases only the initial VMStart suspension after all requests have been installed. */
+  public void resumeStartup() throws Exception {
+    suspended = vm.eventQueue().remove(10000);
+    Assert.assertNotNull("Suspended startup must produce a VMStart event", suspended);
+    Assert.assertTrue(suspended.stream().anyMatch(event -> event instanceof VMStartEvent));
+    resume();
+  }
+
+  private void installBreakpoint(ReferenceType type) {
+    if (!breakpointTypes.add(type)) {
+      return;
+    }
+    Assert.assertEquals(className + "." + methodName, 1, type.methodsByName(methodName).size());
+    final BreakpointRequest breakpoint =
+        vm.eventRequestManager()
+            .createBreakpointRequest(type.methodsByName(methodName).get(0).location());
+    breakpoint.setSuspendPolicy(EventRequest.SUSPEND_EVENT_THREAD);
+    breakpoint.enable();
+  }
+
+  public StackFrame await(Predicate<StackFrame> condition) throws Exception {
+    final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
+    while (System.nanoTime() < deadline) {
+      final EventSet events =
+          vm.eventQueue()
+              .remove(Math.max(1, TimeUnit.NANOSECONDS.toMillis(deadline - System.nanoTime())));
+      if (events == null) {
+        break;
+      }
+      // Remember ownership before inspecting a frame: failed predicates must also release it.
+      suspended = events;
+      for (Event event : events) {
+        if (event instanceof ClassPrepareEvent && !observeMethodEntries) {
+          installBreakpoint(((ClassPrepareEvent) event).referenceType());
+        } else if (event instanceof MethodEntryEvent
+            && ((MethodEntryEvent) event).method().name().equals(methodName)) {
+          final ThreadReference thread = ((MethodEntryEvent) event).thread();
+          final StackFrame frame = thread.frame(0);
+          if (condition.test(frame)) {
+            pausedThread = thread;
+            return frame;
+          }
+        } else if (event instanceof BreakpointEvent) {
+          final StackFrame frame = ((BreakpointEvent) event).thread().frame(0);
+          if (condition.test(frame)) {
+            pausedThread = ((BreakpointEvent) event).thread();
+            return frame;
+          }
+        } else if (event instanceof MethodExitEvent
+            && ((MethodExitEvent) event).method().name().equals(methodName)) {
+          final StackFrame frame = ((MethodExitEvent) event).thread().frame(0);
+          if (condition.test(frame)) {
+            // Resume must let cleanup and the SQL response run without another exit pause.
+            event.request().disable();
+            return frame;
+          }
+        }
+      }
+      resume();
+    }
+    throw new AssertionError("Did not reach " + className + "." + methodName);
+  }
+
+  public void resume() {
+    if (suspended != null) {
+      suspended.resume();
+      suspended = null;
+    }
+  }
+
+  /** Changes an entry pause into an exit pause on the same method and connection. */
+  public void pauseOnExit() {
+    vm.eventRequestManager()
+        .deleteEventRequests(
+            new java.util.ArrayList<>(vm.eventRequestManager().breakpointRequests()));
+    vm.eventRequestManager()
+        .deleteEventRequests(
+            new java.util.ArrayList<>(vm.eventRequestManager().classPrepareRequests()));
+    vm.eventRequestManager()
+        .deleteEventRequests(
+            new java.util.ArrayList<>(vm.eventRequestManager().methodEntryRequests()));
+    final MethodExitRequest exit = vm.eventRequestManager().createMethodExitRequest();
+    exit.addClassFilter(className);
+    Assert.assertNotNull("An entry pause is required before selecting its exit", pausedThread);
+    exit.addThreadFilter(pausedThread);
+    exit.setSuspendPolicy(EventRequest.SUSPEND_EVENT_THREAD);
+    exit.enable();
+  }
+
+  /** Reads java.io.File's path without invoking code in the paused node. */
+  public static String filePath(Value value) {
+    final ObjectReference file = (ObjectReference) value;
+    return ((StringReference) file.getValue(file.referenceType().fieldByName("path"))).value();
+  }
+
+  @Override
+  public void close() {
+    try {
+      resume();
+      vm.dispose();
+    } catch (VMDisconnectedException expectedAfterCrash) {
+      // A forced node stop is the operation under test.
+    }
+  }
+}

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

@@ -62,4 +62,6 @@ private DataNodeQueryMessages() {}
           "There is not enough memory for Query %s, the contextHolder is %s,current remaining free "
               + "memory is %dB, already reserved memory for this context in total is %dB, the memory "
               + "requested this time is %dB";
+  public static final String LOG_DELETE_AFTER_LOADING_ARG_ERROR_C606A060 =
+      "Delete After Loading {} error.";
 }

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

@@ -0,0 +1,48 @@
+/*
+ * 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.i18n;
+
+public final class StorageEngineMessages {
+
+  private StorageEngineMessages() {}
+
+  public static final String
+      LOG_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_ARG_TARGET_ARG_13205CEC =
+          "File renaming failed when loading tsfile. Origin: {}, Target: {}";
+  public static final String
+      EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_ARG_TARGET_ARG_BECAUSE_ARG_4A8CF51E =
+          "File renaming failed when loading tsfile. Origin: %s, Target: %s, because %s";
+  public static final String
+      LOG_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_ARG_TARGET_ARG_5589CD09 =
+          "File renaming failed when loading .resource file. Origin: {}, Target: {}";
+  public static final String
+      EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_ARG_TARGET_ARG_BECAUSE_ARG_77B1F6DE =
+          "File renaming failed when loading .resource file. Origin: %s, Target: %s, because %s";
+  public static final String
+      LOG_FILE_RENAMING_FAILED_WHEN_LOADING_MOD_FILE_ORIGIN_ARG_TARGET_ARG_2493B8EE =
+          "File renaming failed when loading .mod file. Origin: {}, Target: {}";
+  public static final String
+      EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_MOD_FILE_ORIGIN_ARG_TARGET_ARG_BECAUSE_ARG_93A853C4 =
+          "File renaming failed when loading .mod file. Origin: %s, Target: %s, because %s";
+  public static final String LOG_CANNOT_DELETE_LOCALMODFILE_ARG_F9CD0EA6 =
+      "Cannot delete localModFile {}";
+  public static final String LOG_FAILED_TO_DELETE_FILE_OR_DIRECTORY_ARG_7EC21473 =
+      "Failed to delete file or directory: {}";
+}

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

@@ -59,4 +59,5 @@ private DataNodeQueryMessages() {}
   public static final String
       QUERY_EXCEPTION_THERE_IS_NOT_ENOUGH_MEMORY_FOR_QUERY_S_THE_CONTEXTHOLDER_546CDD02 =
           "Query %s 内存不足，contextHolder 为 %s，当前剩余空闲内存为 %dB，该 context 已预留总内存为 %dB，本次请求内存为 %dB。";
+  public static final String LOG_DELETE_AFTER_LOADING_ARG_ERROR_C606A060 = "加载后删除 {} 时出错。";
 }

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

@@ -0,0 +1,46 @@
+/*
+ * 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.i18n;
+
+public final class StorageEngineMessages {
+
+  private StorageEngineMessages() {}
+
+  public static final String
+      LOG_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_ARG_TARGET_ARG_13205CEC =
+          "加载 TsFile 时重命名文件失败。源文件：{}，目标文件：{}";
+  public static final String
+      EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_ARG_TARGET_ARG_BECAUSE_ARG_4A8CF51E =
+          "加载 TsFile 时重命名文件失败。源文件：%s，目标文件：%s，原因：%s";
+  public static final String
+      LOG_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_ARG_TARGET_ARG_5589CD09 =
+          "加载 .resource 文件时重命名失败。源文件：{}，目标文件：{}";
+  public static final String
+      EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_ARG_TARGET_ARG_BECAUSE_ARG_77B1F6DE =
+          "加载 .resource 文件时重命名失败。源文件：%s，目标文件：%s，原因：%s";
+  public static final String
+      LOG_FILE_RENAMING_FAILED_WHEN_LOADING_MOD_FILE_ORIGIN_ARG_TARGET_ARG_2493B8EE =
+          "加载 .mod 文件时重命名失败。源文件：{}，目标文件：{}";
+  public static final String
+      EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_MOD_FILE_ORIGIN_ARG_TARGET_ARG_BECAUSE_ARG_93A853C4 =
+          "加载 .mod 文件时重命名失败。源文件：%s，目标文件：%s，原因：%s";
+  public static final String LOG_CANNOT_DELETE_LOCALMODFILE_ARG_F9CD0EA6 = "无法删除本地 mods 文件 {}";
+  public static final String LOG_FAILED_TO_DELETE_FILE_OR_DIRECTORY_ARG_7EC21473 = "删除文件或目录失败：{}";
+}