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 9b04706c717efd0f13547d7a1b9687180e4e6807 Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 9 15:12:43 2026 +0800 tmp --- .../operator/source/AlignedSeriesScanUtil.java | 1 + .../db/queryengine/plan/analyze/AnalyzeUtils.java | 16 +++- .../db/storageengine/dataregion/DataRegion.java | 8 +- .../memtable/AlignedWritableMemChunk.java | 79 ++++++++++++++++++-- .../memtable/AlignedWritableMemChunkGroup.java | 9 +++ .../db/utils/datastructure/AlignedTVList.java | 85 +++++++++++++++++++--- .../datastructure/ColumnSortedAlignedTVList.java | 22 ++++-- 7 files changed, 190 insertions(+), 30 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java index d70cbf210f4..4dd97e629e3 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/AlignedSeriesScanUtil.java @@ -97,6 +97,7 @@ public class AlignedSeriesScanUtil extends SeriesScanUtil { .map(IMeasurementSchema::getType) .collect(Collectors.toList()); if (!rowKeyColumnIndices.isEmpty()) { + this.rowKeyColumnIndices = rowKeyColumnIndices.stream().mapToInt(Integer::intValue).toArray(); if (getMergeReader() instanceof AlignedPriorityMergeReader alignedMergeReader) { alignedMergeReader.configureRowKeyColumns(dataTypes, rowKeyColumnIndices); } else if (getMergeReader() instanceof AlignedDescPriorityMergeReader alignedMergeReader) { 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 81849ff9dfb..6a4bd7562d5 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 @@ -66,6 +66,7 @@ import org.apache.iotdb.db.queryengine.plan.statement.crud.InsertTabletStatement import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; import org.apache.iotdb.db.schemaengine.table.DataNodeTreeViewSchemaUtils; import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate; +import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate; import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.And; @@ -524,9 +525,13 @@ public class AnalyzeUtils { final String databaseName, final MPPQueryContext queryContext) { if (expression == null) { - return new TableDeletionEntry( - new DeletionPredicate(table.getTableName()), - new TimeRange(Long.MIN_VALUE, Long.MAX_VALUE, true).toTsFileTimeRange()); + 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()); } final PredicateParseContext predicateParseContext = @@ -577,7 +582,10 @@ public class AnalyzeUtils { predicateParseContext.timeRange.getEndTime())); } - return new TableDeletionEntry(predicate, predicateParseContext.timeRange.toTsFileTimeRange()); + return table.isFlatDevice() + ? new FlatDeviceDeletionEntry( + predicate, predicateParseContext.timeRange.toTsFileTimeRange()) + : new TableDeletionEntry(predicate, predicateParseContext.timeRange.toTsFileTimeRange()); } private static void applyPredicateParseResult( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 07308ec0467..12add8d706b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -1741,6 +1741,12 @@ public class DataRegion implements IDataRegionForQuery { private void registerToTsFile(InsertNode node, TsFileProcessor tsFileProcessor) { final String tableName = node.getTableName(); if (tableName != null) { + // A relational batch keeps the layout flag on its child rows, just like the table name. + // Reading the batch's inherited flag would incorrectly mark flat-device files as ordinary. + final boolean isFlatDevice = + node instanceof InsertRowsNode insertRowsNode + ? insertRowsNode.getInsertRowNodeList().get(0).isFlatDevice() + : node.isFlatDevice(); tsFileProcessor.registerToTsFile( tableName, t -> { @@ -1798,7 +1804,7 @@ public class DataRegion implements IDataRegionForQuery { cacheTableSchema(database, t, currentVersion, schema); return schema; }, - node.isFlatDevice()); + isFlatDevice); } } 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 5e6b76b2f5a..eea675f7a6a 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 @@ -27,6 +27,8 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.DataTypeInconsistentException; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; +import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; @@ -37,6 +39,7 @@ import org.apache.iotdb.db.utils.datastructure.MemPointIterator; import org.apache.iotdb.db.utils.datastructure.MemPointIteratorFactory; import org.apache.iotdb.db.utils.datastructure.TVList; +import org.apache.tsfile.common.conf.TSFileConfig; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; import org.apache.tsfile.enums.TSDataType; @@ -688,6 +691,60 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { return deletedNumber; } + public boolean isFlatDevice() { + return useColumnSortedList; + } + + public int deleteFlatDevice(FlatDeviceDeletionEntry deletion) { + List<String> tagColumnNames = deletion.getTagColumnNames(); + // Logical TAG ordinals do not necessarily match physical positions: failed/omitted TAGs + // may be absent, and later TAGs can be appended after FIELD columns. + int[] tagColumnIndices = + tagColumnNames.stream() + .mapToInt(name -> measurementIndexMap.getOrDefault(name, -1)) + .toArray(); + List<Integer> columns = + deletion.getPredicate().getMeasurementNames().isEmpty() ? null : new ArrayList<>(); + if (columns != null) { + for (Map.Entry<String, Integer> entry : measurementIndexMap.entrySet()) { + if (deletion.affects(entry.getKey())) { + columns.add(entry.getValue()); + } + } + } + int deleted = deleteFlatDevice(list, deletion, tagColumnIndices, columns); + for (AlignedTVList tvList : sortedList) { + deleted += deleteFlatDevice(tvList, deletion, tagColumnIndices, columns); + } + return deleted; + } + + private int deleteFlatDevice( + AlignedTVList tvList, + TableDeletionEntry deletion, + int[] tagColumnIndices, + List<Integer> columns) { + // Reuse the logical-device segments buffer for every row in this TVList to avoid allocating + // one array per candidate row during TAG deletion. + Binary[] segments = new Binary[tagColumnIndices.length + 1]; + segments[0] = new Binary(deletion.getTableName(), TSFileConfig.STRING_CHARSET); + return tvList.deleteMatchingRows( + deletion.getStartTime(), + deletion.getEndTime(), + valueIndex -> { + for (int i = 0; i < tagColumnIndices.length; i++) { + int columnIndex = tagColumnIndices[i]; + segments[i + 1] = null; + // Historical TVLists may predate this TAG; an absent column is a null TAG. + if (!tvList.isNullValue(valueIndex, columnIndex)) { + segments[i + 1] = tvList.getBinaryByValueIndex(valueIndex, columnIndex); + } + } + return deletion.affects(segments); + }, + columns); + } + public Pair<Integer, Boolean> deleteDataFromAColumn( long lowerBound, long upperBound, String measurementId) { Pair<Integer, Boolean> deletePair = @@ -775,6 +832,14 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { for (int sortedRowIndex = 0; sortedRowIndex < alignedWorkingListForFlush.rowCount(); sortedRowIndex++) { + // Deleted rows may retain TAG values to preserve compound ordering. Exclude them from + // page boundaries as well as both the time and value encoders below. + if (alignedWorkingListForFlush.isTimeDeleted(sortedRowIndex) + || (allValueColDeletedMap != null + && allValueColDeletedMap.isMarked( + alignedWorkingListForFlush.getValueIndex(sortedRowIndex)))) { + continue; + } if (pointNumInPage == 0) { pageRange.add(sortedRowIndex); } @@ -878,9 +943,10 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { sortedRowIndex <= pageRange.get(pageNum * 2 + 1); sortedRowIndex++) { // skip empty row - if (allValueColDeletedMap != null - && allValueColDeletedMap.isMarked( - alignedWorkingListForFlush.getValueIndex(sortedRowIndex))) { + if (alignedWorkingListForFlush.isTimeDeleted(sortedRowIndex) + || (allValueColDeletedMap != null + && allValueColDeletedMap.isMarked( + alignedWorkingListForFlush.getValueIndex(sortedRowIndex)))) { continue; } // skip time duplicated rows @@ -1038,9 +1104,10 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { sortedRowIndex <= pageRange.get(pageNum * 2 + 1); sortedRowIndex++) { // skip empty row - if (allValueColDeletedMap != null - && allValueColDeletedMap.isMarked( - alignedWorkingListForFlush.getValueIndex(sortedRowIndex))) { + if (alignedWorkingListForFlush.isTimeDeleted(sortedRowIndex) + || (allValueColDeletedMap != null + && allValueColDeletedMap.isMarked( + alignedWorkingListForFlush.getValueIndex(sortedRowIndex)))) { continue; } // skip time duplicated rows diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java index 80641a33cee..3bb49b9d499 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkGroup.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.db.exception.DataTypeInconsistentException; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; @@ -172,6 +173,9 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { @Override public long delete(ModEntry modEntry) { + if (modEntry instanceof FlatDeviceDeletionEntry flatDeviceDeletion) { + return delete(flatDeviceDeletion); + } int deletedPointsNumber = 0; Set<String> measurements = memChunk.getAllMeasurements(); List<String> columnsToBeRemoved = new ArrayList<>(); @@ -195,6 +199,11 @@ public class AlignedWritableMemChunkGroup implements IWritableMemChunkGroup { return deletedPointsNumber; } + /** Deletes only rows whose logical device matches the flat-device TAG predicate. */ + public long delete(FlatDeviceDeletionEntry deletion) { + return memChunk.deleteFlatDevice(deletion); + } + public long deleteTime(ModEntry modEntry) { return memChunk.deleteTime(modEntry.getStartTime(), modEntry.getEndTime()); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java index 1bf9bcf1050..807ecea68eb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java @@ -49,7 +49,6 @@ import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.Pair; import org.apache.tsfile.utils.RamUsageEstimator; -import org.apache.tsfile.utils.ReadWriteForEncodingUtils; import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.utils.TsPrimitiveType; import org.apache.tsfile.write.UnSupportedDataTypeException; @@ -65,6 +64,7 @@ import java.util.HashSet; import java.util.List; import java.util.Objects; import java.util.Set; +import java.util.function.IntPredicate; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -944,6 +944,48 @@ public abstract class AlignedTVList extends TVList { return timeDeletedCnt - prevDeletedCnt; } + /** + * Delete rows selected by their original value index, or only the given columns when non-null. + * Matching and bitmap updates share the sort monitor so a concurrent sort cannot change indices + * between the predicate evaluation and the deletion. + */ + public synchronized int deleteMatchingRows( + long lowerBound, long upperBound, IntPredicate rowMatches, List<Integer> columnIndices) { + int deleted = 0; + for (int i = 0; i < rowCount; i++) { + long time = getTime(i); + int valueIndex = getValueIndex(i); + if (time < lowerBound + || time > upperBound + || isTimeDeleted(i) + || !rowMatches.test(valueIndex)) { + continue; + } + if (columnIndices == null) { + // Retain TAG values: clearing a compound sort key could reorder other equal-time rows. + if (markRowNull(i)) { + deleted++; + } + } else { + for (int columnIndex : columnIndices) { + if (!isNullValue(valueIndex, columnIndex) + && markNullValue(columnIndex, valueIndex / ARRAY_SIZE, valueIndex % ARRAY_SIZE)) { + deleted++; + } + } + } + } + if (columnIndices == null && deleted > 0) { + if (lowerBound <= minTime && minTime <= upperBound) { + updateMinTime(); + } + if (lowerBound <= maxTime && maxTime <= upperBound) { + updateMaxTime(); + } + } + return deleted; + } + private void updateMaxTime() { long maxTime = Long.MIN_VALUE; for (int i = 0; i < rowCount; i++) { @@ -2031,7 +2073,8 @@ public abstract class AlignedTVList extends TVList { size += Byte.BYTES; if (timeColDeletedMap != null) { int length = timeColDeletedMap.getByteArray().length; - return ReadWriteForEncodingUtils.varIntSize(length) + length * Byte.BYTES; + // WALWriteUtils writes the bitmap length as a fixed-width int, after all row data. + size += Integer.BYTES + length * Byte.BYTES; } return size; } @@ -2060,8 +2103,11 @@ public abstract class AlignedTVList extends TVList { for (int columnIndex = 0; columnIndex < values.size(); columnIndex++) { List<Object> columnValues = values.get(columnIndex); for (int rowIndex = 0; rowIndex < rowCount; ++rowIndex) { - int arrayIndex = rowIndex / ARRAY_SIZE; - int elementIndex = rowIndex % ARRAY_SIZE; + // Timestamps are serialized in sorted order; values and deletion bits must use the + // same order even if a query sorted this TVList before the WAL snapshot. + int valueIndex = getValueIndex(rowIndex); + int arrayIndex = valueIndex / ARRAY_SIZE; + int elementIndex = valueIndex % ARRAY_SIZE; Object valueArray = columnValues.get(arrayIndex); // value switch (dataTypes.get(columnIndex)) { @@ -2109,14 +2155,20 @@ public abstract class AlignedTVList extends TVList { throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT); } // bitmap - WALWriteUtils.write(isNullValue(rowIndex, columnIndex), buffer); + WALWriteUtils.write(isNullValue(valueIndex, columnIndex), buffer); } } if (timeColDeletedMap != null) { buffer.put((byte) 1); WALWriteUtils.write(timeColDeletedMap.getByteArray().length, buffer); - buffer.put(timeColDeletedMap.getByteArray()); + BitMap serializedDeletionMap = new BitMap(timeColDeletedMap.getSize()); + for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { + if (isTimeDeleted(rowIndex)) { + serializedDeletionMap.mark(rowIndex); + } + } + buffer.put(serializedDeletionMap.getByteArray()); } else { buffer.put((byte) 0); } @@ -2226,15 +2278,24 @@ public abstract class AlignedTVList extends TVList { AlignedTVList tvList = AlignedTVListFactory.createAlignedList(new ArrayList<>(dataTypes)); tvList.putAlignedValues(times, values, bitMaps, 0, rowCount, null); - boolean hasTimeColDeletedMap = stream.read() == 1; - if (hasTimeColDeletedMap) { - int length = ReadWriteForEncodingUtils.readVarInt(stream); + tvList.deserializeTimeDeletion(stream); + return tvList; + } + + protected final void deserializeTimeDeletion(DataInputStream stream) throws IOException { + if (stream.read() == 1) { + int length = stream.readInt(); byte[] bytes = new byte[length]; stream.readFully(bytes); - tvList.timeColDeletedMap = new BitMap(rowCount, bytes); + timeColDeletedMap = new BitMap(rowCount, bytes); + for (int i = 0; i < rowCount; i++) { + if (isTimeDeleted(i)) { + timeDeletedCnt++; + } + } + updateMinTime(); + updateMaxTime(); } - - return tvList; } public BitMap getTimeColDeletedMap() { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java index 1206eed5476..8bead9f3a3f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/ColumnSortedAlignedTVList.java @@ -27,7 +27,6 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.Pair; -import org.apache.tsfile.utils.ReadWriteForEncodingUtils; import org.apache.tsfile.utils.ReadWriteIOUtils; import java.io.DataInputStream; @@ -37,6 +36,7 @@ import java.util.HashSet; import java.util.List; import java.util.Objects; import java.util.Set; +import java.util.function.IntPredicate; /** * An aligned TVList that orders rows by timestamp followed by selected value columns. Selected @@ -177,6 +177,19 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { } } + @Override + public synchronized int deleteMatchingRows( + long lowerBound, long upperBound, IntPredicate rowMatches, List<Integer> columnIndices) { + int deleted = super.deleteMatchingRows(lowerBound, upperBound, rowMatches, columnIndices); + if (deleted > 0 + && columnIndices != null + && columnIndices.stream().anyMatch(sortColumnIndices::contains)) { + valueColumnsSorted = false; + seqRowCount = 0; + } + return deleted; + } + @Override public synchronized Pair<Integer, Boolean> delete( long lowerBound, long upperBound, int columnIndex) { @@ -333,12 +346,7 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { } tvList.putAlignedValues(times, values, bitMaps, 0, rowCount, null); - if (stream.read() == 1) { - int length = ReadWriteForEncodingUtils.readVarInt(stream); - byte[] bytes = new byte[length]; - stream.readFully(bytes); - tvList.timeColDeletedMap = new BitMap(rowCount, bytes); - } + tvList.deserializeTimeDeletion(stream); return tvList; }
