Fix aligned flush values after time deletion

integration-test/src/test/java/org/apache/iotdb/relational/it/db/it/IoTDBDeletionTableIT.java

@@ -358,6 +358,63 @@ public void testDeleteDataByAttributeFilter() throws SQLException {
     }
   }
 
+  /**
+   * Verifies that a deletion restricted by both an attribute and an exact timestamp remains
+   * effective after the affected data is flushed from the memtable.
+   */
+  @Test
+  public void testDeleteFromWhereAttributeAndTimeAfterFlush() throws SQLException {
+    try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);
+        Statement statement = connection.createStatement()) {
+      statement.execute("use test");
+      statement.execute(
+          "CREATE TABLE ad_stor_001(device_id STRING TAG, color STRING ATTRIBUTE, value INT32 FIELD)");
+      statement.execute(
+          "INSERT INTO ad_stor_001(time, device_id, color, value) VALUES (1, 'd1', 'red', 1)");
+      statement.execute(
+          "INSERT INTO ad_stor_001(time, device_id, color, value) VALUES (2, 'd1', 'red', 2)");
+      statement.execute(
+          "INSERT INTO ad_stor_001(time, device_id, color, value) VALUES (3, 'd1', 'red', 3)");
+      statement.execute(
+          "INSERT INTO ad_stor_001(time, device_id, color, value) VALUES (4, 'd1', 'red', 4)");
+      statement.execute(
+          "INSERT INTO ad_stor_001(time, device_id, color, value) VALUES (1, 'd2', 'blue', 5)");
+
+      assertEquals(5, countRows(statement, "SELECT COUNT(*) FROM ad_stor_001"));
+
+      statement.execute("DELETE FROM ad_stor_001 WHERE color = 'red' AND time = 2");
+      assertEquals(4, countRows(statement, "SELECT COUNT(*) FROM ad_stor_001"));
+
+      statement.execute("FLUSH");
+      assertEquals(4, countRows(statement, "SELECT COUNT(*) FROM ad_stor_001"));
+
+      final List<String> actual = new ArrayList<>();
+      try (ResultSet resultSet =
+          statement.executeQuery(
+              "SELECT device_id, time, color, value FROM ad_stor_001 "
+                  + "ORDER BY device_id, time")) {
+        while (resultSet.next()) {
+          actual.add(
+              resultSet.getString("device_id")
+                  + ","
+                  + resultSet.getLong("time")
+                  + ","
+                  + resultSet.getString("color")
+                  + ","
+                  + resultSet.getInt("value"));
+        }
+      }
+      assertEquals(List.of("d1,1,red,1", "d1,3,red,3", "d1,4,red,4", "d2,1,blue,5"), actual);
+    }
+  }
+
+  private int countRows(final Statement statement, final String query) throws SQLException {
+    try (ResultSet resultSet = statement.executeQuery(query)) {
+      assertTrue(resultSet.next());
+      return resultSet.getInt(1);
+    }
+  }
+
   @Test
   public void testDeleteDataByAttributeFilterWithTagAndTimeRange() throws SQLException {
     try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java

@@ -722,6 +722,10 @@ private void handleEncodingWithoutDeletedMeasurements(
                     alignedWorkingListForFlush.getValueIndex(sortedRowIndex))) {
               continue;
             }
+            // Keep value pages aligned with the time page when an entire timestamp is deleted.
+            if (alignedWorkingListForFlush.isTimeDeleted(sortedRowIndex)) {
+              continue;
+            }
             // skip time duplicated rows
             long time = alignedWorkingListForFlush.getTime(sortedRowIndex);
             if (Objects.nonNull(timeDuplicateInfo)) {
@@ -1116,6 +1120,10 @@ private void handleEncodingWithDeletedMeasurements(
                     alignedWorkingListForFlush.getValueIndex(sortedRowIndex))) {
               continue;
             }
