diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestartIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestartIT.java
index 571826307c667..302d059959a5a 100644
--- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestartIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBRestartIT.java
@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
+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;
@@ -30,6 +31,7 @@
 
 import org.apache.tsfile.enums.TSDataType;
 import org.apache.tsfile.external.commons.io.FileUtils;
+import org.apache.tsfile.file.metadata.IDeviceID;
 import org.apache.tsfile.write.schema.MeasurementSchema;
 import org.junit.After;
 import org.junit.Before;
@@ -41,16 +43,21 @@
 
 import java.io.File;
 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.Collections;
+import java.util.List;
+import java.util.Map;
 
 import static org.apache.iotdb.db.utils.constant.TestConstant.TIMESTAMP_STR;
 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;
 import static org.junit.Assert.fail;
 
 @RunWith(IoTDBTestRunner.class)
@@ -61,7 +68,18 @@ public class IoTDBRestartIT {
 
   @Before
   public void setUp() throws Exception {
-    EnvFactory.getEnv().getConfig().getCommonConfig().setWalMode("SYNC");
+    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()
         .getConfig()
         .getCommonConfig()
@@ -165,6 +183,110 @@ public void testRestartDelete() throws SQLException {
     }
   }
 
+  @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");
+
+      // The first two rows are in a TsFile. The remaining rows are only protected by the SYNC
+      // WAL when the delete is issued.
+      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",
+              IDeviceID.Factory.DEFAULT_FACTORY.create("root.restart_delete.d1"),
+              IDeviceID.Factory.DEFAULT_FACTORY.create("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());
+      }
+
+      // This rewrite is deliberately older than the surviving point. It must survive the old
+      // delete record and remain visible after recovery, while s2 stays NULL.
+      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());
+      }
+      // Only after checking both the target and control do we test continued writes and flush.
+      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,"});
+      }
+    }
+  }
+
   @Test
   public void testRestartQueryLargerThanEndTime() throws SQLException {
     try (Connection connection = EnvFactory.getEnv().getConnection();
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..f8e94848d74be
--- /dev/null
+++ b/integration-test/src/test/java/org/apache/iotdb/db/it/utils/DeletionTestUtils.java
@@ -0,0 +1,119 @@
+/*
+ * 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.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 ITs; SQL assertions still 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;
+  }
+
+  // Read the physical file without applying mods or merging it with newer files.
+  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(IDeviceID.Factory.DEFAULT_FACTORY.create(device));
+      if (series != null) {
+        for (TimeseriesMetadata metadata : series) {
+          if (!measurement.equals(metadata.getMeasurementId())) {
+            continue;
+          }
+          for (IChunkMetadata chunk : metadata.getChunkMetadataList()) {
+            try (ChunkReader chunkReader =
+                new ChunkReader(reader.readMemChunk((ChunkMetadata) chunk))) {
+              while (chunkReader.hasNextSatisfiedPage()) {
+                BatchData page = chunkReader.nextPageData();
+                while (page.hasCurrent()) {
+                  points.put(page.currentTime(), page.getLong());
+                  page.next();
+                }
+              }
+            }
+          }
+        }
+      }
+    }
+    return points;
+  }
+}
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
index 1ff1beebec241..0f29879caad79 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java
@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.db.it.utils.TestUtils;
 import org.apache.iotdb.isession.ITableSession;
+import org.apache.iotdb.isession.SessionConfig;
 import org.apache.iotdb.isession.SessionDataSet;
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
@@ -119,6 +120,9 @@ public static void setUpClass() {
     EnvFactory.getEnv()
         .getConfig()
         .getCommonConfig()
+        .setDataRegionConsensusProtocolClass("org.apache.iotdb.consensus.simple.SimpleConsensus")
+        .setDataReplicationFactor(1)
+        .setEnableLastCache(true)
         .setPartitionInterval(1000)
         .setMemtableSizeThreshold(10000);
     // Adjust MemTable threshold size to make it flush automatically
@@ -1086,24 +1090,465 @@ public void testRangeDelete() throws SQLException {
 
   @Test
   public void testSuccessfullyInvalidateCache() throws SQLException {
-    prepareData(4, 1);
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
         Statement statement = connection.createStatement()) {
       statement.execute("use test");
-      statement.executeQuery(
-          "SELECT last(time), last_by(s0,time), last_by(s1,time), last_by(s2,time), last_by(s3,time), last_by(s4,time) FROM vehicle4 where deviceId = 'd0'");
+      statement.execute(
+          "CREATE TABLE last_delete_cache(device_id STRING TAG, s1 INT32 FIELD, s2 INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO last_delete_cache(time,device_id,s1,s2) VALUES(1,'d0',11,101)");
+      statement.execute("INSERT INTO last_delete_cache(time,device_id,s1) VALUES(2,'d0',12)");
+      statement.execute("INSERT INTO last_delete_cache(time,device_id,s2) VALUES(3,'d0',103)");
+      statement.execute("INSERT INTO last_delete_cache(time,device_id,s1) VALUES(4,'d0',14)");
+      statement.execute("FLUSH");
 
-      // [1, 400] -> [1, 299]
-      statement.execute("DELETE FROM vehicle4 WHERE time >= 300");
-      try (ResultSet set = statement.executeQuery("SELECT s0 FROM vehicle4")) {
-        int cnt = 0;
-        while (set.next()) {
-          cnt++;
+      assertLastValuesOnAllCoordinators(4L, 14, 103, 14, null, true);
+      assertLastCacheRows(
+          "1970-01-01T00:00:00.001Z,d0,11,101,",
+          "1970-01-01T00:00:00.002Z,d0,12,null,",
+          "1970-01-01T00:00:00.003Z,d0,null,103,",
+          "1970-01-01T00:00:00.004Z,d0,14,null,");
+
+      statement.execute("DELETE FROM last_delete_cache WHERE device_id='d0' AND time >= 4");
+      // The sparse fields must fall back independently after their common latest row is deleted.
+      // last ignores NULLs; last_by returns the value (including NULL) at the latest row.
+      assertLastValuesOnAllCoordinators(3L, 12, 103, null, 103);
+      assertLastCacheRows(
+          "1970-01-01T00:00:00.001Z,d0,11,101,",
+          "1970-01-01T00:00:00.002Z,d0,12,null,",
+          "1970-01-01T00:00:00.003Z,d0,null,103,");
+
+      statement.execute("DELETE FROM last_delete_cache WHERE device_id='d0' AND time = 1");
+      assertLastValuesOnAllCoordinators(3L, 12, 103, null, 103);
+      assertLastCacheRows(
+          "1970-01-01T00:00:00.002Z,d0,12,null,", "1970-01-01T00:00:00.003Z,d0,null,103,");
+
+      statement.execute("DELETE FROM last_delete_cache WHERE device_id='d0'");
+      assertLastValuesOnAllCoordinators(null, null, null, null, null);
+      assertLastCacheRows();
+
+      // Rewriting an older timestamp after the full delete must become the new last value; the
+      // deleted timestamp 4 must not reappear from the cache or the flushed file.
+      statement.execute("INSERT INTO last_delete_cache(time,device_id,s1) VALUES(2,'d0',120)");
+      assertLastCacheRows("1970-01-01T00:00:00.002Z,d0,120,null,");
+      assertLastValuesOnAllCoordinators(2L, 120, null, 120, null);
+      statement.execute("FLUSH");
+      assertLastValuesOnAllCoordinators(2L, 120, null, 120, null);
+      assertLastCacheRows("1970-01-01T00:00:00.002Z,d0,120,null,");
+    }
+  }
+
+  @Test
+  public void testRewrittenLastCacheWithTimeFilter() throws SQLException {
+    String query =
+        "SELECT last(time),last(s1),last(s2),last(device_id),last(label),last_by(time,s1) "
+            + "FROM last_filter_cache WHERE device_id='d0'";
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute(
+          "CREATE TABLE last_filter_cache(device_id STRING TAG, label STRING ATTRIBUTE, s1 INT32 FIELD, s2 INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO last_filter_cache(time,device_id,label,s1) VALUES(4,'d0','blue',14)");
+      statement.execute("DELETE FROM last_filter_cache WHERE device_id='d0'");
+    }
+    for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) {
+      try (Connection connection =
+              EnvFactory.getEnv()
+                  .getConnection(
+                      node,
+                      SessionConfig.DEFAULT_USER,
+                      SessionConfig.DEFAULT_PASSWORD,
+                      BaseEnv.TABLE_SQL_DIALECT);
+          Statement statement = connection.createStatement()) {
+        statement.execute("USE test");
+        assertFilteredLastValues(statement, query, true);
+      }
+    }
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute("INSERT INTO last_filter_cache(time,device_id,s1) VALUES(2,'d0',120)");
+    }
+    for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) {
+      try (Connection connection =
+              EnvFactory.getEnv()
+                  .getConnection(
+                      node,
+                      SessionConfig.DEFAULT_USER,
+                      SessionConfig.DEFAULT_PASSWORD,
+                      BaseEnv.TABLE_SQL_DIALECT);
+          Statement statement = connection.createStatement()) {
+        statement.execute("USE test");
+        assertFilteredLastValues(statement, query, false);
+        assertFilteredLastValues(statement, query + " AND time > 2", true);
+        assertFilteredLastValues(statement, query + " AND time >= 3", true);
+        try (ResultSet rows = statement.executeQuery(query + " AND time > 2 GROUP BY device_id")) {
+          assertFalse(rows.next());
+        }
+        assertFilteredLastValues(statement, query + " AND time >= 2", false);
+        assertFilteredLastValues(statement, query + " AND time < 2", true);
+        assertFilteredLastValues(statement, query, false);
+        assertFilteredLastValues(statement, query + " AND time >= 2 GROUP BY device_id", false);
+      }
+    }
+  }
+
+  @Test
+  public void testSparseLastCacheWithoutTime() throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute(
+          "CREATE TABLE last_sparse_cache(device_id STRING TAG, s1 INT32 FIELD, s2 INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO last_sparse_cache(time,device_id,s1) VALUES(3,'d0',30),(1,'d1',10)");
+    }
+
+    // Multiple devices require partial aggregation. An empty s2 cache must not discard s1.
+    assertCachedLastQueryOnAllCoordinators(
+        "SELECT last(s1) AS v1,last(s2) AS v2 FROM last_sparse_cache", "v1,v2,", "30,null,");
+  }
+
+  @Test
+  public void testNullTagAndAttributeDoNotCacheDeviceAsEmpty() throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute(
+          "CREATE TABLE last_null_identity_cache(device_id STRING TAG, site STRING TAG, "
+              + "label STRING ATTRIBUTE, s1 INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO last_null_identity_cache(time,device_id,site,s1) "
+              + "VALUES(3,'d0',null,30),(1,'d1','',10)");
+    }
+    // Populate the FIELD cache without initializing the time cache, then read NULL identities.
+    assertCachedLastQueryOnAllCoordinators(
+        "SELECT last(s1) AS v FROM last_null_identity_cache", "v,", "30,");
+    assertCachedLastQueryOnAllCoordinators(
+        "SELECT last(site) AS site,last(label) AS label "
+            + "FROM last_null_identity_cache WHERE device_id='d0'",
+        "site,label,",
+        "null,null,");
+    String query =
+        "SELECT last(time) AS t,last(s1) AS v,last(site) AS site,last(label) AS label "
+            + "FROM last_null_identity_cache WHERE device_id='d0'";
+    assertCachedLastQueryOnAllCoordinators(
+        query, "t,v,site,label,", "1970-01-01T00:00:00.003Z,30,null,null,");
+    assertCachedLastQueryOnAllCoordinators(
+        query + " GROUP BY device_id", "t,v,site,label,", "1970-01-01T00:00:00.003Z,30,null,null,");
+    assertCachedLastQueryOnAllCoordinators(
+        "SELECT last(time) AS t,last(s1) AS v,last(site) AS site,last(label) AS label "
+            + "FROM last_null_identity_cache WHERE device_id='d1' GROUP BY device_id",
+        "t,v,site,label,",
+        "1970-01-01T00:00:00.001Z,10,,null,");
+  }
+
+  @Test
+  public void testLastByTagAndAttributeUseSourceField() throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute(
+          "CREATE TABLE last_by_source_cache(device_id STRING TAG, site STRING TAG, "
+              + "label STRING ATTRIBUTE, s1 INT32 FIELD, s2 INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO last_by_source_cache(time,device_id,site,label,s1) "
+              + "VALUES(10,'d0',null,'blue',30),(4,'d1','','green',40)");
+    }
+
+    String emptySourceQuery =
+        "SELECT last(time) AS t,last_by(device_id,s2) AS d,last_by(label,s2) AS a,"
+            + "last_by(time,s2) AS st FROM last_by_source_cache WHERE device_id='d0'";
+    assertCachedLastQueryOnAllCoordinators(
+        emptySourceQuery, "t,d,a,st,", "1970-01-01T00:00:00.010Z,null,null,null,");
+    assertCachedLastQueryOnAllCoordinators(
+        emptySourceQuery + " GROUP BY device_id",
+        "t,d,a,st,",
+        "1970-01-01T00:00:00.010Z,null,null,null,");
+
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute(
+          "INSERT INTO last_by_source_cache(time,device_id,site,s2) "
+              + "VALUES(2,'d0',null,20),(3,'d1','',15)");
+    }
+    // s1 and s2 have different last times. The partial states must carry the source field's time,
+    // including when the selected TAG is NULL or an empty string.
+    assertCachedLastQueryOnAllCoordinators(
+        "SELECT last(s1) AS v,last_by(device_id,s2) AS d,last_by(label,s2) AS a,"
+            + "last_by(site,s2) AS site,last_by(site,s1) AS null_site "
+            + "FROM last_by_source_cache",
+        "v,d,a,site,null_site,",
+        "30,d1,green,,null,");
+  }
+
+  @Test
+  public void testFilteredLastCacheMissDoesNotHideOlderRows() throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute(
+          "CREATE TABLE last_filter_miss_cache(device_id STRING TAG, s1 INT32 FIELD, s2 INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO last_filter_miss_cache(time,device_id,s1) VALUES(2,'d0',120),(2,'d1',120)");
+    }
+    String[] queries = {
+      "SELECT last(time) AS t,last(s1) AS v1,last(s2) AS v2 "
+          + "FROM last_filter_miss_cache WHERE device_id='d0'",
+      "SELECT last(time) AS t,last_by(s1,time) AS v1,last_by(s2,time) AS v2 "
+          + "FROM last_filter_miss_cache WHERE device_id='d1'"
+    };
+    // Start with cache misses in both LAST_VALUES and LAST_ROW paths. Empty filtered reads must
+    // leave the older rows available to subsequent unfiltered and grouped queries.
+    for (String query : queries) {
+      assertCachedLastQueryOnAllCoordinators(
+          query + " AND time > 2", "t,v1,v2,", "null,null,null,");
+      assertCachedLastQueryOnAllCoordinators(
+          query, "t,v1,v2,", "1970-01-01T00:00:00.002Z,120,null,");
+      assertCachedLastQueryOnAllCoordinators(
+          query + " GROUP BY device_id", "t,v1,v2,", "1970-01-01T00:00:00.002Z,120,null,");
+    }
+  }
+
+  @Test
+  public void testLastCacheRespectsTableTTL() throws SQLException {
+    String query = "SELECT last(s1) AS v FROM last_ttl_cache WHERE device_id='d0'";
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute("CREATE TABLE last_ttl_cache(device_id STRING TAG, s1 INT32 FIELD)");
+      statement.execute("INSERT INTO last_ttl_cache(time,device_id,s1) VALUES(1,'d0',11)");
+    }
+    assertCachedLastQueryOnAllCoordinators(query, "v,", "11,");
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute("ALTER TABLE last_ttl_cache SET PROPERTIES TTL=1");
+    }
+    assertCachedLastQueryOnAllCoordinators(query, "v,", "null,");
+    assertCachedLastQueryOnAllCoordinators(query + " AND time >= 1", "v,", "null,");
+    assertCachedLastQueryOnAllCoordinators(query + " GROUP BY device_id", "v,");
+  }
+
+  @Test
+  public void testExpiredLastCacheMissDoesNotHideRowsAfterTTLIsExtended() throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute(
+          "CREATE TABLE last_ttl_miss_cache(device_id STRING TAG, s1 INT32 FIELD, s2 INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO last_ttl_miss_cache(time,device_id,s1) VALUES(2,'d0',120),(2,'d1',120)");
+      statement.execute("ALTER TABLE last_ttl_miss_cache SET PROPERTIES TTL=1");
+    }
+    String[] queries = {
+      "SELECT last(time) AS t,last(s1) AS v1,last(s2) AS v2 "
+          + "FROM last_ttl_miss_cache WHERE device_id='d0'",
+      "SELECT last(time) AS t,last_by(s1,time) AS v1,last_by(s2,time) AS v2 "
+          + "FROM last_ttl_miss_cache WHERE device_id='d1'"
+    };
+    // The first reads miss both LAST_VALUES and LAST_ROW caches while TTL hides the rows.
+    for (String query : queries) {
+      assertCachedLastQueryOnAllCoordinators(query, "t,v1,v2,", "null,null,null,");
+      assertCachedLastQueryOnAllCoordinators(query + " GROUP BY device_id", "t,v1,v2,");
+    }
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute("ALTER TABLE last_ttl_miss_cache SET PROPERTIES TTL='INF'");
+      // The rows remain in memory; extending TTL makes them visible without another write.
+      try (ResultSet rows =
+          statement.executeQuery(
+              "SELECT time,device_id,s1,s2 FROM last_ttl_miss_cache ORDER BY device_id,time")) {
+        TestUtils.assertResultSetEqual(
+            rows,
+            "time,device_id,s1,s2,",
+            new String[] {
+              "1970-01-01T00:00:00.002Z,d0,120,null,", "1970-01-01T00:00:00.002Z,d1,120,null,"
+            });
+      }
+    }
+    // Read LAST_VALUES before LAST_ROW can scan again and repair its cached empty entries.
+    String lastValuesAfterLastRowMiss =
+        "SELECT last(time) AS t,last(s1) AS v1,last(s2) AS v2 "
+            + "FROM last_ttl_miss_cache WHERE device_id='d1'";
+    assertCachedLastQueryOnAllCoordinators(
+        lastValuesAfterLastRowMiss, "t,v1,v2,", "1970-01-01T00:00:00.002Z,120,null,");
+    assertCachedLastQueryOnAllCoordinators(
+        lastValuesAfterLastRowMiss + " GROUP BY device_id",
+        "t,v1,v2,",
+        "1970-01-01T00:00:00.002Z,120,null,");
+    for (String query : queries) {
+      assertCachedLastQueryOnAllCoordinators(
+          query, "t,v1,v2,", "1970-01-01T00:00:00.002Z,120,null,");
+      assertCachedLastQueryOnAllCoordinators(
+          query + " GROUP BY device_id", "t,v1,v2,", "1970-01-01T00:00:00.002Z,120,null,");
+    }
+  }
+
+  @Test
+  public void testLastByTimeOverDeletedDeviceDoesNotBreakLastCache() throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute(
+          "CREATE TABLE last_by_time_empty_cache(device_id STRING TAG, s1 INT32 FIELD, s2 INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO last_by_time_empty_cache(time,device_id,s1) VALUES(4,'d0',40),(1,'d1',10)");
+      statement.execute("DELETE FROM last_by_time_empty_cache WHERE device_id='d0'");
+    }
+    String query =
+        "SELECT last_by(time,time) AS t FROM last_by_time_empty_cache WHERE device_id='d0'";
+    String withFields =
+        "SELECT last_by(time,time) AS t,last_by(s1,time) AS v1,last_by(s2,time) AS v2 "
+            + "FROM last_by_time_empty_cache WHERE device_id='d0'";
+    // The device remains in schema after DELETE. An uninitialized time accumulator must still
+    // produce matching cache-column and cache-value arrays on every cold read.
+    assertCachedLastQueryOnAllCoordinators(query, "t,", "null,");
+    assertCachedLastQueryOnAllCoordinators(query + " GROUP BY device_id", "t,");
+    assertCachedLastQueryOnAllCoordinators(withFields, "t,v1,v2,", "null,null,null,");
+    assertCachedLastQueryOnAllCoordinators(withFields + " GROUP BY device_id", "t,v1,v2,");
+    assertCachedLastQueryOnAllCoordinators(
+        "SELECT last_by(time,time) AS t FROM last_by_time_empty_cache",
+        "t,",
+        "1970-01-01T00:00:00.001Z,");
+
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("USE test");
+      statement.execute(
+          "INSERT INTO last_by_time_empty_cache(time,device_id,s1) VALUES(2,'d0',20)");
+    }
+    assertCachedLastQueryOnAllCoordinators(query, "t,", "1970-01-01T00:00:00.002Z,");
+    assertCachedLastQueryOnAllCoordinators(
+        withFields, "t,v1,v2,", "1970-01-01T00:00:00.002Z,20,null,");
+    assertCachedLastQueryOnAllCoordinators(
+        withFields + " GROUP BY device_id", "t,v1,v2,", "1970-01-01T00:00:00.002Z,20,null,");
+    assertCachedLastQueryOnAllCoordinators(
+        "SELECT last(time) AS t,last(s1) AS v1,last(s2) AS v2 "
+            + "FROM last_by_time_empty_cache WHERE device_id='d0'",
+        "t,v1,v2,",
+        "1970-01-01T00:00:00.002Z,20,null,");
+  }
+
+  private static void assertCachedLastQueryOnAllCoordinators(
+      String query, String header, String... expected) throws SQLException {
+    for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) {
+      try (Connection connection =
+              EnvFactory.getEnv()
+                  .getConnection(
+                      node,
+                      SessionConfig.DEFAULT_USER,
+                      SessionConfig.DEFAULT_PASSWORD,
+                      BaseEnv.TABLE_SQL_DIALECT);
+          Statement statement = connection.createStatement()) {
+        statement.execute("USE test");
+        // The first read may populate the cache; the second must have the same result.
+        for (int i = 0; i < 2; i++) {
+          try (ResultSet rows = statement.executeQuery(query)) {
+            TestUtils.assertResultSetEqual(rows, header, expected);
+          }
+        }
+      }
+    }
+  }
+
+  private static void assertFilteredLastValues(Statement statement, String query, boolean empty)
+      throws SQLException {
+    Object[] expected = {2L, 120L, null, "d0", "blue", 2L};
+    try (ResultSet rows = statement.executeQuery(query)) {
+      assertTrue(query, rows.next());
+      for (int i = 0; i < expected.length; i++) {
+        if (empty || expected[i] == null) {
+          assertNull(query, rows.getObject(i + 1));
+        } else if (expected[i] instanceof Long) {
+          assertEquals(query, (long) expected[i], rows.getLong(i + 1));
+          assertFalse(query, rows.wasNull());
+        } else {
+          assertEquals(query, expected[i], rows.getString(i + 1));
+        }
+      }
+      assertFalse(query, rows.next());
+    }
+  }
+
+  private static void assertLastCacheRows(String... expected) {
+    TestUtils.tableResultSetEqualTest(
+        "SELECT time,device_id,s1,s2 FROM last_delete_cache ORDER BY device_id,time",
+        new String[] {"time", "device_id", "s1", "s2"},
+        expected,
+        "test");
+  }
+
+  private static void assertLastValuesOnAllCoordinators(
+      Long expectedTime, Integer s1, Integer s2, Integer byS1, Integer byS2) throws SQLException {
+    assertLastValuesOnAllCoordinators(expectedTime, s1, s2, byS1, byS2, false);
+  }
+
+  private static void assertLastValuesOnAllCoordinators(
+      Long expectedTime,
+      Integer s1,
+      Integer s2,
+      Integer byS1,
+      Integer byS2,
+      boolean verifyCachePath)
+      throws SQLException {
+    // Mixing last(FIELD) with last_by(FIELD,time) disables both last-cache optimizations.
+    String[] queries = {
+      "SELECT last(time),last(s1),last(s2) FROM last_delete_cache WHERE device_id='d0'",
+      "SELECT last(time),last_by(s1,time),last_by(s2,time) "
+          + "FROM last_delete_cache WHERE device_id='d0'"
+    };
+    Number[][] expected = {{expectedTime, s1, s2}, {expectedTime, byS1, byS2}};
+    for (DataNodeWrapper node : EnvFactory.getEnv().getDataNodeWrapperList()) {
+      // Direct connections keep each read on the coordinator whose cache was warmed.
+      try (Connection connection =
+              EnvFactory.getEnv()
+                  .getConnection(
+                      node,
+                      SessionConfig.DEFAULT_USER,
+                      SessionConfig.DEFAULT_PASSWORD,
+                      BaseEnv.TABLE_SQL_DIALECT);
+          Statement statement = connection.createStatement()) {
+        statement.execute("USE test");
+        for (int query = 0; query < queries.length; query++) {
+          String context = node.getIpAndPortString() + ": " + queries[query];
+          try (ResultSet rows = statement.executeQuery(queries[query])) {
+            assertTrue(context, rows.next());
+            for (int column = 0; column < expected[query].length; column++) {
+              String columnContext = context + ", column " + (column + 1);
+              if (expected[query][column] == null) {
+                assertNull(columnContext, rows.getObject(column + 1));
+              } else {
+                long actual = rows.getLong(column + 1);
+                assertFalse(columnContext + " is NULL", rows.wasNull());
+                assertEquals(columnContext, expected[query][column].longValue(), actual);
+              }
+            }
+            assertFalse(context, rows.next());
+          }
+          if (verifyCachePath) {
+            StringBuilder plan = new StringBuilder();
+            try (ResultSet rows = statement.executeQuery("EXPLAIN ANALYZE " + queries[query])) {
+              while (rows.next()) {
+                plan.append(rows.getString(1)).append('\n');
+              }
+            }
+            assertTrue(plan.toString(), plan.toString().contains("LastQueryAggTableScanOperator"));
+          }
+        }
+        if (expectedTime == null) {
+          try (ResultSet rows =
+              statement.executeQuery(
+                  "SELECT device_id,last(time),last_by(s1,time),last_by(s2,time) "
+                      + "FROM last_delete_cache WHERE device_id='d0' GROUP BY device_id")) {
+            assertFalse(rows.next());
+          }
         }
-        assertEquals(299, cnt);
       }
     }
-    cleanData(4);
   }
 
   @Test
diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java
index 127e998cb5dcb..d24c4737c25e5 100644
--- a/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBRestartTableIT.java
@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.db.conf.IoTDBConfig;
 import org.apache.iotdb.db.conf.IoTDBDescriptor;
+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;
@@ -28,6 +29,7 @@
 import org.apache.iotdb.itbase.category.TableLocalStandaloneIT;
 import org.apache.iotdb.itbase.env.BaseEnv;
 
+import org.apache.tsfile.file.metadata.IDeviceID;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
@@ -37,15 +39,20 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.nio.file.Path;
 import java.sql.Connection;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 
 import static org.apache.iotdb.db.utils.constant.TestConstant.TIMESTAMP_STR;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 @RunWith(IoTDBTestRunner.class)
@@ -56,7 +63,18 @@ public class IoTDBRestartTableIT {
 
   @Before
   public void setUp() throws Exception {
-    EnvFactory.getEnv().getConfig().getCommonConfig().setWalMode("SYNC");
+    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()
         .getConfig()
         .getCommonConfig()
@@ -121,50 +139,119 @@ public void testRestart() throws SQLException {
     }
   }
 
-  @Ignore // data deletion
   @Test
-  public void testRestartDelete() throws SQLException {
+  public void testRestartDelete() throws Exception {
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
         Statement statement = connection.createStatement()) {
+      connection.setClientInfo("time_zone", "+00:00");
+      statement.execute("create database test");
       statement.execute("use \"test\"");
-      statement.execute("insert into root.turbine.d1(time,s1) values(1,1)");
-      statement.execute("insert into root.turbine.d1(time,s1) values(2,2)");
-      statement.execute("insert into root.turbine.d1(time,s1) values(3,3)");
+      statement.execute(
+          "create table restart_delete(device_id string tag, site string tag, "
+              + "s1 int64 field, s2 boolean field)");
+      statement.execute(
+          "create table restart_keep(device_id string tag, site string tag, "
+              + "s1 int64 field, s2 boolean field)");
+      statement.execute(
+          "insert into restart_delete(time,device_id,site,s1,s2) "
+              + "values(1,'d1','north',11,true),(2,'d1','north',12,false),"
+              + "(2,'d1','south',212,true),(2,'d1',null,52,true),(2,'d1','',62,false)");
+      statement.execute(
+          "insert into restart_keep(time,device_id,site,s1,s2) "
+              + "values(1,'d1','north',311,true),(2,'d1','north',312,false)");
+      statement.execute("flush");
+      // Identical TAG text in two tables, and NULL/empty trailing TAGs, coexist in one file.
+      DeletionTestUtils.assertSharedTsFile(
+          "test",
+          IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"restart_delete", "d1", "north"}),
+          IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"restart_delete", "d1", "south"}),
+          IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"restart_delete", "d1", null}),
+          IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"restart_delete", "d1", ""}),
+          IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"restart_keep", "d1", "north"}));
+      List<Path> closedFiles = DeletionTestUtils.getClosedTsFiles("test", true);
+      statement.execute(
+          "insert into restart_delete(time,device_id,site,s1) values(1,'d1','north',111)");
+      statement.execute(
+          "insert into restart_delete(time,device_id,site,s1,s2) "
+              + "values(3,'d1','north',13,true),(4,'d1','north',14,false)");
+      assertEquals(closedFiles, DeletionTestUtils.getClosedTsFiles("test", true));
+      assertTrue(DeletionTestUtils.getClosedTsFiles("test", false).isEmpty());
+
+      statement.execute(
+          "delete from restart_delete where device_id='d1' and site='north' and time <= 3");
+      assertRestartTableData(statement, "1970-01-01T00:00:00.004Z,d1,north,14,false,");
+
+      // This older rewrite is issued after DELETE has completed and is protected by the SYNC WAL.
+      statement.execute(
+          "insert into restart_delete(time,device_id,site,s1) values(2,'d1','north',1202)");
+      assertEquals(closedFiles, DeletionTestUtils.getClosedTsFiles("test", true));
+      assertTrue(DeletionTestUtils.getClosedTsFiles("test", false).isEmpty());
+      assertRestartTableData(
+          statement,
+          "1970-01-01T00:00:00.002Z,d1,north,1202,null,",
+          "1970-01-01T00:00:00.004Z,d1,north,14,false,");
     }
 
-    TestUtils.restartDataNodes();
+    TestUtils.stopForciblyAndRestartDataNodes();
 
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
         Statement statement = connection.createStatement()) {
+      connection.setClientInfo("time_zone", "+00:00");
       statement.execute("use \"test\"");
-      statement.execute("delete from root.turbine.d1.s1 where time<=1");
-
-      ResultSet resultSet = statement.executeQuery("SELECT s1 FROM root.turbine.d1");
-      assertNotNull(resultSet);
-      String[] exp = new String[] {"2,2.0", "3,3.0"};
-      int cnt = 0;
-      try {
-        while (resultSet.next()) {
-          String result = resultSet.getString(TIMESTAMP_STR) + "," + resultSet.getString(2);
-          assertEquals(exp[cnt], result);
-          cnt++;
-        }
+      assertRestartTableData(
+          statement,
+          "1970-01-01T00:00:00.002Z,d1,north,1202,null,",
+          "1970-01-01T00:00:00.004Z,d1,north,14,false,");
+      // Full data deletion must also preserve the other identities sharing the original file.
+      statement.execute("delete from restart_delete where device_id='d1' and site='north'");
+      assertRestartTableData(statement);
+      statement.execute(
+          "insert into restart_delete(time,device_id,site,s1) values(1,'d1','north',1101)");
+      statement.execute("flush");
+    }
 
-        statement.execute("flush");
-        statement.execute("delete from root.turbine.d1.s1 where time<=2");
+    TestUtils.stopForciblyAndRestartDataNodes();
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      connection.setClientInfo("time_zone", "+00:00");
+      statement.execute("use test");
+      assertRestartTableData(statement, "1970-01-01T00:00:00.001Z,d1,north,1101,null,");
+      statement.execute(
+          "insert into restart_delete(time,device_id,site,s1,s2) values(5,'d1','north',15,true)");
+      statement.execute("flush");
+      assertRestartTableData(
+          statement,
+          "1970-01-01T00:00:00.001Z,d1,north,1101,null,",
+          "1970-01-01T00:00:00.005Z,d1,north,15,true,");
+    }
+  }
 
-        exp = new String[] {"3,3.0"};
-        resultSet = statement.executeQuery("SELECT s1 FROM root.turbine.d1");
-        assertNotNull(resultSet);
-        cnt = 0;
-        while (resultSet.next()) {
-          String result = resultSet.getString(TIMESTAMP_STR) + "," + resultSet.getString(2);
-          assertEquals(exp[cnt], result);
-          cnt++;
-        }
-      } finally {
-        resultSet.close();
-      }
+  private static void assertRestartTableData(Statement statement, String... targetRows)
+      throws SQLException {
+    List<String> expected =
+        new ArrayList<>(
+            Arrays.asList(
+                "1970-01-01T00:00:00.002Z,d1,null,52,true,",
+                "1970-01-01T00:00:00.002Z,d1,,62,false,"));
+    expected.addAll(Arrays.asList(targetRows));
+    expected.add("1970-01-01T00:00:00.002Z,d1,south,212,true,");
+    try (ResultSet rows =
+        statement.executeQuery(
+            "select time,device_id,site,s1,s2 from restart_delete "
+                + "order by device_id,site nulls first,time")) {
+      TestUtils.assertResultSetEqual(
+          rows, "time,device_id,site,s1,s2,", expected.toArray(new String[0]));
+    }
+    try (ResultSet rows =
+        statement.executeQuery(
+            "select time,device_id,site,s1,s2 from restart_keep order by device_id,site,time")) {
+      TestUtils.assertResultSetEqual(
+          rows,
+          "time,device_id,site,s1,s2,",
+          new String[] {
+            "1970-01-01T00:00:00.001Z,d1,north,311,true,",
+            "1970-01-01T00:00:00.002Z,d1,north,312,false,"
+          });
     }
   }
 
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java
index 3e3b8beee0187..a1cd2af1de277 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/DataNodeQueryContext.java
@@ -168,7 +168,7 @@ public void updateLastCache(QualifiedObjectName tableName, DeviceEntry deviceEnt
         deviceCountAndMeasurementValues.get(tableName).get(deviceEntry).getRight();
     // if a device hits cache each time, the values recorded in context will be null
     if (values != null) {
-      TABLE_DEVICE_SCHEMA_CACHE.updateLastCacheIfExists(
+      TABLE_DEVICE_SCHEMA_CACHE.updateLastCacheFromQuery(
           tableName.getDatabaseName(),
           deviceEntry.getDeviceID(),
           values.keySet().toArray(new String[0]),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
index 3c1e19a80d801..3753a7173695a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/LastQueryAggTableScanOperator.java
@@ -76,7 +76,6 @@ public class LastQueryAggTableScanOperator extends AbstractAggTableScanOperator
   private final int allDeviceCount;
 
   private final boolean needUpdateCache;
-  private final boolean needUpdateNullEntry;
   private final List<Integer> hitCachesIndexes;
   private final List<Pair<OptionalLong, TsPrimitiveType[]>> lastRowCacheResults;
   private final List<TimeValuePair[]> lastValuesCacheResults;
@@ -105,8 +104,6 @@ public LastQueryAggTableScanOperator(
     this.cachedDeviceEntries = cachedDeviceEntries;
     this.needUpdateCache =
         LastQueryUtil.needUpdateCache(parameter.seriesScanOptions.getGlobalTimeFilter());
-    this.needUpdateNullEntry =
-        LastQueryUtil.needUpdateNullEntry(parameter.seriesScanOptions.getGlobalTimeFilter());
     this.hitCachesIndexes = hitCachesIndexes;
     this.lastRowCacheResults = lastRowCacheResults;
     this.lastValuesCacheResults = lastValuesCacheResults;
@@ -322,11 +319,15 @@ private void buildResultUseLastRowCache() {
 
   private void buildResultUseLastValuesCache() {
     TimeValuePair[] currentHitResult = lastValuesCacheResults.get(currentHitCacheIndex);
-    // if it is EMPTY_PRIMITIVE_TYPE, means there is no data in device
-    TsPrimitiveType timeLastValue = currentHitResult[currentHitResult.length - 1].getValue();
+    // The time column uses PLACEHOLDER_NO_VALUE even when its timestamp identifies a real row.
+    // Only an empty time entry means that the device has no data. Without a time entry, an
+    // empty final FIELD must not discard other fields' partial results.
+    boolean deviceHasNoData =
+        currentHitResult.length > measurementCount
+            && currentHitResult[measurementCount] == PLACEHOLDER_EMPTY_COLUMN;
     // when there is no data, no need to append result if the query is GROUP BY or output of
     // aggregator is partial (final operator will produce NULL result)
-    if (timeLastValue == PLACEHOLDER_NO_VALUE
+    if (deviceHasNoData
         && (groupingKeySize != 0 || tableAggregators.get(0).getStep().isOutputPartial())) {
       outputDeviceIndex++;
       currentHitCacheIndex++;
@@ -350,7 +351,7 @@ private void buildResultUseLastValuesCache() {
               getNthIdColumnValue(
                   cachedDeviceEntries.get(currentHitCacheIndex), aggColumnsIndexArray[columnIdx]);
           if (aggregator.getAccumulator() instanceof LastDescAccumulator) {
-            if (timeLastValue == PLACEHOLDER_NO_VALUE || id == null) {
+            if (deviceHasNoData || id == null) {
               columnBuilder.appendNull();
             } else {
               if (aggregator.getStep().isOutputPartial()) {
@@ -368,11 +369,13 @@ private void buildResultUseLastValuesCache() {
             }
           } else {
             // last_by
-            int measurementIdx = aggColumnsIndexArray[aggregatorInputChannels.get(channel)];
-            long lastTime =
-                lastValuesCacheResults.get(currentHitCacheIndex)[measurementIdx].getTimestamp();
+            int measurementIdx = aggColumnsIndexArray[aggregatorInputChannels.get(channel + 1)];
+            TimeValuePair sourceLast = currentHitResult[measurementIdx];
+            long lastTime = sourceLast.getTimestamp();
 
-            if (timeLastValue == PLACEHOLDER_NO_VALUE || id == null) {
+            if (sourceLast.getValue() == PLACEHOLDER_NO_VALUE) {
+              columnBuilder.appendNull();
+            } else if (id == null) {
               if (aggregator.getStep().isOutputPartial()) {
                 columnBuilder.writeBinary(
                     new Binary(
@@ -403,7 +406,7 @@ private void buildResultUseLastValuesCache() {
               cachedDeviceEntries.get(currentHitCacheIndex)
                   .getAttributeColumnValues()[aggColumnsIndexArray[columnIdx]];
           if (aggregator.getAccumulator() instanceof LastDescAccumulator) {
-            if (timeLastValue == PLACEHOLDER_NO_VALUE || attribute == null) {
+            if (deviceHasNoData || attribute == null) {
               columnBuilder.appendNull();
             } else {
               if (aggregator.getStep().isOutputPartial()) {
@@ -419,12 +422,14 @@ private void buildResultUseLastValuesCache() {
               }
             }
           } else {
-            int measurementIdx = aggColumnsIndexArray[aggregatorInputChannels.get(channel)];
-            long lastTime =
-                lastValuesCacheResults.get(currentHitCacheIndex)[measurementIdx].getTimestamp();
+            int measurementIdx = aggColumnsIndexArray[aggregatorInputChannels.get(channel + 1)];
+            TimeValuePair sourceLast = currentHitResult[measurementIdx];
+            long lastTime = sourceLast.getTimestamp();
 
             // last_by
-            if (timeLastValue == PLACEHOLDER_NO_VALUE || attribute == null) {
+            if (sourceLast.getValue() == PLACEHOLDER_NO_VALUE) {
+              columnBuilder.appendNull();
+            } else if (attribute == null) {
               if (aggregator.getStep().isOutputPartial()) {
                 columnBuilder.writeBinary(
                     new Binary(
@@ -452,7 +457,7 @@ private void buildResultUseLastValuesCache() {
         case TIME:
           if (aggregator.getAccumulator() instanceof LastDescAccumulator) {
             // for last(time) aggregation
-            if (timeLastValue == PLACEHOLDER_NO_VALUE) {
+            if (deviceHasNoData) {
               columnBuilder.appendNull();
             } else {
               if (aggregator.getStep().isOutputPartial()) {
@@ -477,14 +482,7 @@ private void buildResultUseLastValuesCache() {
 
             if (tsPrimitiveType == PLACEHOLDER_NO_VALUE) {
               // there is no data
-              if (aggregator.getStep().isOutputPartial()) {
-                columnBuilder.writeBinary(
-                    new Binary(
-                        serializeTimeValueWithNull(
-                            getTSDataType(schema.getType()), lastTime, true, false, null)));
-              } else {
-                columnBuilder.appendNull();
-              }
+              columnBuilder.appendNull();
             } else {
               if (aggregator.getStep().isOutputPartial()) {
                 // output: xDataType, yLastTime, xIsNull, isOrderTimeNull, xResult
@@ -547,6 +545,10 @@ private void updateLastCacheUseLastRowIfPossible() {
     boolean hasSetLastTime = false;
     for (int i = 0; i < tableAggregators.size(); i++) {
       TableAggregator tableAggregator = tableAggregators.get(i);
+      if (!canUpdateLastCache(tableAggregator)) {
+        channel += tableAggregator.getChannelCount();
+        continue;
+      }
       ColumnSchema schema = aggColumnSchemas.get(aggregatorInputChannels.get(channel));
 
       switch (schema.getColumnCategory()) {
@@ -574,6 +576,8 @@ private void updateLastCacheUseLastRowIfPossible() {
                     new TimeValuePair(
                         lastByAccumulator.getLastTimeOfY(),
                         new TsPrimitiveType.TsLong(lastByAccumulator.getLastTimeOfY())));
+              } else {
+                updateTimeValuePairList.add(PLACEHOLDER_EMPTY_COLUMN);
               }
             }
           }
@@ -618,6 +622,10 @@ private void updateLastCacheUseLastValuesIfPossible() {
     List<TimeValuePair> updateTimeValuePairList = new ArrayList<>();
     boolean hasSetLastTime = false;
     for (TableAggregator tableAggregator : tableAggregators) {
+      if (!canUpdateLastCache(tableAggregator)) {
+        channel += tableAggregator.getChannelCount();
+        continue;
+      }
       ColumnSchema schema = aggColumnSchemas.get(aggregatorInputChannels.get(channel));
 
       switch (schema.getColumnCategory()) {
@@ -625,16 +633,19 @@ private void updateLastCacheUseLastValuesIfPossible() {
         case TAG:
         case ATTRIBUTE:
           if (!hasSetLastTime && tableAggregator.getAccumulator() instanceof LastDescAccumulator) {
-            hasSetLastTime = true;
             LastDescAccumulator lastAccumulator =
                 (LastDescAccumulator) tableAggregator.getAccumulator();
-            updateMeasurementList.add("");
             if (lastAccumulator.hasInitResult()) {
+              hasSetLastTime = true;
+              updateMeasurementList.add("");
               updateTimeValuePairList.add(
                   new TimeValuePair(
                       lastAccumulator.getMaxTime(),
                       new TsPrimitiveType.TsLong(lastAccumulator.getMaxTime())));
-            } else {
+            } else if (schema.getColumnCategory() == TsTableColumnCategory.TIME) {
+              // A NULL TAG or ATTRIBUTE does not imply that the device has no rows.
+              hasSetLastTime = true;
+              updateMeasurementList.add("");
               currentDeviceEntry = deviceEntries.get(currentDeviceIndex);
               updateTimeValuePairList.add(PLACEHOLDER_EMPTY_COLUMN);
             }
@@ -667,6 +678,15 @@ private void updateLastCacheUseLastValuesIfPossible() {
     checkIfUpdated(updateMeasurementList, updateTimeValuePairList);
   }
 
+  private boolean canUpdateLastCache(TableAggregator aggregator) {
+    // TTL is added to the time filter when the scan's data source is initialized.
+    // An empty result under a lower time bound does not prove that older data is absent.
+    return LastQueryUtil.needUpdateNullEntry(seriesScanOptions.getGlobalTimeFilter())
+        || (aggregator.getAccumulator() instanceof LastAccumulator
+            ? ((LastAccumulator) aggregator.getAccumulator()).hasInitResult()
+            : ((LastByDescAccumulator) aggregator.getAccumulator()).hasInitResult());
+  }
+
   private void checkIfUpdated(
       List<String> updateMeasurementList, List<TimeValuePair> updateTimeValuePairList) {
     if (!updateMeasurementList.isEmpty()) {
@@ -675,7 +695,7 @@ private void checkIfUpdated(
       boolean deviceInMultiRegion =
           deviceCountMap != null && deviceCountMap.containsKey(currentDeviceEntry);
       if (!deviceInMultiRegion) {
-        TABLE_DEVICE_SCHEMA_CACHE.updateLastCacheIfExists(
+        TABLE_DEVICE_SCHEMA_CACHE.updateLastCacheFromQuery(
             dbName,
             currentDeviceEntry.getDeviceID(),
             updateMeasurementList.toArray(new String[0]),
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java
index 2fae9aa537081..01584fe230776 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/DataNodeTableOperatorGenerator.java
@@ -214,7 +214,7 @@
 import static org.apache.iotdb.db.queryengine.plan.analyze.PredicateUtils.convertPredicateToFilter;
 import static org.apache.iotdb.db.queryengine.plan.planner.OperatorTreeGenerator.isFilterGtOrGe;
 import static org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.SeriesScanOptions.updateFilterUsingTTL;
-import static org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache.PLACEHOLDER_NO_VALUE;
+import static org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache.PLACEHOLDER_EMPTY_COLUMN;
 import static org.apache.iotdb.db.queryengine.plan.relational.metadata.fetcher.cache.TableDeviceLastCache.PLACEHOLDER_STALE_VALUE;
 import static org.apache.tsfile.read.common.type.TimestampType.TIMESTAMP;
 
@@ -1911,23 +1911,18 @@ private LastQueryAggTableScanOperator constructLastQueryAggTableScanOperator(
                         "",
                         parameter.getMeasurementColumnNames());
             boolean allHitCache = true;
-            if (lastByResult.isPresent() && lastByResult.get().getLeft().isPresent()) {
+            // Check the aligned row's time even when the query has no FIELD columns.
+            if (lastByResult.isPresent()
+                && lastByResult.get().getLeft().isPresent()
+                && (updateTimeFilter == null
+                    || updateTimeFilter.satisfy(lastByResult.get().getLeft().getAsLong(), null))) {
               for (int j = 0; j < lastByResult.get().getRight().length; j++) {
                 TsPrimitiveType tsPrimitiveType = lastByResult.get().getRight()[j];
                 if (tsPrimitiveType == null
                     // Known-null at the aligned row time can still hit cache. Only miss or stale
                     // target
                     // values need to fall back to scan for correctness.
-                    || tsPrimitiveType == PLACEHOLDER_STALE_VALUE
-                    || (updateTimeFilter != null
-                        && !LastQueryUtil.satisfyFilter(
-                            updateTimeFilter,
-                            new TimeValuePair(
-                                lastByResult.get().getLeft().getAsLong(), tsPrimitiveType)))) {
-                  // the process logic is different from tree model which examine if
-                  // `isFilterGtOrGe(seriesScanOptions.getGlobalTimeFilter())`, set
-                  // `lastByResult.get().getRight()[j] = EMPTY_PRIMITIVE_TYPE`,
-                  // but it should skip in table model
+                    || tsPrimitiveType == PLACEHOLDER_STALE_VALUE) {
                   allHitCache = false;
                   break;
                 }
@@ -2014,18 +2009,18 @@ private LastQueryAggTableScanOperator constructLastQueryAggTableScanOperator(
                         targetColumns);
             boolean allHitCache = true;
             if (lastResult != null) {
-              for (TimeValuePair timeValuePair : lastResult) {
+              for (int k = 0; k < lastResult.length; k++) {
+                TimeValuePair timeValuePair = lastResult[k];
                 if (timeValuePair == null || timeValuePair.getValue() == null) {
                   allHitCache = false;
                   break;
                 }
 
                 if (updateTimeFilter != null
-                    && !LastQueryUtil.satisfyFilter(
-                        parameter.getSeriesScanOptions().getGlobalTimeFilter(), timeValuePair)) {
+                    && !LastQueryUtil.satisfyFilter(updateTimeFilter, timeValuePair)) {
                   if (isFilterGtOrGe(updateTimeFilter)) {
-                    // it means there is no data meets Filter
-                    timeValuePair.setValue(PLACEHOLDER_NO_VALUE);
+                    // No row satisfies this query. Keep the shared cached pair unchanged.
+                    lastResult[k] = PLACEHOLDER_EMPTY_COLUMN;
                   } else {
                     allHitCache = false;
                     break;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java
index 394cd509571c7..c7f67307a7019 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceCacheEntry.java
@@ -224,6 +224,14 @@ int tryUpdateLastCache(
     return Objects.nonNull(lastCache.get()) ? result : 0;
   }
 
+  int tryUpdateLastCacheFromQuery(
+      final String[] measurements, final TimeValuePair[] timeValuePairs) {
+    final TableDeviceLastCache cache = lastCache.get();
+    final int result =
+        Objects.nonNull(cache) ? cache.tryUpdateFromQuery(measurements, timeValuePairs) : 0;
+    return Objects.nonNull(lastCache.get()) ? result : 0;
+  }
+
   int tryUpdateLastCache(final String[] measurements, final TimeValuePair[] timeValuePairs) {
     return tryUpdateLastCache(measurements, timeValuePairs, false);
   }
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
index 5c79457bada95..f36c200207d00 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCache.java
@@ -144,6 +144,22 @@ int tryUpdate(
       final @Nullable IMeasurementSchema[] measurementSchemas,
       final @Nonnull TimeValuePair[] timeValuePairs,
       final boolean invalidateNull) {
+    return tryUpdate(measurements, measurementSchemas, timeValuePairs, invalidateNull, true);
+  }
+
+  int tryUpdateFromQuery(
+      final @Nonnull String[] measurements, final @Nonnull TimeValuePair[] timeValuePairs) {
+    // A FIELD's last non-NULL value can be older than the device's latest row. Queries must
+    // provide an explicit time entry when they can establish the actual last-row time.
+    return tryUpdate(measurements, null, timeValuePairs, false, false);
+  }
+
+  private int tryUpdate(
+      final @Nonnull String[] measurements,
+      final @Nullable IMeasurementSchema[] measurementSchemas,
+      final @Nonnull TimeValuePair[] timeValuePairs,
+      final boolean invalidateNull,
+      final boolean updateLastTimeFromMeasurements) {
     int diff = 0;
     long lastTime = Long.MIN_VALUE;
 
@@ -172,7 +188,9 @@ int tryUpdate(
         diff += tryUpdateCachedLast(measurement, timeValuePair);
       }
     }
-    tryUpdateLastTime(lastTime);
+    if (updateLastTimeFromMeasurements) {
+      tryUpdateLastTime(lastTime);
+    }
     return diff;
   }
 
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
index 1c11148ffc4b9..17f691d85426a 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceSchemaCache.java
@@ -341,6 +341,20 @@ public void updateLastCacheIfExists(
         false);
   }
 
+  /** Update query results without inferring the device's last-row time from FIELD timestamps. */
+  public void updateLastCacheFromQuery(
+      final String database,
+      final IDeviceID deviceId,
+      final String[] measurements,
+      final TimeValuePair[] timeValuePairs) {
+    dualKeyCache.update(
+        new TableId(database, deviceId.getTableName()),
+        deviceId,
+        null,
+        entry -> entry.tryUpdateLastCacheFromQuery(measurements, timeValuePairs),
+        false);
+  }
+
   /**
    * Update the last cache in writing or the second push of last cache query. If a measurement is
    * with all {@code null}s or is a tag/attribute column, its {@link TimeValuePair}[] shall be
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 c4550898a0529..f4c6c193b55fe 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,7 +19,9 @@
 
 package org.apache.iotdb.db.storageengine.dataregion.wal.recover.file;
 
+import org.apache.iotdb.commons.exception.IllegalPathException;
 import org.apache.iotdb.commons.path.MeasurementPath;
+import org.apache.iotdb.commons.path.PartialPath;
 import org.apache.iotdb.commons.service.metric.MetricService;
 import org.apache.iotdb.commons.service.metric.enums.Metric;
 import org.apache.iotdb.commons.service.metric.enums.Tag;
@@ -35,13 +37,16 @@
 import org.apache.iotdb.db.service.metrics.WritingMetrics;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
 import org.apache.iotdb.db.storageengine.dataregion.memtable.PrimitiveMemTable;
+import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate;
 import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
 import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry;
+import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.FullExactMatch;
 import org.apache.iotdb.db.storageengine.dataregion.modification.TreeDeletionEntry;
 import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
 import org.apache.iotdb.metrics.utils.MetricLevel;
 
 import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.read.common.TimeRange;
 import org.apache.tsfile.utils.Pair;
 
 import java.io.IOException;
@@ -49,6 +54,8 @@
 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
@@ -79,16 +86,63 @@ void redoDelete(DeleteDataNode deleteDataNode) throws IOException {
           new TreeDeletionEntry(
               path, deleteDataNode.getDeleteStartTime(), deleteDataNode.getDeleteEndTime());
       recoveryMemTable.delete(deletionEntry, ignoreAllNullRows);
-      deletionEntries.add(deletionEntry);
+      deletionEntries.addAll(deletionsForFlushedData(deletionEntry));
+    }
+    if (!deletionEntries.isEmpty()) {
+      tsFileResource.getModFileForWrite().write(deletionEntries);
     }
-    tsFileResource.getModFileForWrite().write(deletionEntries);
   }
 
   void redoDelete(RelationalDeleteDataNode node) throws IOException {
+    List<ModEntry> deletionEntries = new ArrayList<>();
     for (TableDeletionEntry modEntry : node.getModEntries()) {
       recoveryMemTable.delete(modEntry, ignoreAllNullRows);
+      deletionEntries.addAll(deletionsForFlushedData(modEntry));
+    }
+    if (!deletionEntries.isEmpty()) {
+      tsFileResource.getModFileForWrite().write(deletionEntries);
+    }
+  }
+
+  private List<ModEntry> deletionsForFlushedData(ModEntry deletion) 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<ModEntry> deletions = new ArrayList<>();
+    for (IDeviceID deviceID : tsFileResource.getDevices()) {
+      long startTime = tsFileResource.getStartTime(deviceID).get();
+      long endTime = tsFileResource.getEndTime(deviceID).get();
+      if (startTime > endTime || !deletion.affects(deviceID, startTime, endTime)) {
+        continue;
+      }
+      TimeRange timeRange =
+          new TimeRange(deletion.getStartTime(), Math.min(deletion.getEndTime(), endTime));
+      if (deletion instanceof TreeDeletionEntry) {
+        try {
+          PartialPath pathPattern = ((TreeDeletionEntry) deletion).getPathPattern();
+          deletions.add(
+              new TreeDeletionEntry(
+                  new MeasurementPath(
+                      deviceID,
+                      pathPattern.endWithMultiLevelWildcard()
+                          ? ONE_LEVEL_PATH_WILDCARD
+                          : pathPattern.getMeasurement()),
+                  timeRange));
+        } catch (IllegalPathException e) {
+          throw new IOException(e);
+        }
+      } else {
+        TableDeletionEntry tableDeletion = (TableDeletionEntry) deletion;
+        deletions.add(
+            new TableDeletionEntry(
+                new DeletionPredicate(
+                    tableDeletion.getTableName(),
+                    new FullExactMatch(deviceID),
+                    tableDeletion.getPredicate().getMeasurementNames()),
+                timeRange));
+      }
     }
-    tsFileResource.getModFileForWrite().write(node.getModEntries());
+    return deletions;
   }
 
   void redoInsert(InsertNode node) throws WriteProcessException {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
index 4ae3228013e56..3270e60157be7 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/metadata/fetcher/cache/TableDeviceLastCacheTest.java
@@ -32,6 +32,39 @@
 
 public class TableDeviceLastCacheTest {
 
+  @Test
+  public void testQueryFieldTimesDoNotInitializeAlignedTime() {
+    final TableDeviceLastCache cache = new TableDeviceLastCache(false);
+    cache.initOrInvalidate(null, null, new String[] {"", "s1"}, false);
+    final TimeValuePair historicalValue = new TimeValuePair(2L, new TsPrimitiveType.TsInt(20));
+
+    cache.tryUpdateFromQuery(new String[] {"s1"}, new TimeValuePair[] {historicalValue});
+
+    Assert.assertEquals(historicalValue, cache.getTimeValuePair("s1"));
+    Assert.assertNull(cache.getTimeValuePair(""));
+    Assert.assertFalse(cache.getLastRow("", Collections.singletonList("s1")).isPresent());
+
+    cache.tryUpdateFromQuery(
+        new String[] {"", "s1"},
+        new TimeValuePair[] {
+          new TimeValuePair(7L, TableDeviceLastCache.PLACEHOLDER_NO_VALUE),
+          new TimeValuePair(7L, TableDeviceLastCache.PLACEHOLDER_NO_VALUE)
+        });
+
+    Assert.assertEquals(historicalValue, cache.getTimeValuePair("s1"));
+    Optional<Pair<OptionalLong, TsPrimitiveType[]>> result =
+        cache.getLastRow("", Collections.singletonList("s1"));
+    Assert.assertTrue(result.isPresent());
+    Assert.assertEquals(OptionalLong.of(7L), result.get().getLeft());
+    Assert.assertArrayEquals(
+        new TsPrimitiveType[] {TableDeviceLastCache.PLACEHOLDER_NO_VALUE}, result.get().getRight());
+
+    final TimeValuePair newerFieldValue = new TimeValuePair(8L, new TsPrimitiveType.TsInt(80));
+    cache.tryUpdateFromQuery(new String[] {"s1"}, new TimeValuePair[] {newerFieldValue});
+    Assert.assertEquals(newerFieldValue, cache.getTimeValuePair("s1"));
+    Assert.assertEquals(7L, cache.getTimeValuePair("").getTimestamp());
+  }
+
   @Test
   public void testLazyUpdateOnlyComposesCachedMeasurements() {
     final TableDeviceLastCache cache = new TableDeviceLastCache(false);
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 a58171017ac1b..aa4450ce13e01 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
@@ -30,11 +30,16 @@
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode;
 import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
 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.DeletionPredicate;
+import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
 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.tsfile.TsFileResource;
 import org.apache.iotdb.db.storageengine.dataregion.wal.utils.TsFileUtilsForRecoverTest;
 import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest;
@@ -49,6 +54,7 @@
 import org.apache.tsfile.file.metadata.enums.TSEncoding;
 import org.apache.tsfile.read.TimeValuePair;
 import org.apache.tsfile.read.common.Path;
+import org.apache.tsfile.read.common.TimeRange;
 import org.apache.tsfile.read.reader.IPointReader;
 import org.apache.tsfile.utils.Binary;
 import org.apache.tsfile.utils.BitMap;
@@ -629,6 +635,124 @@ 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 MeasurementPath(SG_NAME + ".**")),
+            Long.MIN_VALUE,
+            Long.MAX_VALUE));
+    planRedoer.redoDelete(
+        new RelationalDeleteDataNode(
+            new PlanNodeId(""),
+            Collections.singletonList(
+                new TableDeletionEntry(
+                    new DeletionPredicate("table1"),
+                    new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE))),
+            null));
+
+    assertFalse(ModificationFile.getExclusiveMods(file).exists());
+  }
+
+  @Test
+  public void testRedoTreeDeleteDoesNotMaskLaterInserts() 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 = IDeviceID.Factory.DEFAULT_FACTORY.create(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 MeasurementPath(SG_NAME + ".**.s1")),
+            Long.MIN_VALUE,
+            Long.MAX_VALUE));
+
+    List<ModEntry> mods = tsFileResource.getModFileForWrite().getAllMods();
+    for (IDeviceID device : Arrays.asList(DEVICE1_NAME, DEVICE2_NAME, nestedDevice)) {
+      long endTime = tsFileResource.getEndTime(device).get();
+      assertTrue(mods.stream().anyMatch(mod -> mod.affects(device, endTime, endTime)));
+      assertFalse(mods.stream().anyMatch(mod -> mod.affects(device, endTime + 1, endTime + 1)));
+    }
+    assertFalse(mods.stream().anyMatch(mod -> mod.affects(DEVICE3_NAME)));
+    assertTrue(mods.stream().allMatch(mod -> mod.affects("s1") && !mod.affects("s2")));
+
+    planRedoer.redoDelete(
+        new DeleteDataNode(
+            new PlanNodeId(""),
+            Collections.singletonList(new MeasurementPath(SG_NAME + ".**")),
+            Long.MIN_VALUE,
+            Long.MAX_VALUE));
+    List<ModEntry> allMods = tsFileResource.getModFileForWrite().getAllMods();
+    assertTrue(allMods.stream().anyMatch(mod -> mod.affects(nestedDevice) && mod.affects("s2")));
+    assertFalse(allMods.stream().anyMatch(mod -> mod.affects(nestedDevice, 2, 2)));
+    assertFalse(
+        allMods.stream()
+            .anyMatch(
+                mod ->
+                    mod.affects(
+                        IDeviceID.Factory.DEFAULT_FACTORY.create(SG_NAME + ".d1.unflushed"))));
+  }
+
+  @Test
+  public void testRedoTableDeleteDoesNotMaskLaterInserts() throws Exception {
+    File file = new File(FILE_NAME);
+    generateCompleteFile(file);
+    tsFileResource = new TsFileResource(file);
+    List<IDeviceID> devices =
+        Arrays.asList(
+            IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1", "d1", "north"}),
+            IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1", "d1", null}),
+            IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table1", "d1", ""}));
+    for (int i = 0; i < devices.size(); i++) {
+      tsFileResource.updateStartTime(devices.get(i), 1);
+      tsFileResource.updateEndTime(devices.get(i), i + 2);
+    }
+    IDeviceID otherTable =
+        IDeviceID.Factory.DEFAULT_FACTORY.create(new String[] {"table2", "d1", "north"});
+    tsFileResource.updateStartTime(otherTable, 1);
+    tsFileResource.updateEndTime(otherTable, 10);
+    TsFilePlanRedoer planRedoer = new TsFilePlanRedoer(tsFileResource);
+
+    planRedoer.redoDelete(
+        new RelationalDeleteDataNode(
+            new PlanNodeId(""),
+            Collections.singletonList(
+                new TableDeletionEntry(
+                    new DeletionPredicate("table1", new NOP(), Collections.singletonList("s1")),
+                    new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE))),
+            null));
+
+    List<ModEntry> mods = tsFileResource.getModFileForWrite().getAllMods();
+    for (IDeviceID device : devices) {
+      long endTime = tsFileResource.getEndTime(device).get();
+      assertTrue(mods.stream().anyMatch(mod -> mod.affects(device, endTime, endTime)));
+      assertFalse(mods.stream().anyMatch(mod -> mod.affects(device, endTime + 1, endTime + 1)));
+    }
+    assertFalse(mods.stream().anyMatch(mod -> mod.affects(otherTable)));
+    assertFalse(
+        mods.stream()
+            .anyMatch(
+                mod ->
+                    mod.affects(
+                        IDeviceID.Factory.DEFAULT_FACTORY.create(
+                            new String[] {"table1", "d2", "north"}))));
+    assertTrue(mods.stream().allMatch(mod -> mod.affects("s1") && !mod.affects("s2")));
+  }
+
   @Test
   public void testRedoRelationalInsertTablet()
       throws IOException,