This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b9f7337d13644a051e0fed0eae6c034708dbe55f Author: Tian Jiang <[email protected]> AuthorDate: Thu Sep 10 15:24:11 2026 +0800 add flat_device deletion IT --- .../it/insertquery/IoTDBFlatDeviceIT.java | 153 +++++++++++++++++++++ .../execution/operator/source/SeriesScanUtil.java | 52 +++++-- .../db/queryengine/plan/analyze/AnalyzeUtils.java | 31 +++-- .../apache/iotdb/db/utils/ModificationUtils.java | 6 +- 4 files changed, 215 insertions(+), 27 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java index 544664292a8..67fed2339e9 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/insertquery/IoTDBFlatDeviceIT.java @@ -53,6 +53,7 @@ import java.sql.SQLException; import java.sql.Statement; import java.util.ArrayList; import java.util.Arrays; +import java.util.Comparator; import java.util.List; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -436,6 +437,153 @@ public class IoTDBFlatDeviceIT { } } + // Delete sealed rows with equal timestamps using compound TAGs, NULL, empty strings and IN. + // Every step checks raw scans and aggregation so mods cannot incorrectly remove whole chunks. + @Test + public void testFlatDeviceSealedDeletionByTags() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES " + + "(9, 'a', 'x', 9), (10, 'a', 'x', 10), (10, 'a', 'y', 11)," + + "(10, 'b', 'x', 12), (10, 'a', null, 13), (10, 'a', '', 14)," + + "(10, null, 'x', 15), (11, 'a', 'x', 16)"); + statement.execute("FLUSH"); + assertCount(statement, 8); + assertDeletionRows( + statement, + Arrays.asList( + "9,a,x,null,9.0", + "10,a,,null,14.0", + "10,a,x,null,10.0", + "10,a,y,null,11.0", + "10,a,null,null,13.0", + "10,b,x,null,12.0", + "10,null,x,null,15.0", + "11,a,x,null,16.0")); + statement.execute("DELETE FROM sensor WHERE tag_a = 'a' AND tag_b = 'x' AND time = 10"); + statement.execute("DELETE FROM sensor WHERE tag_a = 'a' AND tag_b IS NULL AND time = 10"); + statement.execute("DELETE FROM sensor WHERE tag_a IS NULL AND time = 10"); + assertDeletionRows( + statement, + Arrays.asList( + "9,a,x,null,9.0", + "10,a,,null,14.0", + "10,a,y,null,11.0", + "10,b,x,null,12.0", + "11,a,x,null,16.0")); + // Repeat a deletion and target a nonexistent device: it must not affect survivors. + statement.execute("DELETE FROM sensor WHERE tag_a = 'a' AND tag_b = 'x' AND time = 10"); + statement.execute("DELETE FROM sensor WHERE tag_a = 'missing' AND time = 10"); + statement.execute("DELETE FROM sensor WHERE tag_b = '' AND time = 10"); + assertDeletionRows( + statement, + Arrays.asList( + "9,a,x,null,9.0", "10,a,y,null,11.0", "10,b,x,null,12.0", "11,a,x,null,16.0")); + statement.execute("DELETE FROM sensor WHERE tag_b IS NOT NULL AND time = 10"); + assertDeletionRows(statement, Arrays.asList("9,a,x,null,9.0", "11,a,x,null,16.0")); + } + } + + // One TAG/time deletion spans sealed and in-memory rows, including an overwrite of a disk key. + // Reinsert the deleted key, seal it, and restart: only the new value and unrelated devices + // survive. + @Test + public void testFlatDeviceMixedDeletionAndReinsertAfterRestart() throws Exception { + List<String> expected = + Arrays.asList("10,a,x,null,100.0", "10,b,x,null,20.0", "11,b,x,null,40.0"); + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES " + + "(10, 'a', 'x', 10), (10, 'b', 'x', 20)"); + statement.execute("FLUSH"); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES " + + "(10, 'a', 'x', 15), (11, 'a', 'x', 30), (11, 'b', 'x', 40)"); + statement.execute("DELETE FROM sensor WHERE tag_a = 'a' AND time >= 10 AND time <= 11"); + assertDeletionRows(statement, Arrays.asList("10,b,x,null,20.0", "11,b,x,null,40.0")); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES (10, 'a', 'x', 100)"); + assertDeletionRows(statement, expected); + statement.execute("FLUSH"); + assertDeletionRows(statement, expected); + } + EnvFactory.getEnv().shutdownForciblyAllDataNodes(); + EnvFactory.getEnv().startAllDataNodes(); + ((AbstractEnv) EnvFactory.getEnv()).checkClusterStatusWithoutUnknown(); + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + assertDeletionRows(statement, expected); + } + } + + // Time-only deletion includes both boundary timestamps across logical devices in a sealed file. + // Whole-table deletion must empty raw/aggregate results while allowing the same key to be reused. + @Test + public void testFlatDeviceSealedTimeAndFullDeletion() throws Exception { + try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE); + statement.execute( + "INSERT INTO sensor (time, tag_a, tag_b, value) VALUES " + + "(9, 'a', 'x', 9), (10, 'a', 'x', 10), (10, 'b', 'x', 11)," + + "(11, 'a', 'x', 12), (12, 'b', 'x', 13)"); + statement.execute("FLUSH"); + statement.execute("DELETE FROM sensor WHERE time >= 10 AND time <= 11"); + assertDeletionRows(statement, Arrays.asList("9,a,x,null,9.0", "12,b,x,null,13.0")); + statement.execute("DELETE FROM sensor"); + assertCount(statement, 0); + try (ResultSet resultSet = statement.executeQuery(query())) { + assertFalse(resultSet.next()); + } + statement.execute("INSERT INTO sensor (time, tag_a, tag_b, value) VALUES (10, 'b', 'x', 20)"); + assertDeletionRows(statement, Arrays.asList("10,b,x,null,20.0")); + statement.execute("FLUSH"); + assertDeletionRows(statement, Arrays.asList("10,b,x,null,20.0")); + } + } + + private static void assertDeletionRows(Statement statement, List<String> expected) + throws SQLException { + try (ResultSet resultSet = statement.executeQuery(query())) { + assertRows(resultSet, expected); + } + List<String> reversed = new ArrayList<>(expected); + reversed.sort( + Comparator.comparingLong((String row) -> Long.parseLong(row.substring(0, row.indexOf(',')))) + .reversed() + .thenComparing( + row -> nullableColumn(row, 1), Comparator.nullsLast(Comparator.reverseOrder())) + .thenComparing( + row -> nullableColumn(row, 2), Comparator.nullsLast(Comparator.reverseOrder()))); + try (ResultSet resultSet = + statement.executeQuery( + "SELECT time, tag_a, tag_b, attr, value FROM sensor" + + " ORDER BY time DESC, tag_a DESC, tag_b DESC")) { + assertRows(resultSet, reversed); + } + // TAG columns are absent from the projection, and a value filter supplies an input selection. + try (ResultSet resultSet = + statement.executeQuery("SELECT value FROM sensor WHERE value >= 11 ORDER BY value")) { + List<Double> actual = new ArrayList<>(); + while (resultSet.next()) { + actual.add(resultSet.getDouble(1)); + } + assertEquals( + expected.stream() + .map(row -> Double.valueOf(row.substring(row.lastIndexOf(',') + 1))) + .filter(value -> value >= 11) + .sorted() + .collect(Collectors.toList()), + actual); + } + assertCount(statement, expected.size()); + } + private static void assertRecovery(boolean flushBeforeRestart) throws Exception { final List<String> expected = Arrays.asList("10,a,1,first,1.0", "10,b,2,second,2.0"); try (Connection connection = EnvFactory.getEnv().getConnection(BaseEnv.TABLE_SQL_DIALECT); @@ -518,6 +666,11 @@ public class IoTDBFlatDeviceIT { } } + private static String nullableColumn(String row, int columnIndex) { + String value = row.split(",", -1)[columnIndex]; + return "null".equals(value) ? null : value; + } + private static void assertFlatDeviceAlterRejected(Statement statement, String sql) throws SQLException { try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java index a786170cb43..515f6f92814 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/SeriesScanUtil.java @@ -72,6 +72,7 @@ import org.apache.tsfile.read.common.block.column.IntColumn; import org.apache.tsfile.read.common.block.column.LongColumn; import org.apache.tsfile.read.controller.IChunkLoader; import org.apache.tsfile.read.filter.basic.Filter; +import org.apache.tsfile.read.filter.factory.FilterFactory; import org.apache.tsfile.read.reader.IPageReader; import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.read.reader.page.AlignedPageReader; @@ -726,10 +727,11 @@ public class SeriesScanUtil implements Accountable { chunkMetaData, scanOptions.getGlobalTimeFilter(), getTsDataTypeList(), context); // Flat-device TAG deletions are row-scoped and must be evaluated while decoding each page. - if (rowKeyColumnIndices.length > 0 - && chunkLoader instanceof DiskAlignedChunkLoader diskAlignedLoader) { - addFlatDeviceDeletionFilter(pageReaderList, diskAlignedLoader.getResource()); - } + Filter deletionFilter = + rowKeyColumnIndices.length > 0 + && chunkLoader instanceof DiskAlignedChunkLoader diskAlignedLoader + ? getFlatDeviceDeletionFilter(diskAlignedLoader.getResource()) + : null; // init TsBlockBuilder for each page reader pageReaderList.forEach(p -> p.initTsBlockBuilder(getTsDataTypeList())); @@ -745,7 +747,8 @@ public class SeriesScanUtil implements Accountable { chunkMetaData.getOffsetOfChunkHeader(), iPageReader, true, - tsFile)); + tsFile, + deletionFilter)); } } else { for (int i = pageReaderList.size() - 1; i >= 0; i--) { @@ -757,7 +760,8 @@ public class SeriesScanUtil implements Accountable { chunkMetaData.getOffsetOfChunkHeader(), pageReaderList.get(i), true, - tsFile)); + tsFile, + deletionFilter)); } } } else { @@ -771,7 +775,8 @@ public class SeriesScanUtil implements Accountable { chunkMetaData.getOffsetOfChunkHeader(), pageReader, false, - tsFile))); + tsFile, + deletionFilter))); } if (LOGGER.isDebugEnabled()) { @@ -783,8 +788,7 @@ public class SeriesScanUtil implements Accountable { } } - private void addFlatDeviceDeletionFilter( - List<IPageReader> pageReaderList, TsFileResource resource) { + private Filter getFlatDeviceDeletionFilter(TsFileResource resource) { List<FlatDeviceDeletionEntry> deletions; try { deletions = @@ -798,10 +802,9 @@ public class SeriesScanUtil implements Accountable { } catch (IllegalPathException e) { throw new IllegalStateException(e); } - if (!deletions.isEmpty()) { - Filter filter = new FlatDeviceDeletionFilter(rowKeyColumnIndices, deletions); - pageReaderList.forEach(pageReader -> pageReader.addRecordFilter(filter)); - } + return deletions.isEmpty() + ? null + : new FlatDeviceDeletionFilter(rowKeyColumnIndices, deletions); } private void unpackOneFakeMemChunkMetaData( @@ -2129,6 +2132,7 @@ public class SeriesScanUtil implements Accountable { protected final QueryContext context; protected final MergeReaderPriority version; protected final IPageReader data; + private final Filter deletionFilter; protected final boolean isSeq; protected final boolean isAligned; @@ -2143,9 +2147,26 @@ public class SeriesScanUtil implements Accountable { IPageReader data, boolean isSeq, File tsFile) { + this(context, fileTimestamp, version, offset, data, isSeq, tsFile, null); + } + + VersionPageReader( + QueryContext context, + long fileTimestamp, + long version, + long offset, + IPageReader data, + boolean isSeq, + File tsFile, + Filter deletionFilter) { this.context = context; this.version = new MergeReaderPriority(fileTimestamp, version, offset, isSeq); this.data = data; + this.deletionFilter = deletionFilter; + if (deletionFilter != null) { + // Overlapping pages must apply file-scoped deletions before entering the merge reader. + data.addRecordFilter(deletionFilter); + } this.isSeq = isSeq; this.isAligned = data instanceof AlignedPageReader @@ -2235,11 +2256,12 @@ public class SeriesScanUtil implements Accountable { } public void addPushDownFilter(Filter pushDownFilter) { - data.addRecordFilter(pushDownFilter); + // addRecordFilter replaces the previous filter, including when the new filter is null. + data.addRecordFilter(FilterFactory.and(deletionFilter, pushDownFilter)); } public boolean isModified() { - return data.isModified(); + return deletionFilter != null || data.isModified(); } public boolean isSeq() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java index 6a4bd7562d5..bb574c0b693 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/analyze/AnalyzeUtils.java @@ -525,13 +525,10 @@ public class AnalyzeUtils { final String databaseName, final MPPQueryContext queryContext) { if (expression == null) { - return table.isFlatDevice() - ? new FlatDeviceDeletionEntry( - new DeletionPredicate(table.getTableName()), - new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE, true).toTsFileTimeRange()) - : new TableDeletionEntry( - new DeletionPredicate(table.getTableName()), - new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE, true).toTsFileTimeRange()); + return createTableDeletionEntry( + table, + new DeletionPredicate(table.getTableName()), + new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE, true)); } final PredicateParseContext predicateParseContext = @@ -582,10 +579,22 @@ public class AnalyzeUtils { predicateParseContext.timeRange.getEndTime())); } - return table.isFlatDevice() - ? new FlatDeviceDeletionEntry( - predicate, predicateParseContext.timeRange.toTsFileTimeRange()) - : new TableDeletionEntry(predicate, predicateParseContext.timeRange.toTsFileTimeRange()); + return createTableDeletionEntry(table, predicate, predicateParseContext.timeRange); + } + + private static TableDeletionEntry createTableDeletionEntry( + TsTable table, DeletionPredicate predicate, TimeRange timeRange) { + if (!table.isFlatDevice()) { + return new TableDeletionEntry(predicate, timeRange.toTsFileTimeRange()); + } + FlatDeviceDeletionEntry deletion = + new FlatDeviceDeletionEntry(predicate, timeRange.toTsFileTimeRange()); + // Preserve TAG order so row deletion can resolve columns without consulting the table cache. + deletion.setTagColumnNames( + table.getTagColumnSchemaList().stream() + .map(TsTableColumnSchema::getColumnName) + .collect(Collectors.toList())); + return deletion; } private static void applyPredicateParseResult( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java index 9afd8a189aa..0d7e99ec00c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/ModificationUtils.java @@ -300,7 +300,11 @@ public class ModificationUtils { return false; } for (ModEntry modification : modifications) { - if ((modification.getTimeRange().contains(startTime, endTime))) { + // A flat-device file may contain many logical devices. Time coverage alone cannot prove + // that a TAG predicate deletes every row; the page filter must inspect their TAG values. + if (!(modification instanceof FlatDeviceDeletionEntry deletion + && deletion.getPredicate().getTagPredicateType() != TagPredicate.TagPredicateType.NOP) + && modification.getTimeRange().contains(startTime, endTime)) { return true; } }
