[To dev/1.3] Fix tree deletion recovery during WAL replay (#19081)

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<Path> 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,"});
+      }
+    }
+  }
+}

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<Path> getClosedTsFiles(String database, boolean sequence) throws IOException {
+    List<Path> 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<Path> 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<Long, Long> readTreePoints(Path file, String device, String measurement)
+      throws IOException {
+    Map<Long, Long> points = new TreeMap<>();
+    try (TsFileSequenceReader reader = new TsFileSequenceReader(file.toString())) {
+      List<TimeseriesMetadata> 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;
+  }
+}

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,22 +61,72 @@
 
   void redoDelete(DeleteDataNode deleteDataNode) throws IOException {
     List<PartialPath> paths = deleteDataNode.getPathList();
+    List<Deletion> deletionEntries = new ArrayList<>();
     for (PartialPath path : paths) {
       // path here is device path pattern
       recoveryMemTable.delete(
           path,
           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<Deletion> 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<Deletion> deletions = new ArrayList<>();
+    for (IDeviceID deviceID : tsFileResource.getDevices()) {
+      Optional<Long> startTime = tsFileResource.getStartTime(deviceID);
+      Optional<Long> 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 {