This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch fix/aligned-flush-time-deletion in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 74fa15f363b83e7248622c914f0db46a9ed3edba Author: Tian Jiang <[email protected]> AuthorDate: Mon Sep 14 16:31:12 2026 +0800 Fix aligned flush values after time deletion --- .../relational/it/db/it/IoTDBDeletionTableIT.java | 57 +++++++++++++++++ .../memtable/AlignedWritableMemChunk.java | 8 +++ .../dataregion/memtable/MemTableFlushTaskTest.java | 74 ++++++++++++++++++++++ 3 files changed, 139 insertions(+) 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 b31b853f9b8..f8732645926 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 @@ -358,6 +358,63 @@ public class IoTDBDeletionTableIT { } } + /** + * 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); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java index d1c15384173..ac42128b6cb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java @@ -722,6 +722,10 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { 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 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { 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)) { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java index 8df5e71c836..ad1cbc219de 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/MemTableFlushTaskTest.java @@ -272,6 +272,80 @@ public class MemTableFlushTaskTest { } } + @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
