From 69f65a854db07000527c1c6cbb77b7828deed9f1 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Mon, 9 Nov 2026 11:11:58 +0800 Subject: [PATCH] Prevent deleted values from reappearing after a local LOAD crash (#19136) * Prevent partial local LOAD installation after process crashes * Keep committed LOAD successful when source cleanup fails * Avoid counting converted LOAD mods twice * Record LOAD mods metrics only after publication * Align LOAD cleanup tests with source group retention (cherry picked from commit c6e229aef50608cb67bfb4fd94717d05f4ef3d0f) --- .../it/env/cluster/config/MppJVMConfig.java | 11 + .../env/cluster/node/AbstractNodeWrapper.java | 5 + .../iotdb/db/it/IoTDBLoadTsFileInstallIT.java | 422 ++++++++++++++++++ .../iotdb/db/it/utils/NodeDebugger.java | 232 ++++++++++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 2 + .../iotdb/db/i18n/StorageEngineMessages.java | 48 ++ .../iotdb/db/i18n/DataNodeQueryMessages.java | 1 + .../iotdb/db/i18n/StorageEngineMessages.java | 46 ++ .../plan/node/load/LoadSingleTsFileNode.java | 12 +- .../storageengine/dataregion/DataRegion.java | 248 ++++++---- .../planner/node/load/LoadTsFileNodeTest.java | 34 +- .../dataregion/DataRegionTest.java | 144 ++++++ 12 files changed, 1107 insertions(+), 98 deletions(-) create mode 100644 integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileInstallIT.java create mode 100644 integration-test/src/test/java/org/apache/iotdb/db/it/utils/NodeDebugger.java create mode 100644 iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java create mode 100644 iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java index d9c51717cf8e1..7e46af797360b 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/config/MppJVMConfig.java +++ b/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() { diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java index afad120e94cbb..701c19de81761 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/node/AbstractNodeWrapper.java +++ b/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 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"); diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileInstallIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileInstallIT.java new file mode 100644 index 0000000000000..46e1fd5bc986e --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileInstallIT.java @@ -0,0 +1,422 @@ +/* + * 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; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.path.MeasurementPath; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.db.it.utils.NodeDebugger; +import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileManager; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.config.MppJVMConfig; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.ClusterIT; + +import com.sun.jdi.ObjectReference; +import com.sun.jdi.StackFrame; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.PlainDeviceID; +import org.apache.tsfile.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.awaitility.Awaitility; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.io.File; +import java.net.ServerSocket; +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** A single-replica process fixture for the local whole-file installation commit boundary. */ +@RunWith(IoTDBTestRunner.class) +@Category(ClusterIT.class) +public class IoTDBLoadTsFileInstallIT { + private static final String DATABASE = "root.load.install"; + private static final String DATA_REGION = + "org.apache.iotdb.db.storageengine.dataregion.DataRegion"; + private static final long[] TIMES = {-3, -2, -1}; + private static final List SCHEMAS = + Arrays.asList( + new MeasurementSchema("s", TSDataType.INT64), + new MeasurementSchema("txt", TSDataType.TEXT)); + private static int debugPort; + private File directory; + + @BeforeClass + public static void setUpClass() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + debugPort = socket.getLocalPort(); + } + ((MppJVMConfig) EnvFactory.getEnv().getConfig().getDataNodeJVMConfig()).setDebugPort(debugPort); + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setDataReplicationFactor(1) + .setSchemaReplicationFactor(1) + .setDataRegionConsensusProtocolClass(ConsensusFactory.SIMPLE_CONSENSUS) + .setDefaultDataRegionGroupNumPerDatabase(1) + .setDataRegionGroupExtensionPolicy("CUSTOM"); + EnvFactory.getEnv().initClusterEnvironment(1, 1); + } + + @AfterClass + public static void tearDownClass() { + try { + EnvFactory.getEnv().cleanClusterEnvironment(); + } finally { + ((MppJVMConfig) EnvFactory.getEnv().getConfig().getDataNodeJVMConfig()).setDebugPort(0); + } + } + + @Before + public void setUp() throws Exception { + directory = Files.createTempDirectory("load-install-").toFile(); + execute("create database " + DATABASE); + } + + @After + public void tearDown() throws Exception { + try { + if (!node().isAlive()) { + startNode(); + } + execute("delete database " + DATABASE); + } finally { + try (Stream paths = Files.walk(directory.toPath())) { + for (Path path : paths.sorted(Comparator.reverseOrder()).collect(Collectors.toList())) { + Files.delete(path); + } + } + } + } + + @Test + public void testCrashBeforeDeletionInstallationDoesNotPublishPartialFile() throws Exception { + assertCrashBoundary(false, false); + } + + @Test + public void testCrashAfterDeletionInstallationDoesNotPublishPartialFile() throws Exception { + assertCrashBoundary(false, true); + } + + @Test + public void testPublishedFileSurvivesCrashBeforeClientAcknowledgement() throws Exception { + assertCrashBoundary(true, true); + } + + @Test + public void testSourceCleanupFailureDoesNotFailCommittedLoad() throws Exception { + for (boolean failDataDeletion : new boolean[] {false, true}) { + final String prefix = DATABASE + ".cleanup_data" + failDataDeletion; + final File source = new File(directory, (200 + (failDataDeletion ? 2 : 0)) + "-0-0-0.tsfile"); + writeSource(source, prefix); + final File sourceResource = new File(source + TsFileResource.RESOURCE_SUFFIX); + final File sourceMods = new File(source + ModificationFile.FILE_SUFFIX); + final byte[] sourceData = Files.readAllBytes(source.toPath()); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + try (NodeDebugger debugger = + new NodeDebugger(debugPort, TsFileManager.class.getName(), "add")) { + final Future request = + executor.submit( + () -> { + execute(load(source, true)); + return null; + }); + final StackFrame frame = + debugger.await( + candidate -> { + final ObjectReference resource = + (ObjectReference) candidate.getArgumentValues().get(0); + return NodeDebugger.filePath( + resource.getValue(resource.referenceType().fieldByName("file"))) + .contains(File.separator + DATABASE + File.separator); + }); + final ObjectReference resource = (ObjectReference) frame.getArgumentValues().get(0); + final File target = + new File( + NodeDebugger.filePath( + resource.getValue(resource.referenceType().fieldByName("file")))); + debugger.pauseOnExit(); + debugger.resume(); + debugger.await(candidate -> true); + Assert.assertFalse( + "The original SQL must still be awaiting its response", request.isDone()); + // Registration is complete, and DataRegion has not started source cleanup yet. + assertSource(source, true); + Assert.assertTrue(target.isFile()); + Assert.assertTrue(new File(target + TsFileResource.RESOURCE_SUFFIX).isFile()); + Assert.assertTrue(new File(target + ModificationFile.FILE_SUFFIX).isFile()); + Assert.assertFalse(new File(target + ".load-pending").exists()); + + // A nonempty directory makes deleteIfExists fail on every supported platform. + final File blocked = failDataDeletion ? source : sourceResource; + Files.delete(blocked.toPath()); + Files.createDirectory(blocked.toPath()); + Files.write( + blocked.toPath().resolve("retained"), failDataDeletion ? sourceData : new byte[] {1}); + debugger.resume(); + request.get(30, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + + if (failDataDeletion) { + Assert.assertTrue(source.isDirectory()); + Assert.assertArrayEquals( + sourceData, Files.readAllBytes(source.toPath().resolve("retained"))); + Assert.assertTrue(sourceResource.isFile()); + Assert.assertTrue(sourceMods.isFile()); + } else { + Assert.assertFalse(source.exists()); + Assert.assertTrue(sourceResource.isDirectory()); + Assert.assertTrue(new File(sourceResource, "retained").isFile()); + // SQL batch cleanup can remove the remaining mods once the source data is gone. + Assert.assertFalse(sourceMods.exists()); + } + assertRows(prefix); + } + } + + private void assertCrashBoundary(boolean published, boolean deletionsInstalled) throws Exception { + for (boolean delete : new boolean[] {false, true}) { + final String prefix = DATABASE + ".d" + delete; + final File source = new File(directory, (100 + (delete ? 2 : 0)) + "-0-0-0.tsfile"); + writeSource(source, prefix); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final File target; + final boolean visibleBeforeCrash; + final Future request; + try { + try (NodeDebugger debugger = + new NodeDebugger( + debugPort, DATA_REGION, published ? "loadTsFileToUnSequence" : "loadModFile")) { + request = + executor.submit( + () -> { + execute(load(source, delete)); + return null; + }); + StackFrame frame = + debugger.await( + candidate -> + NodeDebugger.filePath(candidate.getArgumentValues().get(0)) + .equals(source.getAbsolutePath())); + if (published) { + debugger.pauseOnExit(); + debugger.resume(); + frame = debugger.await(candidate -> true); + final ObjectReference resource = (ObjectReference) frame.getArgumentValues().get(1); + target = + new File( + NodeDebugger.filePath( + resource.getValue(resource.referenceType().fieldByName("file")))); + Assert.assertTrue( + "Published file must already include all deletions", + new File(target + ModificationFile.FILE_SUFFIX).isFile()); + } else { + target = new File(NodeDebugger.filePath(frame.getArgumentValues().get(1))); + if (deletionsInstalled) { + debugger.pauseOnExit(); + debugger.resume(); + debugger.await(candidate -> true); + } + Assert.assertEquals( + deletionsInstalled, new File(target + ModificationFile.FILE_SUFFIX).isFile()); + } + Assert.assertTrue(new File(target + TsFileResource.RESOURCE_SUFFIX).isFile()); + visibleBeforeCrash = target.isFile(); + Assert.assertFalse( + "The original SQL must still be awaiting its response", request.isDone()); + node().stopForcibly(); + final ExecutionException failure = + Assert.assertThrows( + ExecutionException.class, () -> request.get(30, TimeUnit.SECONDS)); + Assert.assertTrue(failure.toString(), failure.getCause() instanceof SQLException); + } + // The commit can complete its source cleanup before the client receives a response. + assertSource(source, !published || !delete); + startNode(); + if (published) { + Assert.assertTrue(visibleBeforeCrash); + assertRows(prefix); + } else { + // Check recovery before any new LOAD; retrying first could conceal resurrection. + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement(); + ResultSet rows = + statement.executeQuery("select s from " + prefix + ".d0 where time=-2")) { + Assert.assertFalse("Recovery must not resurrect a deleted value", rows.next()); + } + Assert.assertFalse( + "A file without its deletions must never be published", visibleBeforeCrash); + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement(); + ResultSet rows = statement.executeQuery("select ** from " + prefix + ".**")) { + Assert.assertFalse("Unpublished data must stay invisible after recovery", rows.next()); + } + Assert.assertFalse(target.exists()); + Assert.assertFalse(new File(target + ".load-pending").exists()); + Assert.assertFalse(new File(target + TsFileResource.RESOURCE_SUFFIX).exists()); + Assert.assertFalse(new File(target + ModificationFile.FILE_SUFFIX).exists()); + // This is an explicit new submission, not recovery of the interrupted request. + execute(load(source, delete)); + assertRows(prefix); + assertSource(source, !delete); + } + } finally { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + } + + private static void writeSource(File source, String prefix) throws Exception { + try (TsFileWriter writer = new TsFileWriter(source)) { + for (int device = 0; device < 2; device++) { + final String path = prefix + ".d" + device; + writer.registerAlignedTimeseries(new org.apache.tsfile.read.common.Path(path), SCHEMAS); + final Tablet tablet = new Tablet(path, SCHEMAS, TIMES.length); + for (int row = 0; row < TIMES.length; row++) { + tablet.addTimestamp(row, TIMES[row]); + tablet.addValue("s", row, device * 100L + row); + tablet.addValue("txt", row, row == 0 ? null : "d" + device + "_r" + row); + } + tablet.rowSize = TIMES.length; + writer.writeAligned(tablet); + } + } + final TsFileResource resource = new TsFileResource(source); + for (int device = 0; device < 2; device++) { + final IDeviceID id = new PlainDeviceID(prefix + ".d" + device); + resource.updateStartTime(id, TIMES[0]); + resource.updateEndTime(id, TIMES[TIMES.length - 1]); + } + resource.serialize(); + try (ModificationFile mods = resource.getModFile()) { + for (int device = 0; device < 2; device++) { + mods.write( + new Deletion( + new MeasurementPath(prefix + ".d" + device + ".s"), + Long.MAX_VALUE, + TIMES[1], + TIMES[1])); + } + } + } + + private static void assertRows(String prefix) throws Exception { + for (int device = 0; device < 2; device++) { + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement(); + ResultSet rows = statement.executeQuery("select s,txt from " + prefix + ".d" + device)) { + for (int row = 0; row < TIMES.length; row++) { + Assert.assertTrue(rows.next()); + Assert.assertEquals(TIMES[row], rows.getLong(1)); + Assert.assertEquals(row == 1 ? null : device * 100L + row, rows.getObject(2)); + Assert.assertEquals(row == 0 ? null : "d" + device + "_r" + row, rows.getString(3)); + } + Assert.assertFalse(rows.next()); + } + } + } + + private static void assertSource(File source, boolean exists) { + Assert.assertEquals(exists, source.isFile()); + Assert.assertEquals(exists, new File(source + TsFileResource.RESOURCE_SUFFIX).isFile()); + Assert.assertEquals(exists, new File(source + ModificationFile.FILE_SUFFIX).isFile()); + } + + private static String load(File source, boolean delete) { + return "load '" + + source.getAbsolutePath() + + "' with ('database-level'='2','tablet-conversion-threshold'='-1','on-success'='" + + (delete ? "delete" : "none") + + "')"; + } + + private static void execute(String sql) throws Exception { + try (Connection connection = + EnvFactory.getEnv().getWriteOnlyConnectionWithSpecifiedDataNode(node()); + Statement statement = connection.createStatement()) { + statement.execute(sql); + } + } + + private static DataNodeWrapper node() { + return EnvFactory.getEnv().getDataNodeWrapperList().get(0); + } + + private static void startNode() throws Exception { + node().start(); + EnvFactory.getEnv() + .ensureNodeStatus( + Collections.singletonList(node()), Collections.singletonList(NodeStatus.Running)); + Awaitility.await() + .atMost(60, TimeUnit.SECONDS) + .untilAsserted( + () -> { + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement(); + ResultSet regions = statement.executeQuery("show regions")) { + int count = 0; + while (regions.next()) { + if (DATABASE.equals(regions.getString("Database")) + && "DataRegion".equals(regions.getString("Type"))) { + count++; + Assert.assertEquals( + NodeStatus.Running.getStatus(), regions.getString("Status")); + } + } + Assert.assertTrue("The actual DataRegion must be ready after restart", count > 0); + } + }); + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/NodeDebugger.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/NodeDebugger.java new file mode 100644 index 0000000000000..c37bd06f1a10a --- /dev/null +++ b/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 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 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 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. + } + } +} 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 1f8300c2807ed..39bacf00a529a 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 @@ -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."; } diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/StorageEngineMessages.java new file mode 100644 index 0000000000000..d799152d9b5fc --- /dev/null +++ b/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: {}"; +} 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 32e74a0e5b94f..0950eea371854 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 @@ -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 = "加载后删除 {} 时出错。"; } diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/StorageEngineMessages.java new file mode 100644 index 0000000000000..6f43d0b34c8ce --- /dev/null +++ b/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 = "删除文件或目录失败：{}"; +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java index b73068d1395e8..8784f1970813c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/load/LoadSingleTsFileNode.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.utils.RetryUtils; import org.apache.iotdb.commons.utils.TimePartitionUtils; import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.i18n.DataNodeQueryMessages; import org.apache.iotdb.db.queryengine.plan.analyze.IAnalysis; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.PlanNodeId; @@ -219,20 +220,25 @@ public void clean() { if (!deleteAfterLoad) { return; } - deleteFile(tsFile); + // A retained source TsFile still needs its sidecars for any later explicit LOAD. + if (!deleteFile(tsFile)) { + return; + } deleteFile(new File(tsFile.getAbsolutePath() + TsFileResource.RESOURCE_SUFFIX)); deleteFile(new File(tsFile.getAbsolutePath() + ModificationFile.FILE_SUFFIX)); } - private void deleteFile(final File file) { + private boolean deleteFile(final File file) { try { RetryUtils.retryOnException( () -> { Files.deleteIfExists(file.toPath()); return null; }); + return true; } catch (final Exception e) { - LOGGER.warn("Delete After Loading {} error.", file, e); + LOGGER.warn(DataNodeQueryMessages.LOG_DELETE_AFTER_LOADING_ARG_ERROR_C606A060, file, e); + return false; } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 986d06fa8173f..43043aa7668ee 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -51,6 +51,7 @@ import org.apache.iotdb.db.exception.query.OutOfTTLException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.exception.quota.ExceedQuotaException; +import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.pipe.source.dataregion.realtime.listener.PipeInsertionDataNodeListener; import org.apache.iotdb.db.queryengine.common.DeviceContext; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; @@ -130,7 +131,6 @@ import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; -import org.apache.commons.io.FileUtils; import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.file.metadata.PlainDeviceID; @@ -150,6 +150,7 @@ import java.io.IOException; import java.nio.file.Files; import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -219,6 +220,7 @@ public class DataRegion implements IDataRegionForQuery { private static final int MERGE_MOD_START_VERSION_NUM = 1; private static final Logger logger = LoggerFactory.getLogger(DataRegion.class); + private static final String LOAD_PENDING_SUFFIX = ".load-pending"; /** * A read write lock for guaranteeing concurrent safety when accessing all fields in this class @@ -760,6 +762,14 @@ private Map> getAllFiles(List folders) } else { // some TsFileResource may be being persisted when the system crashed, try recovering // such resources + final File[] pendingLoads = + fsFactory.listFilesBySuffix( + partitionFolder.getAbsolutePath(), TSFILE_SUFFIX + LOAD_PENDING_SUFFIX); + if (pendingLoads != null) { + for (File pendingLoad : pendingLoads) { + deleteUnpublishedLoad(pendingLoad); + } + } continueFailedRenames(partitionFolder, TEMP_SUFFIX); String partitionName = partitionFolder.getName(); File[] tsFilesInThisFolder = @@ -801,6 +811,20 @@ private void continueFailedRenames(File fileFolder, String suffix) throws IOExce } } + private void deleteUnpublishedLoad(File pendingFile) throws IOException { + final File targetFile = + new File( + pendingFile + .getPath() + .substring(0, pendingFile.getPath().length() - LOAD_PENDING_SUFFIX.length())); + if (!targetFile.exists()) { + Files.deleteIfExists(new File(targetFile + RESOURCE_SUFFIX).toPath()); + Files.deleteIfExists(new File(targetFile + ModificationFile.FILE_SUFFIX).toPath()); + } + // Keep the pending file until all sidecars are removed, so interrupted cleanup is retryable. + Files.deleteIfExists(pendingFile.toPath()); + } + /** check if the tsfile's time is smaller than system current time. */ private void checkTsFileTime(File tsFile, long currentTime) throws DataRegionException { String[] items = tsFile.getName().replace(TSFILE_SUFFIX, "").split(FILE_NAME_SEPARATOR); @@ -3535,116 +3559,93 @@ private boolean loadTsFileToUnSequence( if (!targetFile.getParentFile().exists()) { targetFile.getParentFile().mkdirs(); } - try { - if (deleteOriginFile) { - RetryUtils.retryOnException( - () -> { - FileUtils.moveFile(tsFileToLoad, targetFile); - return null; - }); - } else { - RetryUtils.retryOnException( - () -> { - Files.copy(tsFileToLoad.toPath(), targetFile.toPath()); - return null; - }); - } - } catch (final IOException e) { - logger.warn( - "File renaming failed when loading tsfile. Origin: {}, Target: {}", - tsFileToLoad.getAbsolutePath(), - targetFile.getAbsolutePath(), - e); - throw new LoadFileException( - String.format( - "File renaming failed when loading tsfile. Origin: %s, Target: %s, because %s", - tsFileToLoad.getAbsolutePath(), targetFile.getAbsolutePath(), e.getMessage())); - } - + // Recovery only discovers names ending in .tsfile. Keep the data invisible until its resource + // and every deletion are installed, including while the data is being copied. + final File pendingFile = new File(targetFile + LOAD_PENDING_SUFFIX); final File resourceFileToLoad = fsFactory.getFile(tsFileToLoad.getAbsolutePath() + RESOURCE_SUFFIX); final File targetResourceFile = fsFactory.getFile(targetFile.getAbsolutePath() + RESOURCE_SUFFIX); + boolean loadPublished = false; try { - if (deleteOriginFile) { + try { + if (pendingFile.exists()) { + deleteUnpublishedLoad(pendingFile); + } RetryUtils.retryOnException( () -> { - FileUtils.moveFile(resourceFileToLoad, targetResourceFile); + Files.copy( + tsFileToLoad.toPath(), pendingFile.toPath(), StandardCopyOption.REPLACE_EXISTING); return null; }); - } else { + } catch (final IOException e) { + logger.warn( + StorageEngineMessages + .LOG_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_ARG_TARGET_ARG_13205CEC, + tsFileToLoad.getAbsolutePath(), + targetFile.getAbsolutePath(), + e); + throw new LoadFileException( + String.format( + StorageEngineMessages + .EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_ARG_TARGET_ARG_BECAUSE_ARG_4A8CF51E, + tsFileToLoad.getAbsolutePath(), + targetFile.getAbsolutePath(), + e.getMessage())); + } + + try { RetryUtils.retryOnException( () -> { - Files.copy(resourceFileToLoad.toPath(), targetResourceFile.toPath()); + Files.copy( + resourceFileToLoad.toPath(), + targetResourceFile.toPath(), + StandardCopyOption.REPLACE_EXISTING); return null; }); + } catch (final IOException e) { + logger.warn( + StorageEngineMessages + .LOG_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_ARG_TARGET_ARG_5589CD09, + resourceFileToLoad.getAbsolutePath(), + targetResourceFile.getAbsolutePath(), + e); + throw new LoadFileException( + String.format( + StorageEngineMessages + .EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_ARG_TARGET_ARG_BECAUSE_ARG_77B1F6DE, + resourceFileToLoad.getAbsolutePath(), + targetResourceFile.getAbsolutePath(), + e.getMessage())); } - } catch (final IOException e) { - logger.warn( - "File renaming failed when loading .resource file. Origin: {}, Target: {}", - resourceFileToLoad.getAbsolutePath(), - targetResourceFile.getAbsolutePath(), - e); - throw new LoadFileException( - String.format( - "File renaming failed when loading .resource file. Origin: %s, Target: %s, because %s", - resourceFileToLoad.getAbsolutePath(), - targetResourceFile.getAbsolutePath(), - e.getMessage())); - } - final File modFileToLoad = - fsFactory.getFile(tsFileToLoad.getAbsolutePath() + ModificationFile.FILE_SUFFIX); - try { - if (modFileToLoad.exists()) { - // when successfully loaded, the filepath of the resource will be changed to the IoTDB data - // dir, so we can add a suffix to find the old modification file. - final File targetModFile = - fsFactory.getFile(targetFile.getAbsolutePath() + ModificationFile.FILE_SUFFIX); - try { - RetryUtils.retryOnException( - () -> { - Files.deleteIfExists(targetModFile.toPath()); - return null; - }); - } catch (final IOException e) { - logger.warn("Cannot delete localModFile {}", targetModFile, e); - } + // Keep all source files intact until the complete destination is published. + loadModFile(tsFileToLoad, targetFile); + try { + Files.move(pendingFile.toPath(), targetFile.toPath(), StandardCopyOption.ATOMIC_MOVE); + } catch (IOException e) { + throw new LoadFileException(e); + } + loadPublished = true; + } finally { + tsFileResource.setModFile(null); + if (!loadPublished && pendingFile.exists()) { try { - final long modFileSize = modFileToLoad.length(); - if (deleteOriginFile) { - RetryUtils.retryOnException( - () -> { - FileUtils.moveFile(modFileToLoad, targetModFile); - return null; - }); - } else { - RetryUtils.retryOnException( - () -> { - Files.copy(modFileToLoad.toPath(), targetModFile.toPath()); - return null; - }); - } - - FileMetrics.getInstance().increaseModFileNum(1); - FileMetrics.getInstance().increaseModFileSize(modFileSize); - } catch (final IOException e) { + deleteUnpublishedLoad(pendingFile); + } catch (IOException e) { logger.warn( - "File renaming failed when loading .mod file. Origin: {}, Target: {}", - modFileToLoad.getAbsolutePath(), - targetModFile.getAbsolutePath(), + StorageEngineMessages.LOG_FAILED_TO_DELETE_FILE_OR_DIRECTORY_ARG_7EC21473, + pendingFile, e); - throw new LoadFileException( - String.format( - "File renaming failed when loading .mod file. Origin: %s, Target: %s, because %s", - modFileToLoad.getAbsolutePath(), - targetModFile.getAbsolutePath(), - e.getMessage())); } } - } finally { - // ModFile will be updated during the next call to `getModFile` - tsFileResource.setModFile(null); + } + + // Only published mods belong in FileMetrics. Also replace any cached source-side handle. + tsFileResource.setModFile(ModificationFile.getNormalMods(tsFileResource)); + if (tsFileResource.modFileExists()) { + FileMetrics.getInstance().increaseModFileNum(1); + FileMetrics.getInstance().increaseModFileSize(tsFileResource.getModFile().getSize()); } // Listen before the tsFile is added into tsFile manager to avoid it being compacted @@ -3653,9 +3654,74 @@ private boolean loadTsFileToUnSequence( tsFileManager.add(tsFileResource, false); + if (deleteOriginFile) { + for (File sourceFile : + new File[] { + tsFileToLoad, resourceFileToLoad, new File(tsFileToLoad + ModificationFile.FILE_SUFFIX) + }) { + try { + RetryUtils.retryOnException( + () -> { + Files.deleteIfExists(sourceFile.toPath()); + return null; + }); + } catch (IOException e) { + // The installed TsFile is already visible and registered; source cleanup cannot roll it + // back. + logger.warn( + StorageEngineMessages.LOG_FAILED_TO_DELETE_FILE_OR_DIRECTORY_ARG_7EC21473, + sourceFile, + e); + break; + } + } + } + return true; } + private void loadModFile(File tsFileToLoad, File targetFile) throws LoadFileException { + final File modFileToLoad = + fsFactory.getFile(tsFileToLoad.getAbsolutePath() + ModificationFile.FILE_SUFFIX); + if (modFileToLoad.exists()) { + // when successfully loaded, the filepath of the resource will be changed to the IoTDB data + // dir, so we can add a suffix to find the old modification file. + final File targetModFile = + fsFactory.getFile(targetFile.getAbsolutePath() + ModificationFile.FILE_SUFFIX); + try { + RetryUtils.retryOnException( + () -> { + Files.deleteIfExists(targetModFile.toPath()); + return null; + }); + } catch (final IOException e) { + logger.warn( + StorageEngineMessages.LOG_CANNOT_DELETE_LOCALMODFILE_ARG_F9CD0EA6, targetModFile, e); + } + try { + RetryUtils.retryOnException( + () -> { + Files.copy(modFileToLoad.toPath(), targetModFile.toPath()); + return null; + }); + } catch (final IOException e) { + logger.warn( + StorageEngineMessages + .LOG_FILE_RENAMING_FAILED_WHEN_LOADING_MOD_FILE_ORIGIN_ARG_TARGET_ARG_2493B8EE, + modFileToLoad.getAbsolutePath(), + targetModFile.getAbsolutePath(), + e); + throw new LoadFileException( + String.format( + StorageEngineMessages + .EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_MOD_FILE_ORIGIN_ARG_TARGET_ARG_BECAUSE_ARG_93A853C4, + modFileToLoad.getAbsolutePath(), + targetModFile.getAbsolutePath(), + e.getMessage())); + } + } + } + /** * Get all working sequence tsfile processors * diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java index 9dc760e9a65db..3edb49f1daf9c 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/node/load/LoadTsFileNodeTest.java @@ -90,11 +90,11 @@ public void testLoadTsFilePieceNode() { } @Test - public void testCleanContinuesAfterOneFileCannotBeDeleted() throws Exception { + public void testCleanRetainsSidecarsWhenTsFileCannotBeDeleted() throws Exception { final File tempDir = Files.createTempDirectory("load-node-clean").toFile(); try { - // A non-empty directory at the TsFile path makes that deletion fail deterministically. The - // companion cleanup must still continue instead of sharing the same try-catch block. + // A retained source TsFile must keep its resource and deletions for any later explicit LOAD. + // A non-empty directory at the TsFile path makes that deletion fail deterministically. final File tsFile = new File(tempDir, "1-0-0-0.tsfile"); Assert.assertTrue(tsFile.mkdirs()); Assert.assertTrue(new File(tsFile, "non-empty").createNewFile()); @@ -108,7 +108,33 @@ public void testCleanContinuesAfterOneFileCannotBeDeleted() throws Exception { node.clean(); Assert.assertTrue(tsFile.exists()); - Assert.assertFalse(resourceFile.exists()); + Assert.assertTrue(resourceFile.isFile()); + Assert.assertTrue(modsFile.isFile()); + } finally { + deleteRecursively(tempDir); + } + } + + @Test + public void testCleanContinuesAfterOneSidecarCannotBeDeleted() throws Exception { + final File tempDir = Files.createTempDirectory("load-node-clean").toFile(); + try { + final File tsFile = new File(tempDir, "1-0-0-0.tsfile"); + final File resourceFile = new File(tsFile + TsFileResource.RESOURCE_SUFFIX); + final File modsFile = new File(tsFile + ModificationFile.FILE_SUFFIX); + Assert.assertTrue(tsFile.createNewFile()); + Assert.assertTrue(resourceFile.mkdir()); + final File retained = new File(resourceFile, "non-empty"); + Assert.assertTrue(retained.createNewFile()); + Assert.assertTrue(modsFile.createNewFile()); + + final LoadSingleTsFileNode node = + new LoadSingleTsFileNode(new PlanNodeId(""), new TsFileResource(tsFile), true, 0L); + node.clean(); + + Assert.assertFalse(tsFile.exists()); + Assert.assertTrue(resourceFile.isDirectory()); + Assert.assertTrue(retained.isFile()); Assert.assertFalse(modsFile.exists()); } finally { deleteRecursively(tempDir); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java index 93223cc8ccf83..ce3973c23f890 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java @@ -37,6 +37,7 @@ import org.apache.iotdb.db.exception.TsFileProcessorException; import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.exception.WriteProcessRejectException; +import org.apache.iotdb.db.exception.load.LoadFileException; import org.apache.iotdb.db.exception.query.QueryProcessException; import org.apache.iotdb.db.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; @@ -61,6 +62,8 @@ import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk; import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor; +import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.read.QueryDataSource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; @@ -194,6 +197,147 @@ public static InsertRowNode buildInsertRowNodeByTSRecord(TSRecord record) return insertRowNode; } + @Test + public void testLoadInstallationRetainsOrDeletesTheCompleteSourceGroup() throws Exception { + for (boolean deleteOrigin : new boolean[] {false, true}) { + for (boolean withMods : new boolean[] {false, true}) { + final File sourceDir = + new File(TestConstant.OUTPUT_DATA_DIR, "load-source-" + deleteOrigin + "-" + withMods); + Files.createDirectories(sourceDir.toPath()); + final File source = new File(sourceDir, "1-1-0-0.tsfile"); + final MeasurementSchema schema = new MeasurementSchema(measurementId, TSDataType.INT32); + try (TsFileWriter writer = new TsFileWriter(source)) { + writer.registerAlignedTimeseries(new Path(deviceId), Collections.singletonList(schema)); + final Tablet tablet = new Tablet(deviceId, Collections.singletonList(schema), 3); + for (int row = 0; row < 3; row++) { + tablet.addTimestamp(row, row + 1L); + tablet.addValue(measurementId, row, row); + } + tablet.rowSize = 3; + writer.writeAligned(tablet); + } + final TsFileResource resource = Mockito.spy(new TsFileResource(source)); + resource.setStatus(TsFileResourceStatus.NORMAL); + resource.updateStartTime(new PlainDeviceID(deviceId), 1); + resource.updateEndTime(new PlainDeviceID(deviceId), 3); + resource.serialize(); + final File sourceMods = new File(source + ModificationFile.FILE_SUFFIX); + if (withMods) { + try (ModificationFile mods = resource.getModFile()) { + mods.write( + new Deletion( + new MeasurementPath(deviceId + "." + measurementId), Long.MAX_VALUE, 2, 2)); + } + } + // Exercise rebinding even when the source handle was cached before installation. + resource.getModFile(); + final byte[] data = Files.readAllBytes(source.toPath()); + final byte[] deletions = withMods ? Files.readAllBytes(sourceMods.toPath()) : null; + final File sourceResource = new File(source + TsFileResource.RESOURCE_SUFFIX); + final File savedResource = new File(sourceDir, "saved-resource"); + Files.move(sourceResource.toPath(), savedResource.toPath()); + try { + final LoadFileException failure = + Assert.assertThrows( + LoadFileException.class, + () -> dataRegion.loadNewTsFile(resource, deleteOrigin, false, true)); + Assert.assertTrue( + failure.getMessage(), + failure.getMessage().contains(sourceResource.getAbsolutePath())); + final File rejectedTarget = resource.getTsFile(); + Assert.assertNotEquals(source.getAbsolutePath(), rejectedTarget.getAbsolutePath()); + Assert.assertFalse(rejectedTarget.exists()); + Assert.assertFalse(new File(rejectedTarget + ".load-pending").exists()); + Assert.assertFalse(new File(rejectedTarget + TsFileResource.RESOURCE_SUFFIX).exists()); + Assert.assertFalse(new File(rejectedTarget + ModificationFile.FILE_SUFFIX).exists()); + Assert.assertArrayEquals(data, Files.readAllBytes(source.toPath())); + Assert.assertEquals(withMods, sourceMods.exists()); + } finally { + Files.move(savedResource.toPath(), sourceResource.toPath()); + resource.setFile(source); + } + final int modFilesBeforeLoad = FileMetrics.getInstance().getModFileNum(); + final long modSizeBeforeLoad = getModFileSize(); + if (withMods) { + // A nonempty destination directory rejects atomic publication on Windows and Unix. + Mockito.doAnswer( + invocation -> { + invocation.callRealMethod(); + final File file = invocation.getArgument(0); + if (!file.getParentFile() + .getCanonicalFile() + .equals(sourceDir.getCanonicalFile())) { + Files.createDirectories(file.toPath()); + Files.write(file.toPath().resolve("block-publication"), new byte[] {1}); + } + return null; + }) + .when(resource) + .setFile(Mockito.any(File.class)); + File rejectedTarget = null; + try { + Assert.assertThrows( + LoadFileException.class, + () -> dataRegion.loadNewTsFile(resource, deleteOrigin, false, true)); + rejectedTarget = resource.getTsFile(); + Assert.assertTrue(rejectedTarget.isDirectory()); + Assert.assertTrue(new File(rejectedTarget + TsFileResource.RESOURCE_SUFFIX).isFile()); + Assert.assertTrue(new File(rejectedTarget + ModificationFile.FILE_SUFFIX).isFile()); + Assert.assertFalse(new File(rejectedTarget + ".load-pending").exists()); + Assert.assertEquals(modFilesBeforeLoad, FileMetrics.getInstance().getModFileNum()); + Assert.assertEquals(modSizeBeforeLoad, getModFileSize()); + Assert.assertArrayEquals(data, Files.readAllBytes(source.toPath())); + Assert.assertArrayEquals(deletions, Files.readAllBytes(sourceMods.toPath())); + Assert.assertTrue(sourceResource.isFile()); + } finally { + Mockito.doCallRealMethod().when(resource).setFile(Mockito.any(File.class)); + if (rejectedTarget != null) { + // Rollback preserves sidecars at an occupied target; remove this test's + // blocker/group. + Files.delete(rejectedTarget.toPath().resolve("block-publication")); + Files.delete(rejectedTarget.toPath()); + Files.deleteIfExists( + new File(rejectedTarget + TsFileResource.RESOURCE_SUFFIX).toPath()); + Files.deleteIfExists( + new File(rejectedTarget + ModificationFile.FILE_SUFFIX).toPath()); + } + resource.setFile(source); + } + } + dataRegion.loadNewTsFile(resource, deleteOrigin, false, true); + final File installed = resource.getTsFile(); + Assert.assertArrayEquals(data, Files.readAllBytes(installed.toPath())); + Assert.assertTrue(new File(installed + TsFileResource.RESOURCE_SUFFIX).isFile()); + Assert.assertFalse(new File(installed + ".load-pending").exists()); + final File installedMods = new File(installed + ModificationFile.FILE_SUFFIX); + Assert.assertEquals( + installedMods.getCanonicalFile(), + new File(resource.getModFile().getFilePath()).getCanonicalFile()); + Assert.assertEquals(withMods ? 1 : 0, resource.getModFile().getModifications().size()); + Assert.assertEquals(withMods, installedMods.exists()); + Assert.assertEquals( + modFilesBeforeLoad + (withMods ? 1 : 0), FileMetrics.getInstance().getModFileNum()); + Assert.assertEquals(modSizeBeforeLoad + installedMods.length(), getModFileSize()); + if (withMods) { + Assert.assertArrayEquals(deletions, Files.readAllBytes(installedMods.toPath())); + } + Assert.assertEquals(!deleteOrigin, source.exists()); + Assert.assertEquals( + !deleteOrigin, new File(source + TsFileResource.RESOURCE_SUFFIX).exists()); + Assert.assertEquals(withMods && !deleteOrigin, sourceMods.exists()); + } + } + } + + private static long getModFileSize() throws Exception { + final Field metricsField = FileMetrics.class.getDeclaredField("MODS_FILE_METRICS"); + metricsField.setAccessible(true); + final Object metrics = metricsField.get(null); + final Field sizeField = metrics.getClass().getDeclaredField("modFileSize"); + sizeField.setAccessible(true); + return ((java.util.concurrent.atomic.AtomicLong) sizeField.get(metrics)).get(); + } + @Test public void testLoadRateLimiterAccountsForSourceFileBeforeTransfer() throws Exception { final LoadTsFileCostMetricsSet metrics = LoadTsFileCostMetricsSet.getInstance();