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 50207ed6c5bc0cfc348a2499f23da45f35dd74be Author: Tian Jiang <[email protected]> AuthorDate: Tue Sep 1 16:04:08 2026 +0800 tmp --- .../memtable/AlignedWritableMemChunk.java | 50 +++++++++++++++++++--- 1 file changed, 44 insertions(+), 6 deletions(-) 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 513dacbab33..6172c2c8163 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 @@ -128,12 +128,18 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { boolean useColumnSortedList, EncryptParameter encryptParameter) { this.measurementIndexMap = new LinkedHashMap<>(); + this.measurementCategoryMap = new LinkedHashMap<>(); this.dataTypes = new ArrayList<>(); - this.schemaList = schemaList; + // Keep the chunk's schema order mutable because late schema discovery may insert a TAG. + this.schemaList = new ArrayList<>(schemaList); this.sortColumnCount = useColumnSortedList ? sortColumnCount : 0; this.useColumnSortedList = useColumnSortedList; for (int i = 0; i < schemaList.size(); i++) { measurementIndexMap.put(schemaList.get(i).getMeasurementName(), i); + if (i < this.sortColumnCount) { + measurementCategoryMap.put( + schemaList.get(i).getMeasurementName(), TsTableColumnCategory.TAG); + } dataTypes.add(schemaList.get(i).getType()); } this.list = @@ -165,7 +171,8 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { int sortColumnCount, EncryptParameter encryptParameter) { this.measurementIndexMap = new LinkedHashMap<>(); - this.schemaList = schemaList; + this.measurementCategoryMap = new LinkedHashMap<>(); + this.schemaList = new ArrayList<>(schemaList); this.sortColumnCount = list instanceof ColumnSortedAlignedTVList columnSortedList ? columnSortedList.getSortColumnCount() @@ -173,9 +180,13 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { this.useColumnSortedList = list instanceof ColumnSortedAlignedTVList; for (int i = 0; i < schemaList.size(); i++) { measurementIndexMap.put(schemaList.get(i).getMeasurementName(), i); + if (i < this.sortColumnCount) { + measurementCategoryMap.put( + schemaList.get(i).getMeasurementName(), TsTableColumnCategory.TAG); + } } this.list = list; - this.dataTypes = list.getTsDataTypes(); + this.dataTypes = this.list.getTsDataTypes(); this.sortedList = new ArrayList<>(); this.ignoreAllNullRows = !isTableModel; this.encryptParameter = encryptParameter; @@ -286,14 +297,23 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { ? AlignedTVListFactory.createAlignedList(new ArrayList<>(dataTypes)) : AlignedTVListFactory.createColumnSortedAlignedList( new ArrayList<>(dataTypes), this.sortColumnCount); - this.dataTypes = list.getTsDataTypes(); + this.dataTypes = this.list.getTsDataTypes(); } @Override public void writeAlignedPoints( long insertTime, Object[] objectValue, List<IMeasurementSchema> schemaList) { + writeAlignedPoints(insertTime, objectValue, schemaList, null); + } + + public void writeAlignedPoints( + long insertTime, + Object[] objectValue, + List<IMeasurementSchema> schemaList, + TsTableColumnCategory[] columnCategories) { Object[] reorderedValue = - checkAndReorderColumnValuesInInsertPlan(schemaList, objectValue, null).left; + checkAndReorderColumnValuesInInsertPlan(schemaList, objectValue, null, columnCategories) + .left; putAlignedRow(insertTime, reorderedValue); if (TVLIST_SORT_THRESHOLD > 0 && list.rowCount() >= TVLIST_SORT_THRESHOLD) { handoverAlignedTvList(); @@ -305,7 +325,13 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { || !(list instanceof ColumnSortedAlignedTVList columnSortedList)) { return; } + // Historical lists keep the prefix they were sorted with; the comparator treats later key + // columns as null instead of invalidating and resorting those lists. columnSortedList.addSortColumnCount(sortColumnCount); + for (int columnIndex = this.sortColumnCount; columnIndex < sortColumnCount; columnIndex++) { + measurementCategoryMap.put( + schemaList.get(columnIndex).getMeasurementName(), TsTableColumnCategory.TAG); + } this.sortColumnCount = sortColumnCount; } @@ -318,8 +344,20 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { int start, int end, TSStatus[] results) { + writeAlignedTablet(times, valueList, bitMaps, schemaList, null, start, end, results); + } + + public void writeAlignedTablet( + long[] times, + Object[] valueList, + BitMap[] bitMaps, + List<IMeasurementSchema> schemaList, + TsTableColumnCategory[] columnCategories, + int start, + int end, + TSStatus[] results) { Pair<Object[], BitMap[]> pair = - checkAndReorderColumnValuesInInsertPlan(schemaList, valueList, bitMaps); + checkAndReorderColumnValuesInInsertPlan(schemaList, valueList, bitMaps, columnCategories); Object[] reorderedColumnValues = pair.left; BitMap[] reorderedBitMaps = pair.right; putAlignedTablet(times, reorderedColumnValues, reorderedBitMaps, start, end, results);
