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 5ad7441e3f9a17f8711fbc114eee194dd3f743fc Author: Tian Jiang <[email protected]> AuthorDate: Wed Sep 23 10:02:20 2026 +0800 tmp --- .../db/utils/datastructure/AlignedTVList.java | 50 ++++++++++++++++------ 1 file changed, 36 insertions(+), 14 deletions(-) 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 0111159ff1a..ff26cce7e9a 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 @@ -104,6 +104,7 @@ public abstract class AlignedTVList extends TVList { private final Set<Integer> retainedColumns; private final long sourceBitmapMemoryCost; private final long cloneBitmapMemoryCost; + private final Runnable commitAction; private boolean committed; @@ -118,6 +119,18 @@ public abstract class AlignedTVList extends TVList { this.retainedColumns = retainedColumns; this.sourceBitmapMemoryCost = sourceBitmapMemoryCost; this.cloneBitmapMemoryCost = cloneBitmapMemoryCost; + this.commitAction = null; + } + + // Specialized layouts can prepare their own storage transfer while preserving the same + // allocation-before-publication contract used by MemChunk. + PartialClonePlan(AlignedTVList cloneList, Runnable commitAction) { + this.sourceList = null; + this.cloneList = cloneList; + this.retainedColumns = null; + this.sourceBitmapMemoryCost = 0; + this.cloneBitmapMemoryCost = 0; + this.commitAction = commitAction; } public AlignedTVList getCloneList() { @@ -129,7 +142,11 @@ public abstract class AlignedTVList extends TVList { if (committed) { return; } - sourceList.commitPartialClone(this); + if (commitAction == null) { + sourceList.commitPartialClone(this); + } else { + commitAction.run(); + } committed = true; } } @@ -621,18 +638,16 @@ public abstract class AlignedTVList extends TVList { if (valueIndex >= rowCount) { throw new ArrayIndexOutOfBoundsException(valueIndex); } - TsPrimitiveType[] vector = new TsPrimitiveType[values.size()]; - for (int columnIndex = 0; columnIndex < values.size(); columnIndex++) { - List<Object> columnValues = values.get(columnIndex); + // Logical columns may be backed by device metadata instead of a physical value array. + TsPrimitiveType[] vector = new TsPrimitiveType[dataTypes.size()]; + for (int columnIndex = 0; columnIndex < dataTypes.size(); columnIndex++) { int validValueIndex; if (validIndexesForTimeDuplicatedRows != null) { validValueIndex = validIndexesForTimeDuplicatedRows[columnIndex]; } else { validValueIndex = valueIndex; } - int arrayIndex = validValueIndex / ARRAY_SIZE; - int elementIndex = validValueIndex % ARRAY_SIZE; - if (columnValues == null || isNullValue(validValueIndex, columnIndex)) { + if (isNullValue(validValueIndex, columnIndex)) { continue; } switch (dataTypes.get(columnIndex)) { @@ -644,7 +659,7 @@ public abstract class AlignedTVList extends TVList { vector[columnIndex] = TsPrimitiveType.getByType(TSDataType.TEXT, valueT); break; case FLOAT: - float valueF = ((float[]) columnValues.get(arrayIndex))[elementIndex]; + float valueF = getFloatByValueIndex(validValueIndex, columnIndex); if (floatPrecision != null && encodingList != null && !Float.isNaN(valueF) @@ -656,16 +671,16 @@ public abstract class AlignedTVList extends TVList { break; case INT32: case DATE: - int valueI = ((int[]) columnValues.get(arrayIndex))[elementIndex]; + int valueI = getIntByValueIndex(validValueIndex, columnIndex); vector[columnIndex] = TsPrimitiveType.getByType(TSDataType.INT32, valueI); break; case INT64: case TIMESTAMP: - long valueL = ((long[]) columnValues.get(arrayIndex))[elementIndex]; + long valueL = getLongByValueIndex(validValueIndex, columnIndex); vector[columnIndex] = TsPrimitiveType.getByType(TSDataType.INT64, valueL); break; case DOUBLE: - double valueD = ((double[]) columnValues.get(arrayIndex))[elementIndex]; + double valueD = getDoubleByValueIndex(validValueIndex, columnIndex); if (floatPrecision != null && encodingList != null && !Double.isNaN(valueD) @@ -676,7 +691,7 @@ public abstract class AlignedTVList extends TVList { vector[columnIndex] = TsPrimitiveType.getByType(TSDataType.DOUBLE, valueD); break; case BOOLEAN: - boolean valueB = ((boolean[]) columnValues.get(arrayIndex))[elementIndex]; + boolean valueB = getBooleanByValueIndex(validValueIndex, columnIndex); vector[columnIndex] = TsPrimitiveType.getByType(TSDataType.BOOLEAN, valueB); break; default: @@ -1569,7 +1584,7 @@ public abstract class AlignedTVList extends TVList { return bitMaps.get(columnIndex).get(arrayIndex); } - private boolean markNullValue(int columnIndex, int arrayIndex, int elementIndex) { + protected boolean markNullValue(int columnIndex, int arrayIndex, int elementIndex) { List<Object> columnValues = values.get(columnIndex); if (columnValues == null || isValueBlockNull(columnIndex, arrayIndex)) { return false; @@ -2236,6 +2251,14 @@ public abstract class AlignedTVList extends TVList { dataTypes.add(ReadWriteIOUtils.readDataType(stream)); } + return deserializeRows( + stream, AlignedTVListFactory.createAlignedList(new ArrayList<>(dataTypes)), dataTypes); + } + + static AlignedTVList deserializeRows( + DataInputStream stream, AlignedTVList tvList, List<TSDataType> dataTypes) throws IOException { + int dataTypeNum = dataTypes.size(); + int rowCount = stream.readInt(); // time long[] times = new long[rowCount]; @@ -2321,7 +2344,6 @@ public abstract class AlignedTVList extends TVList { bitMaps[columnIndex] = bitMap; } - AlignedTVList tvList = AlignedTVListFactory.createAlignedList(new ArrayList<>(dataTypes)); tvList.putAlignedValues(times, values, bitMaps, 0, rowCount, null); tvList.deserializeTimeDeletion(stream);
