This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a05c63e0abb Fix aligned flush values after time deletion (#18634)
a05c63e0abb is described below
commit a05c63e0abb7c39fea333f428136ea5e86f15092
Author: Jiang Tian <[email protected]>
AuthorDate: Tue Sep 15 10:44:31 2026 +0800
Fix aligned flush values after time deletion (#18634)
---
.../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