+            // Keep value pages aligned with the time page when an entire timestamp is deleted.
+            if (alignedWorkingListForFlush.isTimeDeleted(sortedRowIndex)) {
+              continue;
+            }
             // skip time duplicated rows
             long time = alignedWorkingListForFlush.getTime(sortedRowIndex);
             if (Objects.nonNull(timeDuplicateInfo)) {

iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java

@@ -272,6 +272,80 @@ public void testAlignedFastPathKeepsPagesAndValuesAlignedAfterPartialSegmentSort
     }
   }
 
+  @Test
+  public void testAlignedFlushKeepsValuesAlignedAfterTimeDeletion() throws IOException {
+    // Deleted rows must be omitted from both time and value pages, including after sorting.
+    checkAlignedFlushAfterTimeDeletion(false);
+  }
+
+  @Test
+  public void testAlignedFlushKeepsValuesAlignedAfterTimeAndColumnDeletion() throws IOException {
+    // Removing the first measurement also exercises the remapped value-column encoding path.
+    checkAlignedFlushAfterTimeDeletion(true);
+  }
+
+  private void checkAlignedFlushAfterTimeDeletion(boolean removeColumn) throws IOException {
+    for (int pageSize : new int[] {2, 100}) {
+      List<IMeasurementSchema> schemas =
+          Arrays.asList(
+              new MeasurementSchema("s0", TSDataType.INT32, TSEncoding.PLAIN),
+              new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.PLAIN));
+      AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas, false);
+      String alignedFilePath =
+          TestConstant.OUTPUT_DATA_DIR.concat("testAlignedTimeDeletion" + pageSize + ".tsfile");
+      try {
+        for (int time : new int[] {4, 1, 6, 2, 5, 3}) {
+          memChunk.putAlignedRow(time, new Object[] {time, time * 10L});
+        }
+        memChunk.deleteTime(2, 2);
+        memChunk.deleteTime(6, 6);
+        if (removeColumn) {
+          memChunk.removeColumn("s0");
+        }
+        memChunk.sortTvListForFlush();
+
+        BlockingQueue<Object> ioTaskQueue = new LinkedBlockingQueue<>();
+        // Cover a single page and boundaries between pages and chunks.
+        memChunk.encodeWorkingAlignedTVList(ioTaskQueue, pageSize + 1, pageSize);
+        try (TsFileIOWriter alignedWriter = new TsFileIOWriter(new File(alignedFilePath))) {
+          alignedWriter.startChunkGroup(IDeviceID.Factory.DEFAULT_FACTORY.create("root.d"));
+          Object task;
+          while ((task = ioTaskQueue.poll()) != null) {
+            if (task instanceof IChunkWriter chunkWriter) {
+              chunkWriter.writeToFileWriter(alignedWriter);
+            }
+          }
+          alignedWriter.endChunkGroup();
+          alignedWriter.endFile();
+        }
+
+        try (TsFileSequenceReader sequenceReader = new TsFileSequenceReader(alignedFilePath);
+            TsFileReader fileReader = new TsFileReader(sequenceReader)) {
+          List<Path> paths = new ArrayList<>();
+          paths.add(new Path("root.d", "s1", false));
+          if (!removeColumn) {
+            paths.add(new Path("root.d", "s0", false));
+          }
+          QueryDataSet dataSet = fileReader.query(QueryExpression.create(paths, null));
+          for (int time : new int[] {1, 3, 4, 5}) {
+            assertTrue(dataSet.hasNext());
+            RowRecord row = dataSet.next();
+            assertEquals(time, row.getTimestamp());
+            assertEquals(TSDataType.INT64, row.getFields().get(0).getDataType());
+            assertEquals(time * 10L, row.getFields().get(0).getLongV());
+            if (!removeColumn) {
+              assertEquals(TSDataType.INT32, row.getFields().get(1).getDataType());
+              assertEquals(time, row.getFields().get(1).getIntV());
+            }
+          }
+          assertFalse(dataSet.hasNext());
+        }
+      } finally {
+        memChunk.release();
+      }
+    }
+  }
+
   @Test
   public void testAlignedFastPathEncodesUnmaterializedSegments() throws Exception {
     // Exercise all six value representations with null/dense/null segments, partial nulls, an