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 7304e3024cd14..75ed973a74242 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 @@ -33,8 +33,18 @@ public class MppJVMConfig implements JVMConfig { private int maxHeapSize; private int maxDirectMemorySize; private String timezone; + private int debugPort; private List jvmOptions = Collections.emptyList(); + /** 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; + } + /** Additional arguments for child test-node JVMs, for example a test-only Java agent. */ public void setJvmOptions(List options) { this.jvmOptions = new ArrayList<>(options); @@ -108,6 +118,7 @@ public void override(@Nullable MppJVMConfig config) { this.setMaxHeapSize(config.getMaxHeapSize()); this.setMaxDirectMemorySize(config.getMaxDirectMemorySize()); this.setTimezone(config.getTimezone()); + this.setDebugPort(config.getDebugPort()); this.setJvmOptions(config.getJvmOptions()); } 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 a5bc823eafda2..1b9bf2f9fc97b 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 @@ -483,6 +483,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()); + } startCmd.addAll(jvmConfig.getJvmOptions()); if (!SystemUtils.IS_JAVA_1_8) { startCmd.add("--add-opens=java.base/java.util.concurrent=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..a6cf4797c9f0b --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileInstallIT.java @@ -0,0 +1,451 @@ +/* + * 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.ModificationFile; +import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry; +import org.apache.iotdb.db.storageengine.dataregion.modification.v1.Deletion; +import org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1; +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.write.TsFileWriter; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +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}) { + for (boolean legacy : new boolean[] {false, true}) { + final String prefix = DATABASE + ".cleanup_data" + failDataDeletion + "_v" + legacy; + final File source = + new File( + directory, (200 + (failDataDeletion ? 2 : 0) + (legacy ? 1 : 0)) + "-0-0-0.tsfile"); + writeSource(source, prefix, legacy); + final File sourceResource = new File(source + TsFileResource.RESOURCE_SUFFIX); + final File sourceMods = + legacy + ? ModificationFileV1.getNormalMods(source) + : ModificationFile.getExclusiveMods(source); + 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, legacy, true); + Assert.assertTrue(target.isFile()); + Assert.assertTrue(new File(target + TsFileResource.RESOURCE_SUFFIX).isFile()); + Assert.assertTrue(ModificationFile.getExclusiveMods(target).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}) { + for (boolean legacy : new boolean[] {false, true}) { + final String prefix = DATABASE + ".d" + delete + "_v" + legacy; + final File source = + new File(directory, (100 + (delete ? 2 : 0) + (legacy ? 1 : 0)) + "-0-0-0.tsfile"); + writeSource(source, prefix, legacy); + 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", + ModificationFile.getExclusiveMods(target).isFile()); + } else { + target = new File(NodeDebugger.filePath(frame.getArgumentValues().get(1))); + if (deletionsInstalled) { + debugger.pauseOnExit(); + debugger.resume(); + debugger.await(candidate -> true); + } + Assert.assertEquals( + deletionsInstalled, ModificationFile.getExclusiveMods(target).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, legacy, !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(ModificationFile.getExclusiveMods(target).exists()); + Assert.assertFalse(ModificationFileV1.getNormalMods(target).exists()); + // This is an explicit new submission, not recovery of the interrupted request. + execute(load(source, delete)); + assertRows(prefix); + assertSource(source, legacy, !delete); + } + } finally { + executor.shutdownNow(); + Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + } + } + + private static void writeSource(File source, String prefix, boolean legacy) throws Exception { + try (TsFileWriter writer = new TsFileWriter(source)) { + for (int device = 0; device < 2; device++) { + final String path = prefix + ".d" + device; + writer.registerAlignedTimeseries(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(row, "s", device * 100L + row); + tablet.addValue(row, "txt", row == 0 ? null : "d" + device + "_r" + row); + } + writer.writeAligned(tablet); + } + } + final TsFileResource resource = new TsFileResource(source); + for (int device = 0; device < 2; device++) { + final IDeviceID id = IDeviceID.Factory.DEFAULT_FACTORY.create(prefix + ".d" + device); + resource.updateStartTime(id, TIMES[0]); + resource.updateEndTime(id, TIMES[TIMES.length - 1]); + } + resource.serialize(); + if (legacy) { + try (ModificationFileV1 mods = ModificationFileV1.getNormalMods(resource)) { + for (int device = 0; device < 2; device++) { + mods.write( + new Deletion( + new MeasurementPath(prefix + ".d" + device + ".s"), + Long.MAX_VALUE, + TIMES[1], + TIMES[1])); + } + } + } else { + try (ModificationFile mods = resource.getExclusiveModFile()) { + for (int device = 0; device < 2; device++) { + mods.write( + new TreeDeletionEntry( + new MeasurementPath(prefix + ".d" + device + ".s"), 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 legacy, boolean exists) { + Assert.assertEquals(exists, source.isFile()); + Assert.assertEquals(exists, new File(source + TsFileResource.RESOURCE_SUFFIX).isFile()); + Assert.assertEquals( + exists, + (legacy + ? ModificationFileV1.getNormalMods(source) + : ModificationFile.getExclusiveMods(source)) + .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/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 ecaf0aea6ec2f..a9eb48c48c37a 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 @@ -245,21 +245,26 @@ 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(LoadUtil.getTsFileResourcePath(tsFile.getAbsolutePath()))); deleteFile(ModificationFile.getExclusiveMods(tsFile)); deleteFile(new File(LoadUtil.getTsFileModsV1Path(tsFile.getAbsolutePath()))); } - 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(DataNodeQueryMessages.DELETE_AFTER_LOADING_ERROR, 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 ee877628cd992..94f44022ceffe 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 @@ -211,6 +211,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.nio.file.StandardCopyOption; import java.util.ArrayList; import java.util.Arrays; import java.util.BitSet; @@ -283,6 +284,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 @@ -954,6 +956,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 = @@ -995,6 +1005,21 @@ 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(ModificationFileV1.getNormalMods(targetFile).toPath()); + Files.deleteIfExists(ModificationFile.getExclusiveMods(targetFile).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); @@ -4842,75 +4867,85 @@ private boolean loadTsFileToUnSequence( if (!targetFile.getParentFile().exists()) { targetFile.getParentFile().mkdirs(); } - try { - if (deleteOriginFile) { - RetryUtils.retryOnException( - () -> { - FileUtils.moveFile(tsFileToLoad, targetFile); - return null; - }); - } else { - // The staged bytes have to stay for a replica that has not applied this command yet, so - // they cannot be moved. A sealed TsFile is never written again, so a hard link gives this - // region its own path to the very same bytes instead of a second copy of them; the resource - // file below is updated by the load and is still copied. - RetryUtils.retryOnException( - () -> { - LoadUtil.transferFile(tsFileToLoad, targetFile, true); - return null; - }); - } - } catch (final IOException e) { - logger.warn( - StorageEngineMessages - .STORAGE_LOG_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_TARGET_28E43D85, - tsFileToLoad.getAbsolutePath(), - targetFile.getAbsolutePath(), - e); - throw new LoadFileException( - String.format( - StorageEngineMessages - .STORAGE_EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_S_TARGET_37BDA16F, - 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 when hard-link creation falls back to a copy. + 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); + LoadUtil.transferFile(tsFileToLoad, pendingFile, true); return null; }); - } else { + } catch (final IOException e) { + logger.warn( + StorageEngineMessages + .STORAGE_LOG_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_TARGET_28E43D85, + tsFileToLoad.getAbsolutePath(), + targetFile.getAbsolutePath(), + e); + throw new LoadFileException( + String.format( + StorageEngineMessages + .STORAGE_EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_TSFILE_ORIGIN_S_TARGET_37BDA16F, + 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 + .STORAGE_LOG_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_TARGET_9C22DDF3, + resourceFileToLoad.getAbsolutePath(), + targetResourceFile.getAbsolutePath(), + e); + throw new LoadFileException( + String.format( + StorageEngineMessages + .STORAGE_EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_S_9622AA6D, + resourceFileToLoad.getAbsolutePath(), + targetResourceFile.getAbsolutePath(), + e.getMessage())); + } + + // Keep all source files intact until the complete destination is published. + loadModFile(tsFileToLoad, targetFile, false, tsFileResource); + try { + Files.move(pendingFile.toPath(), targetFile.toPath(), StandardCopyOption.ATOMIC_MOVE); + } catch (IOException e) { + throw new LoadFileException(e); + } + loadPublished = true; + } finally { + if (!loadPublished && pendingFile.exists()) { + try { + deleteUnpublishedLoad(pendingFile); + } catch (IOException e) { + logger.warn(StorageEngineMessages.FAILED_TO_DELETE_FILE_OR_DIR, pendingFile, e); + } } - } catch (final IOException e) { - logger.warn( - StorageEngineMessages - .STORAGE_LOG_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_TARGET_9C22DDF3, - resourceFileToLoad.getAbsolutePath(), - targetResourceFile.getAbsolutePath(), - e); - throw new LoadFileException( - String.format( - StorageEngineMessages - .STORAGE_EXCEPTION_FILE_RENAMING_FAILED_WHEN_LOADING_RESOURCE_FILE_ORIGIN_S_9622AA6D, - resourceFileToLoad.getAbsolutePath(), - targetResourceFile.getAbsolutePath(), - e.getMessage())); } - loadModFile(tsFileToLoad, targetFile, deleteOriginFile, tsFileResource); + // Only published mods belong in FileMetrics. Also replace any cached source-side handle. + tsFileResource.setExclusiveModFile(ModificationFile.getExclusiveMods(tsFileResource)); // Listen before the tsFile is added into tsFile manager to avoid it being compacted PipeInsertionDataNodeListener.getInstance() @@ -4918,6 +4953,29 @@ private boolean loadTsFileToUnSequence( tsFileManager.add(tsFileResource, false); + if (deleteOriginFile) { + for (File sourceFile : + new File[] { + tsFileToLoad, + resourceFileToLoad, + ModificationFileV1.getNormalMods(tsFileToLoad), + ModificationFile.getExclusiveMods(tsFileToLoad) + }) { + 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.FAILED_TO_DELETE_FILE_OR_DIR, sourceFile, e); + break; + } + } + } + return true; } @@ -4930,7 +4988,9 @@ private void loadModFile( final File oldTargetModFile = ModificationFileV1.getNormalMods(targetTsFile); moveModFile(oldModFileToLoad, oldTargetModFile, deleteOriginFile); try { - tsFileResource.upgradeModFile(upgradeModFileThreadPool); + // Publishing must wait for a successful conversion; a background failure must not be + // swallowed by getExclusiveModFile and expose an incomplete deletion set after restart. + tsFileResource.upgradeModFile(null, false); } catch (IOException e) { throw new LoadFileException(e); } @@ -4938,8 +4998,6 @@ private void loadModFile( final File newTargetModFile = ModificationFile.getExclusiveMods(targetTsFile); moveModFile(newModFileToLoad, newTargetModFile, deleteOriginFile); } - // force update mod file metrics - tsFileResource.getExclusiveModFile(); } @SuppressWarnings("java:S2139") diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java index 85778719a354b..a431665aa17ca 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/tsfile/TsFileResource.java @@ -1588,21 +1588,33 @@ private void closeModIterator(Iterator modIterator) { } public void upgradeModFile(ExecutorService upgradeModFileThreadPool) throws IOException { + upgradeModFile(upgradeModFileThreadPool, true); + } + + /** Unpublished LOAD installations defer metrics until their data file is published. */ + public void upgradeModFile(ExecutorService upgradeModFileThreadPool, boolean updateModMetrics) + throws IOException { ModificationFileV1 oldModFile = ModificationFileV1.getNormalMods(this); if (!oldModFile.exists()) { return; } if (upgradeModFileThreadPool != null) { - exclusiveModFileFuture = upgradeModFileThreadPool.submit(() -> doUpgradeModFile(oldModFile)); + exclusiveModFileFuture = + upgradeModFileThreadPool.submit(() -> doUpgradeModFile(oldModFile, updateModMetrics)); } else { - exclusiveModFileFuture = CompletableFuture.completedFuture(doUpgradeModFile(oldModFile)); + exclusiveModFileFuture = + CompletableFuture.completedFuture(doUpgradeModFile(oldModFile, updateModMetrics)); } } @SuppressWarnings({"java:S4042", "java:S899", "ResultOfMethodCallIgnored"}) - private ModificationFile doUpgradeModFile(ModificationFileV1 oldModFile) throws IOException { - ModificationFile newMFile = ModificationFile.getExclusiveMods(this); + private ModificationFile doUpgradeModFile(ModificationFileV1 oldModFile, boolean updateModMetrics) + throws IOException { + ModificationFile newMFile = + updateModMetrics + ? ModificationFile.getExclusiveMods(this) + : new ModificationFile(ModificationFile.getExclusiveMods(file), false); FileUtils.deleteFileIfExist(newMFile.getFile()); try { for (Modification oldMod : oldModFile.getModifications()) { 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 238eef0112ff9..2fb23cf51c1ad 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 @@ -96,11 +96,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()); @@ -117,7 +117,37 @@ public void testCleanContinuesAfterOneFileCannotBeDeleted() throws Exception { node.clean(); Assert.assertTrue(tsFile.exists()); - Assert.assertFalse(resourceFile.exists()); + Assert.assertTrue(resourceFile.isFile()); + Assert.assertTrue(modsV2File.isFile()); + Assert.assertTrue(modsV1File.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(LoadUtil.getTsFileResourcePath(tsFile.getAbsolutePath())); + final File modsV2File = ModificationFile.getExclusiveMods(tsFile); + final File modsV1File = new File(LoadUtil.getTsFileModsV1Path(tsFile.getAbsolutePath())); + Assert.assertTrue(tsFile.createNewFile()); + Assert.assertTrue(resourceFile.mkdir()); + final File retained = new File(resourceFile, "non-empty"); + Assert.assertTrue(retained.createNewFile()); + Assert.assertTrue(modsV2File.createNewFile()); + Assert.assertTrue(modsV1File.createNewFile()); + + final LoadSingleTsFileNode node = + new LoadSingleTsFileNode( + new PlanNodeId(""), new TsFileResource(tsFile), false, null, true, 0L, false); + node.clean(); + + Assert.assertFalse(tsFile.exists()); + Assert.assertTrue(resourceFile.isDirectory()); + Assert.assertTrue(retained.isFile()); Assert.assertFalse(modsV2File.exists()); Assert.assertFalse(modsV1File.exists()); } finally { 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 a63e5a5e6917c..250ccc18dadce 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 @@ -42,6 +42,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.queryengine.common.QueryId; import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; @@ -55,6 +56,7 @@ import org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TreeDeviceSchemaCacheManager; import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; +import org.apache.iotdb.db.service.metrics.FileMetrics; import org.apache.iotdb.db.storageengine.StorageEngine; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.ICompactionPerformer; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.performer.impl.FastCompactionPerformer; @@ -77,6 +79,9 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.NOP; +import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry; +import org.apache.iotdb.db.storageengine.dataregion.modification.v1.Deletion; +import org.apache.iotdb.db.storageengine.dataregion.modification.v1.ModificationFileV1; 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; @@ -241,6 +246,188 @@ public static InsertRowNode buildInsertRowNodeByTSRecord(TSRecord record) return insertRowNode; } + @Test + public void testLoadInstallationRetainsOrDeletesTheCompleteSourceGroup() throws Exception { + for (boolean deleteOrigin : new boolean[] {false, true}) { + for (int modsVersion : new int[] {0, 1, 2}) { + final boolean withMods = modsVersion != 0; + final File sourceDir = + new File( + TestConstant.OUTPUT_DATA_DIR, "load-source-" + deleteOrigin + "-" + modsVersion); + 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(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(row, measurementId, row); + } + writer.writeAligned(tablet); + } + final TsFileResource resource = Mockito.spy(new TsFileResource(source)); + resource.setStatus(TsFileResourceStatus.NORMAL); + resource.updateStartTime(device, 1); + resource.updateEndTime(device, 3); + resource.serialize(); + final File sourceMods = + modsVersion == 1 + ? ModificationFileV1.getNormalMods(source) + : ModificationFile.getExclusiveMods(source); + if (modsVersion == 1) { + try (ModificationFileV1 mods = ModificationFileV1.getNormalMods(resource)) { + mods.write( + new Deletion( + new MeasurementPath(deviceId + "." + measurementId), Long.MAX_VALUE, 2, 2)); + } + } else if (withMods) { + try (ModificationFile mods = resource.getExclusiveModFile()) { + mods.write( + new TreeDeletionEntry(new MeasurementPath(deviceId + "." + measurementId), 2, 2)); + } + } + // Exercise rebinding even when the source handle was cached before installation. + resource.getExclusiveModFile(); + 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, Optional.empty())); + 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(ModificationFile.getExclusiveMods(rejectedTarget).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 = FileMetrics.getInstance().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, Optional.empty())); + rejectedTarget = resource.getTsFile(); + Assert.assertTrue(rejectedTarget.isDirectory()); + Assert.assertTrue(new File(rejectedTarget + TsFileResource.RESOURCE_SUFFIX).isFile()); + Assert.assertTrue(ModificationFile.getExclusiveMods(rejectedTarget).isFile()); + Assert.assertFalse(new File(rejectedTarget + ".load-pending").exists()); + Assert.assertEquals(modFilesBeforeLoad, FileMetrics.getInstance().getModFileNum()); + Assert.assertEquals(modSizeBeforeLoad, FileMetrics.getInstance().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(ModificationFileV1.getNormalMods(rejectedTarget).toPath()); + Files.deleteIfExists(ModificationFile.getExclusiveMods(rejectedTarget).toPath()); + } + resource.setFile(source); + } + } + if (modsVersion == 1) { + Mockito.doAnswer( + invocation -> { + invocation.callRealMethod(); + throw new IOException("injected conversion completion failure"); + }) + .when(resource) + .upgradeModFile(Mockito.isNull(), Mockito.anyBoolean()); + try { + final LoadFileException failure = + Assert.assertThrows( + LoadFileException.class, + () -> + dataRegion.loadNewTsFile( + resource, deleteOrigin, false, true, Optional.empty())); + Assert.assertTrue( + failure.getMessage().contains("injected conversion completion failure")); + final File rejectedTarget = resource.getTsFile(); + Assert.assertFalse(rejectedTarget.exists()); + Assert.assertFalse(new File(rejectedTarget + ".load-pending").exists()); + Assert.assertFalse(new File(rejectedTarget + TsFileResource.RESOURCE_SUFFIX).exists()); + Assert.assertFalse(ModificationFileV1.getNormalMods(rejectedTarget).exists()); + Assert.assertFalse(ModificationFile.getExclusiveMods(rejectedTarget).exists()); + Assert.assertEquals(modFilesBeforeLoad, FileMetrics.getInstance().getModFileNum()); + Assert.assertEquals(modSizeBeforeLoad, FileMetrics.getInstance().getModFileSize()); + Assert.assertArrayEquals(data, Files.readAllBytes(source.toPath())); + Assert.assertArrayEquals(deletions, Files.readAllBytes(sourceMods.toPath())); + Assert.assertTrue(sourceResource.isFile()); + } finally { + Mockito.doCallRealMethod() + .when(resource) + .upgradeModFile(Mockito.isNull(), Mockito.anyBoolean()); + resource.setFile(source); + } + } + dataRegion.loadNewTsFile(resource, deleteOrigin, false, true, Optional.empty()); + 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 = ModificationFile.getExclusiveMods(installed); + Assert.assertEquals( + installedMods.getCanonicalFile(), + resource.getExclusiveModFile().getFile().getCanonicalFile()); + Assert.assertEquals(withMods ? 1 : 0, resource.getExclusiveModFile().getAllMods().size()); + Assert.assertEquals(withMods, installedMods.exists()); + Assert.assertFalse(ModificationFileV1.getNormalMods(installed).exists()); + Assert.assertEquals( + modFilesBeforeLoad + (withMods ? 1 : 0), FileMetrics.getInstance().getModFileNum()); + Assert.assertEquals( + modSizeBeforeLoad + installedMods.length(), FileMetrics.getInstance().getModFileSize()); + if (modsVersion == 2) { + 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()); + } + } + } + @Test public void testLoadRateLimiterAccountsForSourceFileBeforeTransfer() throws Exception { final LoadTsFileCostMetricsSet metrics = LoadTsFileCostMetricsSet.getInstance();