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 830507ac56aec7b4e9f9c843781744a6d91ab737 Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 23 11:41:12 2026 +0800 tmp --- .../apache/iotdb/db/i18n/DataNodeMiscMessages.java | 2 + .../apache/iotdb/db/i18n/DataNodeMiscMessages.java | 2 + .../compaction/repair/RepairDataFileScanUtil.java | 12 +- .../memtable/AlignedWritableMemChunk.java | 171 ++++++++++++--------- .../dataregion/memtable/TsFileProcessor.java | 5 + .../utils/datastructure/AlignedTVListFactory.java | 7 +- .../datastructure/ColumnSortedAlignedTVList.java | 90 +---------- .../memtable/FlatDeviceMemTableDeletionTest.java | 35 +++-- .../memtable/FlatDeviceMemTableTest.java | 31 ++-- 9 files changed, 154 insertions(+), 201 deletions(-) diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java index e76211672ff..feca81a893f 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java @@ -1531,4 +1531,6 @@ public final class DataNodeMiscMessages { public static final String EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_MARK_NULL_VALUE_2893628E = "Missing value arrays for aligned column index %d during mark null value"; + public static final String EXCEPTION_DELETING_TAG_COLUMN_ARG_IS_NOT_SUPPORTED_FOR_FLAT_DEVICE_TABLE_ARG_9B440C82 = + "Deleting TAG column %d is not supported for flat device table %s."; } diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java index c9a3f49710f..da8a836c83b 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java @@ -1508,4 +1508,6 @@ public final class DataNodeMiscMessages { public static final String EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_MARK_NULL_VALUE_2893628E = "标记空值过程中缺少对齐列索引 %d 的值数组"; + public static final String EXCEPTION_DELETING_TAG_COLUMN_ARG_IS_NOT_SUPPORTED_FOR_FLAT_DEVICE_TABLE_ARG_9B440C82 = + "不支持删除索引为 %d 的 TAG 列（flat device 表：%s）。"; } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java index fabb79cd615..309aed2ca27 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/compaction/repair/RepairDataFileScanUtil.java @@ -19,7 +19,6 @@ package org.apache.iotdb.db.storageengine.dataregion.compaction.repair; -import java.util.Objects; import org.apache.iotdb.commons.schema.table.TsFileTableSchemaUtil; import org.apache.iotdb.db.i18n.StorageEngineMessages; import org.apache.iotdb.db.storageengine.dataregion.compaction.execute.exception.CompactionLastTimeCheckFailedException; @@ -70,12 +69,12 @@ import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; -import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import static org.apache.tsfile.read.reader.chunk.ChunkReader.decryptAndUncompressPageData; @@ -290,9 +289,7 @@ public class RepairDataFileScanUtil { IPointReader rows = chunkReader.nextPageData().getBatchDataIterator(); while (rows.hasNextTimeValuePair()) { TimeValuePair row = rows.nextTimeValuePair(); - if (previous != null - && compareAlignedRowKey(previous, row) - >= 0) { + if (previous != null && compareAlignedRowKey(previous, row) >= 0) { throw new CompactionLastTimeCheckFailedException( device.toString(), row.getTimestamp(), previous.getTimestamp()); } @@ -306,8 +303,9 @@ public class RepairDataFileScanUtil { TsPrimitiveType[] previousValues = previous.getValue().getVector(); TsPrimitiveType[] currentValues = current.getValue().getVector(); for (int i = 0; i < previousValues.length; i++) { - int compare = Objects.compare(previousValues[i].getBinary(), currentValues[i].getBinary(), - Binary::compareTo); + int compare = + Objects.compare( + previousValues[i].getBinary(), currentValues[i].getBinary(), Binary::compareTo); if (compare != 0) { return compare; } 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 e1eb0944600..0eee722067d 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 @@ -35,7 +35,6 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.iotdb.db.utils.datastructure.AlignedTVListFactory; import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo; -import org.apache.iotdb.db.utils.datastructure.ColumnSortedAlignedTVList; import org.apache.iotdb.db.utils.datastructure.FlatDeviceAlignedTVList; import org.apache.iotdb.db.utils.datastructure.MemPointIterator; import org.apache.iotdb.db.utils.datastructure.MemPointIteratorFactory; @@ -60,6 +59,7 @@ import org.apache.tsfile.write.schema.MeasurementSchema; import java.io.DataInputStream; import java.io.IOException; import java.util.ArrayList; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -69,7 +69,6 @@ import java.util.TreeMap; import java.util.concurrent.BlockingQueue; import java.util.concurrent.atomic.AtomicInteger; -import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; import static org.apache.iotdb.db.utils.ModificationUtils.isPointDeleted; public class AlignedWritableMemChunk extends AbstractWritableMemChunk { @@ -238,14 +237,14 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { this.measurementIndexMap = new LinkedHashMap<>(); this.measurementCategoryMap = new LinkedHashMap<>(); this.schemaList = new ArrayList<>(schemaList); - this.useColumnSortedList = list instanceof ColumnSortedAlignedTVList; + this.useColumnSortedList = list instanceof FlatDeviceAlignedTVList; if (list instanceof FlatDeviceAlignedTVList) { Objects.requireNonNull(physicalDeviceId); } this.physicalDeviceId = physicalDeviceId; this.sortColumnIndices = - this.useColumnSortedList - ? new ArrayList<>(((ColumnSortedAlignedTVList) list).getSortColumnIndices()) + list instanceof FlatDeviceAlignedTVList flat + ? flat.getTagColumnIndices() : new ArrayList<>(); for (int i = 0; i < schemaList.size(); i++) { measurementIndexMap.put(schemaList.get(i).getMeasurementName(), i); @@ -458,10 +457,9 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { boolean isTagColumn = useColumnSortedList && measurementCategory == TsTableColumnCategory.TAG; index = this.list.getTsDataTypes().size(); - insertColumn(index, measurementSchema); + insertColumn(index, measurementSchema, isTagColumn); if (isTagColumn) { sortColumnIndices.add(index); - ((ColumnSortedAlignedTVList) this.list).addSortColumn(index); } } if (index != i) { @@ -494,14 +492,17 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { } /** - * Appends a newly discovered measurement to the working list and updates the physical index map. + * Appends a newly discovered measurement and updates the logical column-index map. * - * <p>Flat-device TAG columns are appended physically and their indices are recorded separately in - * {@code sortColumnIndices}. Historical TVLists remain unchanged because queries may still hold - * their original column-index mapping. + * <p>Flat-device TAGs extend only device metadata, without allocating FIELD storage. Historical + * TVLists remain unchanged because queries may still hold their original column-index mapping. */ - private void insertColumn(int columnIndex, IMeasurementSchema measurementSchema) { - this.list.insertColumn(columnIndex, measurementSchema.getType()); + private void insertColumn(int columnIndex, IMeasurementSchema measurementSchema, boolean isTag) { + if (isTag && list instanceof FlatDeviceAlignedTVList flat) { + flat.addTagColumn(columnIndex, measurementSchema.getType()); + } else { + this.list.insertColumn(columnIndex, measurementSchema.getType()); + } this.schemaList.add(measurementSchema); this.measurementIndexMap.put(measurementSchema.getMeasurementName(), columnIndex); } @@ -582,76 +583,69 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { // timestamp: 1 bitmap: 011 // timestamp: 2 bitmap: 101 // timestamp: 4 bitmap: 110 - List<long[]> timestampsList = alignedTVList.getTimestamps(); - List<int[]> indicesList = alignedTVList.getIndices(); - int row = -1; - for (int i = 0; i < timestampsList.size(); i++) { - long[] timestamps = timestampsList.get(i); - int[] indices = indicesList == null ? null : indicesList.get(i); - int limit = (i == timestampsList.size() - 1) ? rowCount - i * ARRAY_SIZE : ARRAY_SIZE; - for (int j = 0; j < limit; j++) { - row++; - // the row is deleted - if (allValueColDeletedMap != null && allValueColDeletedMap.isMarked(row)) { - continue; - } - long timestamp = timestamps[j]; - if (globalTimeFilter != null && !globalTimeFilter.satisfy(timestamp, null)) { + for (int row = 0; row < rowCount; row++) { + int valueIndex = alignedTVList.getValueIndex(row); + // the row is deleted + if (alignedTVList.isTimeDeleted(row) + || (allValueColDeletedMap != null && allValueColDeletedMap.isMarked(valueIndex))) { + continue; + } + long timestamp = alignedTVList.getTime(row); + if (globalTimeFilter != null && !globalTimeFilter.satisfy(timestamp, null)) { + continue; + } + + // Note that this method will only perform bitmap unmarking on the first occurrence of a + // non-null value in multiple timestamps for the same column. + BitMap currentRowNullValueBitmap = null; + for (int column = 0; column < schemaList.size(); column++) { + // Late-added columns do not exist in historical lists and are therefore null there. + if (column >= alignedTVList.getTsDataTypes().size() + || alignedTVList.isNullValue(valueIndex, column)) { continue; } - // Note that this method will only perform bitmap unmarking on the first occurrence of a - // non-null value in multiple timestamps for the same column. - BitMap currentRowNullValueBitmap = null; - for (int column = 0; column < schemaList.size(); column++) { - // Late-added columns do not exist in historical lists and are therefore null there. - if (column >= alignedTVList.getTsDataTypes().size() - || alignedTVList.isNullValue(indices == null ? row : indices[j], column)) { - continue; - } - - // skip deleted row - if (valueColumnsDeletionList != null && !valueColumnsDeletionList.isEmpty()) { - List<TimeRange> columnDeletionList = valueColumnsDeletionList.get(column); - int[] deleteCursor = valueColumnDeleteCursor.get(column); - if (columnDeletionList != null && !columnDeletionList.isEmpty()) { - if (!alignedTVList.isSorted() || !alignedTVList.isTimeOrdered()) { - deleteCursor[0] = 0; - } - if (isPointDeleted(timestamp, columnDeletionList, deleteCursor)) { - continue; - } + // skip deleted row + if (valueColumnsDeletionList != null && !valueColumnsDeletionList.isEmpty()) { + List<TimeRange> columnDeletionList = valueColumnsDeletionList.get(column); + int[] deleteCursor = valueColumnDeleteCursor.get(column); + if (columnDeletionList != null && !columnDeletionList.isEmpty()) { + if (!alignedTVList.isSorted() || !alignedTVList.isTimeOrdered()) { + deleteCursor[0] = 0; + } + if (isPointDeleted(timestamp, columnDeletionList, deleteCursor)) { + continue; } - } - if (!columnHasNonNullValue.isMarked(column)) { - hasNonNullValueColumnCount.incrementAndGet(); - columnHasNonNullValue.mark(column); - currentRowNullValueBitmap = - currentRowNullValueBitmap != null - ? currentRowNullValueBitmap - : timestampWithBitmap.computeIfAbsent( - timestamp, k -> getAllMarkedBitmap(schemaList.size())); - currentRowNullValueBitmap.unmark(column); } } - - if (!ignoreAllNullRows) { - timestampWithBitmap.put( - timestamp, + if (!columnHasNonNullValue.isMarked(column)) { + hasNonNullValueColumnCount.incrementAndGet(); + columnHasNonNullValue.mark(column); + currentRowNullValueBitmap = currentRowNullValueBitmap != null ? currentRowNullValueBitmap - : getAllMarkedBitmap(schemaList.size())); - return; - } - if (currentRowNullValueBitmap == null) { - continue; + : timestampWithBitmap.computeIfAbsent( + timestamp, k -> getAllMarkedBitmap(schemaList.size())); + currentRowNullValueBitmap.unmark(column); } - // found new column with non-null value - timestampWithBitmap.put(timestamp, currentRowNullValueBitmap); + } - if (hasNonNullValueColumnCount.get() == schemaList.size()) { - return; - } + if (!ignoreAllNullRows) { + timestampWithBitmap.put( + timestamp, + currentRowNullValueBitmap != null + ? currentRowNullValueBitmap + : getAllMarkedBitmap(schemaList.size())); + return; + } + if (currentRowNullValueBitmap == null) { + continue; + } + // found new column with non-null value + timestampWithBitmap.put(timestamp, currentRowNullValueBitmap); + + if (hasNonNullValueColumnCount.get() == schemaList.size()) { + return; } } } @@ -868,6 +862,18 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { BitMap allValueColDeletedMap; AlignedTVList alignedWorkingListForFlush = (AlignedTVList) workingListForFlush; + if (alignedWorkingListForFlush instanceof FlatDeviceAlignedTVList flat) { + FlatDeviceAlignedTVList.encode( + Collections.singletonList(flat), + activeSchemaList, + buildColumnIndexList(activeSchemaList), + encryptParameter, + ioTaskQueue, + maxNumberOfPointsInChunk, + maxNumberOfPointsInPage); + return; + } + allValueColDeletedMap = ignoreAllNullRows ? alignedWorkingListForFlush.getAllValueColDeletedMap() : null; @@ -1301,6 +1307,25 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { encodeInfo.maxNumberOfPointsInChunk, (encodeInfo.targetChunkSize / getAvgPointSizeOfLargestColumn())); + if (useColumnSortedList) { + List<FlatDeviceAlignedTVList> lists = new ArrayList<>(sortedList.size() + 1); + for (AlignedTVList historical : sortedList) { + lists.add((FlatDeviceAlignedTVList) historical); + } + lists.add((FlatDeviceAlignedTVList) workingListForFlush); + List<IMeasurementSchema> activeSchemas = getActiveSchemaList(); + FlatDeviceAlignedTVList.encode( + lists, + activeSchemas, + buildColumnIndexList(activeSchemas), + encryptParameter, + ioTaskQueue, + encodeInfo.maxNumberOfPointsInChunk, + encodeInfo.maxNumberOfPointsInPage); + encodeInfo.reset(); + return; + } + if (TVLIST_SORT_THRESHOLD == 0) { encodeWorkingAlignedTVList( ioTaskQueue, encodeInfo.maxNumberOfPointsInChunk, encodeInfo.maxNumberOfPointsInPage); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java index 7541f2fa468..8570124ac3b 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java @@ -81,6 +81,7 @@ import org.apache.iotdb.db.utils.EncryptDBUtils; import org.apache.iotdb.db.utils.MemUtils; import org.apache.iotdb.db.utils.ModificationUtils; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; +import org.apache.iotdb.db.utils.datastructure.FlatDeviceAlignedTVList; import org.apache.iotdb.db.utils.datastructure.TVList; import org.apache.iotdb.rpc.RpcUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -1354,6 +1355,10 @@ public class TsFileProcessor { if (alignedMemChunk == null || !alignedMemChunk.containsMeasurement(measurement)) { return false; } + if (alignedMemChunk.getWorkingTVList() instanceof FlatDeviceAlignedTVList flat) { + return flat.isValueArrayMaterialized( + alignedMemChunk.getMeasurementIndex(measurement), arrayIndex); + } List<Object> valueArrays = alignedMemChunk .getWorkingTVList() diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java index 4fc8f3662ac..e635b465a65 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListFactory.java @@ -75,8 +75,11 @@ public final class AlignedTVListFactory { int dataTypeCount = stream.readInt(); if (encodedDataType == FlatDeviceAlignedTVList.WAL_DATA_TYPE_MARKER) { Objects.requireNonNull(physicalDeviceId); - return ColumnSortedAlignedTVList.deserialize( - stream, dataTypeCount, true, physicalDeviceId.getTableName()); + return FlatDeviceAlignedTVList.deserialize( + stream, dataTypeCount, physicalDeviceId.getTableName()); + } + if (encodedDataType == ColumnSortedAlignedTVList.WAL_DATA_TYPE_MARKER) { + return ColumnSortedAlignedTVList.deserialize(stream, dataTypeCount); } return AlignedTVList.deserialize( 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 242fd968eb6..afc2524e0fe 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 @@ -24,7 +24,6 @@ import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; 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.ReadWriteIOUtils; @@ -253,8 +252,7 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { serializeWALRows(buffer); } - static ColumnSortedAlignedTVList deserialize( - DataInputStream stream, int dataTypeCount, boolean flatDevice, String tableName) + static ColumnSortedAlignedTVList deserialize(DataInputStream stream, int dataTypeCount) throws IOException { List<TSDataType> dataTypes = new ArrayList<>(dataTypeCount); for (int columnIndex = 0; columnIndex < dataTypeCount; columnIndex++) { @@ -266,91 +264,9 @@ public class ColumnSortedAlignedTVList extends AlignedTVList { sortColumnIndices.add(stream.readInt()); } ColumnSortedAlignedTVList tvList = - flatDevice - ? new FlatDeviceAlignedTVList(dataTypes, sortColumnIndices, true, tableName) - : new ColumnSortedAlignedTVList(dataTypes, sortColumnIndices, true); - - int rowCount = stream.readInt(); - long[] times = new long[rowCount]; - for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { - times[rowIndex] = stream.readLong(); - } - - Object[] values = new Object[dataTypeCount]; - BitMap[] bitMaps = new BitMap[dataTypeCount]; - for (int columnIndex = 0; columnIndex < dataTypeCount; columnIndex++) { - BitMap bitMap = BitMap.createBitMapDynamically(rowCount); - Object valuesOfOneColumn = - switch (dataTypes.get(columnIndex)) { - case TEXT, BLOB, STRING, OBJECT -> { - Binary[] binaryValues = new Binary[rowCount]; - for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { - binaryValues[rowIndex] = ReadWriteIOUtils.readBinary(stream); - if (ReadWriteIOUtils.readBool(stream)) { - bitMap.mark(rowIndex); - } - } - yield binaryValues; - } - case FLOAT -> { - float[] floatValues = new float[rowCount]; - for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { - floatValues[rowIndex] = stream.readFloat(); - if (ReadWriteIOUtils.readBool(stream)) { - bitMap.mark(rowIndex); - } - } - yield floatValues; - } - case INT32, DATE -> { - int[] intValues = new int[rowCount]; - for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { - intValues[rowIndex] = stream.readInt(); - if (ReadWriteIOUtils.readBool(stream)) { - bitMap.mark(rowIndex); - } - } - yield intValues; - } - case INT64, TIMESTAMP -> { - long[] longValues = new long[rowCount]; - for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { - longValues[rowIndex] = stream.readLong(); - if (ReadWriteIOUtils.readBool(stream)) { - bitMap.mark(rowIndex); - } - } - yield longValues; - } - case DOUBLE -> { - double[] doubleValues = new double[rowCount]; - for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { - doubleValues[rowIndex] = stream.readDouble(); - if (ReadWriteIOUtils.readBool(stream)) { - bitMap.mark(rowIndex); - } - } - yield doubleValues; - } - case BOOLEAN -> { - boolean[] booleanValues = new boolean[rowCount]; - for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { - booleanValues[rowIndex] = ReadWriteIOUtils.readBool(stream); - if (ReadWriteIOUtils.readBool(stream)) { - bitMap.mark(rowIndex); - } - } - yield booleanValues; - } - case VECTOR, UNKNOWN -> - throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT); - }; - values[columnIndex] = valuesOfOneColumn; - bitMaps[columnIndex] = bitMap; - } + new ColumnSortedAlignedTVList(dataTypes, sortColumnIndices, true); - tvList.putAlignedValues(times, values, bitMaps, 0, rowCount, null); - tvList.deserializeTimeDeletion(stream); + AlignedTVList.deserializeRows(stream, tvList, dataTypes); return tvList; } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java index 3d60d0754a7..d8ac7c3b37e 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableDeletionTest.java @@ -56,6 +56,7 @@ import java.util.concurrent.LinkedBlockingQueue; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; public class FlatDeviceMemTableDeletionTest { @@ -276,30 +277,32 @@ public class FlatDeviceMemTableDeletionTest { } } - // A selective TAG-column deletion changes the compound key, unlike a whole-row deletion. - // The next sort must restore TAG order without clearing the other row's TAG or FIELD. + // Reject TAG deletion before changing the logical device identity or the sorted rows. @Test - public void testSelectiveTagColumnDeletionInvalidatesSort() { + public void testSelectiveTagColumnDeletionIsRejected() { PrimitiveMemTable memTable = newMemTable(); insert(memTable, 1, "a", "x", 10); insert(memTable, 1, "b", "x", 20); AlignedTVList list = chunk(memTable).getWorkingTVList(); list.sort(); - assertEquals( - 1, - memTable.delete( - flatDeletion( - new DeletionPredicate( - TABLE, new TagPredicate.SegmentExactMatch("b", 1), List.of("tag1")), - new TimeRange(1, 1)))); - assertFalse(list.isSorted()); - list.sort(); - assertTrue(list.isNullValue(list.getValueIndex(0), 0)); - assertEquals(List.of(20L, 10L), remainingValues(list, 2)); + assertThrows( + UnsupportedOperationException.class, + () -> + memTable.delete( + flatDeletion( + new DeletionPredicate( + TABLE, new TagPredicate.SegmentExactMatch("b", 1), List.of("tag1")), + new TimeRange(1, 1)))); + assertThrows(UnsupportedOperationException.class, () -> chunk(memTable).removeColumn("tag1")); + assertTrue(chunk(memTable).containsMeasurement("tag1")); + assertTrue(list.isSorted()); + assertEquals(binary("a"), list.getBinaryByValueIndex(list.getValueIndex(0), 0)); + assertEquals(binary("b"), list.getBinaryByValueIndex(list.getValueIndex(1), 0)); + assertEquals(List.of(10L, 20L), remainingValues(list, 2)); } // Deletions at the first, middle and last sorted positions must not contribute to page sizes - // or value chunks. Exercise both flush encoders, including a physically dropped TAG column. + // or value chunks. Cover active-column projection after dropping a FIELD; TAGs remain intact. @Test public void testFlushSkipsDeletedRowsInEveryColumn() { for (boolean dropColumn : new boolean[] {false, true}) { @@ -313,7 +316,7 @@ public class FlatDeviceMemTableDeletionTest { } AlignedWritableMemChunk chunk = chunk(memTable); if (dropColumn) { - chunk.removeColumn("tag2"); + chunk.removeColumn("field1"); } chunk.sortTvListForFlush(); BlockingQueue<Object> queue = new LinkedBlockingQueue<>(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java index 03d3188ffdb..f4161c890b1 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/FlatDeviceMemTableTest.java @@ -28,7 +28,6 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalIn import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; import org.apache.iotdb.db.utils.datastructure.AlignedTVList; import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo; -import org.apache.iotdb.db.utils.datastructure.ColumnSortedAlignedTVList; import org.apache.iotdb.db.utils.datastructure.FlatDeviceAlignedTVList; import org.apache.tsfile.common.conf.TSFileConfig; @@ -117,16 +116,16 @@ public class FlatDeviceMemTableTest { Assert.assertEquals(TSEncoding.DICTIONARY, chunk.getSchemaList().get(1).getEncodingType()); Assert.assertNotEquals(TSEncoding.DICTIONARY, chunk.getSchemaList().get(2).getEncodingType()); AlignedTVList tvList = chunk.getWorkingTVList(); - Assert.assertTrue(tvList instanceof ColumnSortedAlignedTVList); - Assert.assertEquals(2, ((ColumnSortedAlignedTVList) tvList).getSortColumnCount()); + Assert.assertTrue(tvList instanceof FlatDeviceAlignedTVList); + Assert.assertEquals(2, ((FlatDeviceAlignedTVList) tvList).getSortColumnCount()); tvList.sort(); Assert.assertArrayEquals(new long[] {10, 15, 20, 30}, payloads(tvList)); Assert.assertTrue(tvList.isNullValue(tvList.getValueIndex(0), 0)); chunk.handoverAlignedTvList(); - Assert.assertTrue(chunk.getSortedList().get(0) instanceof ColumnSortedAlignedTVList); - Assert.assertTrue(chunk.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + Assert.assertTrue(chunk.getSortedList().get(0) instanceof FlatDeviceAlignedTVList); + Assert.assertTrue(chunk.getWorkingTVList() instanceof FlatDeviceAlignedTVList); } // The tablet keeps its complete logical device IDs for partition routing, while the MemTable @@ -160,7 +159,7 @@ public class FlatDeviceMemTableTest { TSEncoding.DICTIONARY, group.getAlignedMemChunk().getSchemaList().get(2).getEncodingType()); AlignedTVList tvList = group.getAlignedMemChunk().getWorkingTVList(); - Assert.assertTrue(tvList instanceof ColumnSortedAlignedTVList); + Assert.assertTrue(tvList instanceof FlatDeviceAlignedTVList); tvList.sort(); Assert.assertArrayEquals(new long[] {10, 11, 20}, payloads(tvList)); } @@ -221,7 +220,7 @@ public class FlatDeviceMemTableTest { }); Assert.assertEquals(2, chunk.rowCount()); Assert.assertEquals(3, chunk.getSortColumnCount()); - Assert.assertTrue(chunk.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + Assert.assertTrue(chunk.getWorkingTVList() instanceof FlatDeviceAlignedTVList); Assert.assertEquals( List.of("tag1", "field1", "tag2", "tag3"), chunk.getSchemaList().stream() @@ -276,11 +275,11 @@ public class FlatDeviceMemTableTest { Assert.assertEquals(2, historicalList.getTsDataTypes().size()); Assert.assertEquals(binary("a"), historicalList.getBinaryByValueIndex(0, 0)); Assert.assertEquals(10L, historicalList.getLongByValueIndex(0, 1)); - Assert.assertEquals(1, ((ColumnSortedAlignedTVList) historicalList).getSortColumnCount()); + Assert.assertEquals(1, ((FlatDeviceAlignedTVList) historicalList).getSortColumnCount()); Assert.assertEquals(4, currentList.getTsDataTypes().size()); Assert.assertEquals( - List.of(0, 2, 3), ((ColumnSortedAlignedTVList) currentList).getSortColumnIndices()); + List.of(0, 2, 3), ((FlatDeviceAlignedTVList) currentList).getSortColumnIndices()); Assert.assertEquals(binary("b"), currentList.getBinaryByValueIndex(0, 2)); Assert.assertEquals(binary("c"), currentList.getBinaryByValueIndex(0, 3)); Assert.assertEquals(20L, currentList.getLongByValueIndex(0, 1)); @@ -333,7 +332,7 @@ public class FlatDeviceMemTableTest { .getAlignedMemChunk(); Assert.assertEquals(2, chunk.getSortColumnCount()); AlignedTVList tvList = chunk.getWorkingTVList(); - Assert.assertEquals(List.of(0, 2), ((ColumnSortedAlignedTVList) tvList).getSortColumnIndices()); + Assert.assertEquals(List.of(0, 2), ((FlatDeviceAlignedTVList) tvList).getSortColumnIndices()); Assert.assertEquals( List.of("tag1", "field1", "tag3"), chunk.getSchemaList().stream() @@ -362,7 +361,7 @@ public class FlatDeviceMemTableTest { .map(IMeasurementSchema::getMeasurementName) .collect(Collectors.toList())); Assert.assertEquals( - List.of(0), ((ColumnSortedAlignedTVList) chunk.getWorkingTVList()).getSortColumnIndices()); + List.of(0), ((FlatDeviceAlignedTVList) chunk.getWorkingTVList()).getSortColumnIndices()); chunk.getWorkingTVList().sort(); Assert.assertEquals( binary("a"), @@ -397,7 +396,7 @@ public class FlatDeviceMemTableTest { .map(IMeasurementSchema::getMeasurementName) .collect(Collectors.toList())); Assert.assertEquals( - List.of(0), ((ColumnSortedAlignedTVList) chunk.getWorkingTVList()).getSortColumnIndices()); + List.of(0), ((FlatDeviceAlignedTVList) chunk.getWorkingTVList()).getSortColumnIndices()); chunk.getWorkingTVList().sort(); Assert.assertEquals( binary("x"), @@ -437,12 +436,12 @@ public class FlatDeviceMemTableTest { AlignedWritableMemChunk recovered = AlignedWritableMemChunk.deserialize( new DataInputStream(new ByteArrayInputStream(buffer.array())), true, PHYSICAL_DEVICE); - Assert.assertTrue(recovered.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + Assert.assertTrue(recovered.getWorkingTVList() instanceof FlatDeviceAlignedTVList); Assert.assertEquals( - 2, ((ColumnSortedAlignedTVList) recovered.getWorkingTVList()).getSortColumnCount()); + 2, ((FlatDeviceAlignedTVList) recovered.getWorkingTVList()).getSortColumnCount()); recovered.handoverAlignedTvList(); - Assert.assertTrue(recovered.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + Assert.assertTrue(recovered.getWorkingTVList() instanceof FlatDeviceAlignedTVList); } // Covers the direct MemChunk flush path used when TVList handover is disabled. Equal timestamps @@ -601,7 +600,7 @@ public class FlatDeviceMemTableTest { private static void assertNoSortColumns(AlignedWritableMemChunkGroup group) { AlignedWritableMemChunk chunk = group.getAlignedMemChunk(); Assert.assertEquals(0, chunk.getSortColumnCount()); - Assert.assertFalse(chunk.getWorkingTVList() instanceof ColumnSortedAlignedTVList); + Assert.assertFalse(chunk.getWorkingTVList() instanceof FlatDeviceAlignedTVList); } private static Binary binary(String value) {
