diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBDeletionRecoveryIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBDeletionRecoveryIT.java new file mode 100644 index 0000000000000..8b6b710a392a7 --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBDeletionRecoveryIT.java @@ -0,0 +1,171 @@ +/* + * 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.db.it.utils.DeletionTestUtils; +import org.apache.iotdb.db.it.utils.TestUtils; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; + +import org.apache.tsfile.file.metadata.PlainDeviceID; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.List; +import java.util.Map; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +@RunWith(IoTDBTestRunner.class) +@Category(LocalStandaloneIT.class) +public class IoTDBDeletionRecoveryIT { + + @Before + public void setUp() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setWalMode("SYNC") + .setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.simple.SimpleConsensus") + .setDataReplicationFactor(1) + .setDataRegionGroupExtensionPolicy("CUSTOM") + .setDefaultDataRegionGroupNumPerDatabase(1) + .setMemtableSizeThreshold(16 * 1024 * 1024) + .setEnableSeqSpaceCompaction(false) + .setEnableUnseqSpaceCompaction(false) + .setEnableCrossSpaceCompaction(false); + EnvFactory.getEnv().initClusterEnvironment(); + } + + @After + public void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testDeleteAndRewriteSurviveForcedRestart() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + statement.execute("CREATE DATABASE root.restart_delete"); + statement.execute("CREATE TIMESERIES root.restart_delete.d1.s1 WITH DATATYPE=INT64"); + statement.execute("CREATE TIMESERIES root.restart_delete.d1.s2 WITH DATATYPE=INT64"); + statement.execute("CREATE TIMESERIES root.restart_delete.d2.s1 WITH DATATYPE=INT64"); + statement.execute("CREATE TIMESERIES root.restart_delete.d2.s2 WITH DATATYPE=INT64"); + + statement.execute("INSERT INTO root.restart_delete.d1(time,s1,s2) VALUES(1,11,101)"); + statement.execute("INSERT INTO root.restart_delete.d1(time,s1) VALUES(2,12)"); + statement.execute("INSERT INTO root.restart_delete.d2(time,s1,s2) VALUES(2,22,202)"); + statement.execute("FLUSH"); + Path history = + DeletionTestUtils.assertSharedTsFile( + "root.restart_delete", + new PlainDeviceID("root.restart_delete.d1"), + new PlainDeviceID("root.restart_delete.d2")); + assertEquals( + Map.of(1L, 11L, 2L, 12L), + DeletionTestUtils.readTreePoints(history, "root.restart_delete.d1", "s1")); + List closedFiles = DeletionTestUtils.getClosedTsFiles("root.restart_delete", true); + + statement.execute("INSERT INTO root.restart_delete.d1(time,s1) VALUES(1,111)"); + statement.execute("INSERT INTO root.restart_delete.d1(time,s1) VALUES(3,13)"); + statement.execute("INSERT INTO root.restart_delete.d1(time,s1,s2) VALUES(4,14,104)"); + assertEquals(closedFiles, DeletionTestUtils.getClosedTsFiles("root.restart_delete", true)); + assertTrue(DeletionTestUtils.getClosedTsFiles("root.restart_delete", false).isEmpty()); + assertEquals( + Map.of(1L, 11L, 2L, 12L), + DeletionTestUtils.readTreePoints(history, "root.restart_delete.d1", "s1")); + + statement.execute("DELETE FROM root.restart_delete.d1.s1 WHERE time <= 3"); + try (ResultSet resultSet = + statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d1 ORDER BY time")) { + assertTrue(resultSet.next()); + assertEquals(1, resultSet.getLong(1)); + assertNull(resultSet.getObject(2)); + assertEquals(101, resultSet.getLong(3)); + assertTrue(resultSet.next()); + assertEquals(4, resultSet.getLong(1)); + assertEquals(14, resultSet.getLong(2)); + assertEquals(104, resultSet.getLong(3)); + assertFalse(resultSet.next()); + } + + statement.execute("INSERT INTO root.restart_delete.d1(time,s1) VALUES(2,1202)"); + assertEquals(closedFiles, DeletionTestUtils.getClosedTsFiles("root.restart_delete", true)); + assertTrue(DeletionTestUtils.getClosedTsFiles("root.restart_delete", false).isEmpty()); + try (ResultSet rows = + statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d1 ORDER BY time")) { + TestUtils.assertResultSetEqual( + rows, + "Time,root.restart_delete.d1.s1,root.restart_delete.d1.s2,", + new String[] {"1,null,101,", "2,1202,null,", "4,14,104,"}); + } + } + + TestUtils.stopForciblyAndRestartDataNodes(); + + try (Connection connection = EnvFactory.getEnv().getConnection(); + Statement statement = connection.createStatement()) { + try (ResultSet resultSet = + statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d1 ORDER BY time")) { + assertTrue(resultSet.next()); + assertEquals(1, resultSet.getLong(1)); + assertNull(resultSet.getObject(2)); + assertEquals(101, resultSet.getLong(3)); + assertTrue(resultSet.next()); + assertEquals(2, resultSet.getLong(1)); + assertEquals(1202, resultSet.getLong(2)); + assertNull(resultSet.getObject(3)); + assertTrue(resultSet.next()); + assertEquals(4, resultSet.getLong(1)); + assertEquals(14, resultSet.getLong(2)); + assertEquals(104, resultSet.getLong(3)); + assertFalse(resultSet.next()); + } + try (ResultSet resultSet = + statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d2 ORDER BY time")) { + assertTrue(resultSet.next()); + assertEquals(2, resultSet.getLong(1)); + assertEquals(22, resultSet.getLong(2)); + assertEquals(202, resultSet.getLong(3)); + assertFalse(resultSet.next()); + } + + statement.execute("INSERT INTO root.restart_delete.d1(time,s1,s2) VALUES(5,15,105)"); + statement.execute("FLUSH"); + try (ResultSet rows = + statement.executeQuery("SELECT s1,s2 FROM root.restart_delete.d1 ORDER BY time")) { + TestUtils.assertResultSetEqual( + rows, + "Time,root.restart_delete.d1.s1,root.restart_delete.d1.s2,", + new String[] {"1,null,101,", "2,1202,null,", "4,14,104,", "5,15,105,"}); + } + } + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/utils/DeletionTestUtils.java b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/DeletionTestUtils.java new file mode 100644 index 0000000000000..d852d29d6074f --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/DeletionTestUtils.java @@ -0,0 +1,118 @@ +/* + * 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 org.apache.iotdb.commons.conf.IoTDBConstant; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; + +import org.apache.tsfile.file.metadata.ChunkMetadata; +import org.apache.tsfile.file.metadata.IChunkMetadata; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.file.metadata.PlainDeviceID; +import org.apache.tsfile.file.metadata.TimeseriesMetadata; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.BatchData; +import org.apache.tsfile.read.reader.chunk.ChunkReader; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.TreeMap; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import static org.apache.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX; +import static org.junit.Assert.fail; + +/** Physical preconditions for deletion recovery; SQL assertions verify the visible data. */ +public final class DeletionTestUtils { + private DeletionTestUtils() {} + + public static List getClosedTsFiles(String database, boolean sequence) throws IOException { + List result = new ArrayList<>(); + for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) { + Path directory = + Paths.get( + node.getDataPath(), + sequence ? IoTDBConstant.SEQUENCE_FOLDER_NAME : IoTDBConstant.UNSEQUENCE_FOLDER_NAME, + database); + if (Files.exists(directory)) { + try (Stream files = Files.walk(directory)) { + result.addAll( + files + .filter(Files::isRegularFile) + .filter(file -> file.toString().endsWith(TSFILE_SUFFIX)) + .filter(file -> Files.isRegularFile(Paths.get(file + ".resource"))) + .sorted() + .collect(Collectors.toList())); + } + } + } + return result; + } + + public static Path assertSharedTsFile(String database, IDeviceID... devices) throws IOException { + for (Path file : getClosedTsFiles(database, true)) { + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.toString())) { + if (reader.getAllDevices().containsAll(Arrays.asList(devices))) { + return file; + } + } + } + fail("No closed TsFile contains all expected devices: " + Arrays.toString(devices)); + return null; + } + + public static Map readTreePoints(Path file, String device, String measurement) + throws IOException { + Map points = new TreeMap<>(); + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.toString())) { + List series = + reader.getAllTimeseriesMetadata(true).get(new PlainDeviceID(device)); + if (series != null) { + for (TimeseriesMetadata metadata : series) { + if (!measurement.equals(metadata.getMeasurementId())) { + continue; + } + for (IChunkMetadata chunk : metadata.getChunkMetadataList()) { + ChunkReader chunkReader = new ChunkReader(reader.readMemChunk((ChunkMetadata) chunk)); + try { + while (chunkReader.hasNextSatisfiedPage()) { + BatchData page = chunkReader.nextPageData(); + while (page.hasCurrent()) { + points.put(page.currentTime(), page.getLong()); + page.next(); + } + } + } finally { + chunkReader.close(); + } + } + } + } + } + return points; + } +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java index d38954db211f3..cadc96ad7d8e9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoer.java @@ -19,6 +19,7 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.recover.file; +import org.apache.iotdb.commons.exception.IllegalPathException; import org.apache.iotdb.commons.path.PartialPath; import org.apache.iotdb.db.exception.WriteProcessException; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; @@ -32,10 +33,15 @@ import org.apache.iotdb.db.storageengine.dataregion.modification.Deletion; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.tsfile.file.metadata.IDeviceID; + import java.io.IOException; +import java.util.ArrayList; import java.util.List; import java.util.Optional; +import static org.apache.iotdb.commons.conf.IoTDBConstant.ONE_LEVEL_PATH_WILDCARD; + /** * This class helps redo wal logs into a TsFile. Notice: You should update time map in {@link * TsFileResource} before using this class to avoid duplicated insertion and this class doesn't @@ -55,6 +61,7 @@ public TsFilePlanRedoer(TsFileResource tsFileResource) { void redoDelete(DeleteDataNode deleteDataNode) throws IOException { List paths = deleteDataNode.getPathList(); + List deletionEntries = new ArrayList<>(); for (PartialPath path : paths) { // path here is device path pattern recoveryMemTable.delete( @@ -62,15 +69,64 @@ void redoDelete(DeleteDataNode deleteDataNode) throws IOException { path.getDevicePath(), deleteDataNode.getDeleteStartTime(), deleteDataNode.getDeleteEndTime()); - tsFileResource - .getModFile() - .write( - new Deletion( - path, - tsFileResource.getTsFileSize(), - deleteDataNode.getDeleteStartTime(), - deleteDataNode.getDeleteEndTime())); + deletionEntries.addAll( + deletionsForFlushedData( + path, deleteDataNode.getDeleteStartTime(), deleteDataNode.getDeleteEndTime())); + } + if (!deletionEntries.isEmpty()) { + for (Deletion deletion : deletionEntries) { + tsFileResource.getModFile().write(deletion); + } + } + } + + private List deletionsForFlushedData( + PartialPath pathPattern, long deleteStartTime, long deleteEndTime) throws IOException { + // WAL inserts are replayed into memory and flushed after all deletions have been replayed. + // A file-wide deletion would also hide later inserts. The resource still describes only the + // data already on disk, and redoInsert skips that data using each device's end time. + List deletions = new ArrayList<>(); + for (IDeviceID deviceID : tsFileResource.getDevices()) { + Optional startTime = tsFileResource.getStartTime(deviceID); + Optional endTime = tsFileResource.getEndTime(deviceID); + if (!startTime.isPresent() || !endTime.isPresent() || startTime.get() > endTime.get()) { + continue; + } + if (deleteEndTime < startTime.get() || deleteStartTime > endTime.get()) { + continue; + } + + PartialPath devicePath; + PartialPath measurementPath; + try { + devicePath = new PartialPath(deviceID); + } catch (IllegalPathException e) { + throw new IOException(e); + } + boolean affectsDevice = + pathPattern.getDevicePath().matchFullPath(devicePath) + || (pathPattern.endWithMultiLevelWildcard() && pathPattern.matchFullPath(devicePath)); + if (!affectsDevice) { + continue; + } + + String measurement = + pathPattern.endWithMultiLevelWildcard() + ? ONE_LEVEL_PATH_WILDCARD + : pathPattern.getMeasurement(); + try { + measurementPath = new PartialPath(deviceID, measurement); + } catch (IllegalPathException e) { + throw new IOException(e); + } + deletions.add( + new Deletion( + measurementPath, + tsFileResource.getTsFileSize(), + deleteStartTime, + Math.min(deleteEndTime, endTime.get()))); } + return deletions; } void redoInsert(InsertNode node) throws WriteProcessException { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java index acbaaae6620bb..4183136c2d4a5 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/recover/file/TsFilePlanRedoerTest.java @@ -29,6 +29,8 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; 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.modification.Deletion; +import org.apache.iotdb.db.storageengine.dataregion.modification.Modification; import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.TsFileUtilsForRecoverTest; @@ -70,6 +72,8 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; public class TsFilePlanRedoerTest { @@ -599,7 +603,7 @@ public void testRedoDeleteDataNode() throws Exception { DeleteDataNode deleteDataNode = new DeleteDataNode( new PlanNodeId(""), - Collections.singletonList(new PartialPath(DEVICE1_NAME)), + Collections.singletonList(new PartialPath(DEVICE1_NAME, "**")), Long.MIN_VALUE, Long.MAX_VALUE); @@ -611,6 +615,100 @@ public void testRedoDeleteDataNode() throws Exception { assertTrue(modsFile.exists()); } + @Test + public void testRedoDeleteWithoutFlushedDataDoesNotCreateMods() throws Exception { + File file = new File(FILE_NAME); + assertTrue(file.getParentFile().isDirectory() || file.getParentFile().mkdirs()); + tsFileResource = new TsFileResource(file); + TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); + + planRedoer.redoDelete( + new DeleteDataNode( + new PlanNodeId(""), + Collections.singletonList(new PartialPath(SG_NAME + ".**")), + Long.MIN_VALUE, + Long.MAX_VALUE)); + + assertFalse(ModificationFile.getNormalMods(tsFileResource).exists()); + } + + @Test + public void testRedoTreeDeleteIsBoundedByFlushedDevicesAndFields() throws Exception { + File file = new File(FILE_NAME); + generateCompleteFile(file); + tsFileResource = new TsFileResource(file); + tsFileResource.updateStartTime(DEVICE1_NAME, 1); + tsFileResource.updateEndTime(DEVICE1_NAME, 2); + tsFileResource.updateStartTime(DEVICE2_NAME, 3); + tsFileResource.updateEndTime(DEVICE2_NAME, 4); + IDeviceID nestedDevice = new PlainDeviceID(SG_NAME + ".d1.child"); + tsFileResource.updateStartTime(nestedDevice, 1); + tsFileResource.updateEndTime(nestedDevice, 1); + TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource); + + planRedoer.redoDelete( + new DeleteDataNode( + new PlanNodeId(""), + Collections.singletonList(new PartialPath(SG_NAME + ".**.s1")), + Long.MIN_VALUE, + Long.MAX_VALUE)); + + List deletions = getWrittenDeletions(); + assertEquals(3, deletions.size()); + assertDeletion(deletions, DEVICE1_NAME, "s1", 2); + assertDeletion(deletions, DEVICE2_NAME, "s1", 4); + assertDeletion(deletions, nestedDevice, "s1", 1); + assertNull(findDeletion(deletions, DEVICE1_NAME, "s2")); + assertNull(findDeletion(deletions, DEVICE2_NAME, "s2")); + assertNull(findDeletion(deletions, nestedDevice, "s2")); + assertNull(findDeletion(deletions, DEVICE3_NAME, "s1")); + assertNull(findDeletion(deletions, DEVICE3_NAME, "s2")); + + planRedoer.redoDelete( + new DeleteDataNode( + new PlanNodeId(""), + Collections.singletonList(new PartialPath(SG_NAME + ".**")), + Long.MIN_VALUE, + Long.MAX_VALUE)); + + List allDeletions = getWrittenDeletions(); + assertEquals(6, allDeletions.size()); + Deletion nestedFieldDeletion = findDeletion(allDeletions.subList(3, 6), nestedDevice, "s2"); + assertNotNull(nestedFieldDeletion); + assertEquals(1, nestedFieldDeletion.getEndTime()); + IDeviceID unflushedDevice = new PlainDeviceID(SG_NAME + ".d1.unflushed"); + assertNull(findDeletion(allDeletions, unflushedDevice, "s1")); + } + + private void assertDeletion( + List deletions, IDeviceID deviceID, String measurement, long endTime) + throws Exception { + Deletion deletion = findDeletion(deletions, deviceID, measurement); + assertNotNull(deletion); + assertEquals(Long.MIN_VALUE, deletion.getStartTime()); + assertEquals(endTime, deletion.getEndTime()); + } + + private Deletion findDeletion(List deletions, IDeviceID deviceID, String measurement) + throws Exception { + PartialPath seriesPath = new PartialPath(deviceID, measurement); + for (Deletion deletion : deletions) { + if (deletion.getPath().matchFullPath(seriesPath)) { + return deletion; + } + } + return null; + } + + private List getWrittenDeletions() { + List deletions = new ArrayList<>(); + for (Modification modification : tsFileResource.getModFile().getModifications()) { + assertTrue(modification instanceof Deletion); + deletions.add((Deletion) modification); + } + return deletions; + } + @Test public void testRedoAlignedInsertAfterDeleteTimeseries() throws Exception { // generate .tsfile and update resource in memory