You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/20 06:33:29 UTC
(iotdb) branch dev/1.3 updated: [dev/1.3] Batch aligned memtable column expansion to reduce region lock contention (#18830)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 8cee2db544b [dev/1.3] Batch aligned memtable column expansion to reduce region lock contention (#18830)
8cee2db544b is described below
commit 8cee2db544b73922d470c30a51e4de5bbe4cafc5
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Tue Oct 20 14:33:23 2026 +0800
[dev/1.3] Batch aligned memtable column expansion to reduce region lock contention (#18830)
* Batch aligned memtable column expansion to reduce region lock contention
(cherry picked from commit 1f05d5d3fb858519d580e4df6c1b766e47c03f63)
* Keep aligned writes without schema expansion on a single pass
(cherry picked from commit 5f27eb1ac3bc00f8568f672a14cc1ef9bdf581a3)
* Preserve implicit nulls and WAL serialization for lazy aligned columns
---
.../memtable/AlignedWritableMemChunk.java | 39 ++--
.../db/utils/datastructure/AlignedTVList.java | 67 ++++---
.../memtable/AlignedWritableMemChunkTest.java | 217 +++++++++++++++++++++
.../db/utils/datastructure/AlignedTVListTest.java | 124 ++++++++++++
4 files changed, 410 insertions(+), 37 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 2c53a3ca916..ed02cac5fed 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
@@ -1083,12 +1083,12 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk {
}
/**
- * Check metadata of columns and return array that mapping existed metadata to index of data
- * column.
+ * Register missing measurements in insertion order, then reorder values and bitmaps into the
+ * memchunk's column order.
*
* @param schemaListInInsertPlan Contains all existed schema in InsertPlan. If some timeseries
* have been deleted, there will be null in its slot.
- * @return columnIndexArray: schemaList[i] is schema of columns[columnIndexArray[i]]
+ * @return values and bitmaps aligned with this memchunk's schema list
*/
private Pair<Object[], BitMap[]> checkAndReorderColumnValuesInInsertPlan(
List<IMeasurementSchema> schemaListInInsertPlan, Object[] columnValues, BitMap[] bitMaps) {
@@ -1097,20 +1097,16 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk {
for (int i = 0; i < schemaListInInsertPlan.size(); i++) {
IMeasurementSchema measurementSchema = schemaListInInsertPlan.get(i);
if (measurementSchema != null) {
- Integer index = this.measurementIndexMap.get(measurementSchema.getMeasurementId());
- // Index is null means this measurement was not in this AlignedTVList before.
- // We need to extend a new column in AlignedMemChunk and AlignedTVList.
- // And the reorderedColumnValues should extend one more column for the new measurement
+ Integer index = measurementIndexMap.get(measurementSchema.getMeasurementId());
if (index == null) {
- index = this.list.getTsDataTypes().size();
- this.measurementIndexMap.put(schemaListInInsertPlan.get(i).getMeasurementId(), index);
- this.schemaList.add(schemaListInInsertPlan.get(i));
- this.list.extendColumn(schemaListInInsertPlan.get(i).getType());
- reorderedColumnValues =
- Arrays.copyOf(reorderedColumnValues, reorderedColumnValues.length + 1);
+ // Keep writes without new measurements on a single-pass path. Only the first missing
+ // measurement triggers a batch extension and one resize of the reordered arrays.
+ extendColumnsInInsertPlan(schemaListInInsertPlan, i);
+ reorderedColumnValues = Arrays.copyOf(reorderedColumnValues, schemaList.size());
if (reorderedBitMaps != null) {
- reorderedBitMaps = Arrays.copyOf(reorderedBitMaps, reorderedBitMaps.length + 1);
+ reorderedBitMaps = Arrays.copyOf(reorderedBitMaps, schemaList.size());
}
+ index = measurementIndexMap.get(measurementSchema.getMeasurementId());
}
reorderedColumnValues[index] = columnValues[i];
if (bitMaps != null) {
@@ -1121,6 +1117,21 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk {
return new Pair<>(reorderedColumnValues, reorderedBitMaps);
}
+ private void extendColumnsInInsertPlan(
+ List<IMeasurementSchema> schemaListInInsertPlan, int startIndex) {
+ List<TSDataType> newDataTypes = new ArrayList<>();
+ for (int i = startIndex; i < schemaListInInsertPlan.size(); i++) {
+ IMeasurementSchema measurementSchema = schemaListInInsertPlan.get(i);
+ if (measurementSchema != null
+ && !measurementIndexMap.containsKey(measurementSchema.getMeasurementId())) {
+ measurementIndexMap.put(measurementSchema.getMeasurementId(), schemaList.size());
+ schemaList.add(measurementSchema);
+ newDataTypes.add(measurementSchema.getType());
+ }
+ }
+ list.extendColumns(newDataTypes);
+ }
+
public long[] getAnySatisfiedTimestamp(
List<List<TimeRange>> deletionList, List<BitMap> bitMaps, Filter globalTimeFilter) {
BitMap columnHasNonNullValue = new BitMap(schemaList.size());
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 0b03aebaefc..89cd0230842 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
@@ -54,6 +54,7 @@ import java.io.DataInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
@@ -497,22 +498,30 @@ public abstract class AlignedTVList extends TVList {
}
public void extendColumn(TSDataType dataType) {
- List<Object> columnValue = new ArrayList<>();
- for (int i = 0; i < timestamps.size(); i++) {
- columnValue.add(null);
+ extendColumns(Collections.singletonList(dataType));
+ }
+
+ /** Add columns with implicit nulls for existing rows, resizing accounting arrays once. */
+ public void extendColumns(List<TSDataType> newDataTypes) {
+ if (newDataTypes.isEmpty()) {
+ return;
}
- if (bitMaps != null) {
- bitMaps.add(null);
+ for (TSDataType dataType : newDataTypes) {
+ List<Object> columnValue = new ArrayList<>(timestamps.size());
+ for (int i = 0; i < timestamps.size(); i++) {
+ // New columns contain only implicit nulls until their first non-null write.
+ columnValue.add(null);
+ }
+ if (bitMaps != null) {
+ bitMaps.add(null);
+ }
+ values.add(columnValue);
+ dataTypes.add(dataType);
}
- this.values.add(columnValue);
- this.dataTypes.add(dataType);
- materializedValueArrayCounts = Arrays.copyOf(materializedValueArrayCounts, dataTypes.size());
+ // Recompute the full-width memory estimate only once for the entire batch.
refreshArrayMemCostWithoutIndex();
-
- long[] tmpValueChunkRawSize = memoryBinaryChunkSize;
- memoryBinaryChunkSize = new long[dataTypes.size()];
- System.arraycopy(
- tmpValueChunkRawSize, 0, memoryBinaryChunkSize, 0, tmpValueChunkRawSize.length);
+ memoryBinaryChunkSize = Arrays.copyOf(memoryBinaryChunkSize, dataTypes.size());
+ materializedValueArrayCounts = Arrays.copyOf(materializedValueArrayCounts, dataTypes.size());
}
/**
@@ -617,12 +626,16 @@ public abstract class AlignedTVList extends TVList {
if (columnIndex < 0 || columnIndex >= values.size() || values.get(columnIndex) == null) {
return true;
}
+ int arrayIndex = unsortedRowIndex / ARRAY_SIZE;
+ // A lazy value block contains only implicit nulls, even when a bitmap already exists.
+ if (values.get(columnIndex).get(arrayIndex) == null) {
+ return true;
+ }
if (bitMaps == null
|| bitMaps.get(columnIndex) == null
- || bitMaps.get(columnIndex).get(unsortedRowIndex / ARRAY_SIZE) == null) {
- return values.get(columnIndex).get(unsortedRowIndex / ARRAY_SIZE) == null;
+ || bitMaps.get(columnIndex).get(arrayIndex) == null) {
+ return false;
}
- int arrayIndex = unsortedRowIndex / ARRAY_SIZE;
int elementIndex = unsortedRowIndex % ARRAY_SIZE;
List<BitMap> columnBitMaps = bitMaps.get(columnIndex);
return columnBitMaps.get(arrayIndex).isMarked(elementIndex);
@@ -1748,8 +1761,14 @@ public abstract class AlignedTVList extends TVList {
case TEXT:
case BLOB:
case STRING:
+ List<Object> columnValues = values.get(columnIndex);
for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) {
- size += ReadWriteIOUtils.sizeToWrite(getBinaryByValueIndex(rowIdx, columnIndex));
+ Binary value =
+ columnValues == null || columnValues.get(rowIdx / ARRAY_SIZE) == null
+ ? null
+ : getBinaryByValueIndex(rowIdx, columnIndex);
+ // Null blocks and null Binary entries use an empty Binary placeholder in the WAL.
+ size += value == null ? Integer.BYTES : ReadWriteIOUtils.sizeToWrite(value);
}
break;
case FLOAT:
@@ -1796,12 +1815,14 @@ public abstract class AlignedTVList extends TVList {
for (int rowIndex = 0; rowIndex < rowCount; ++rowIndex) {
int arrayIndex = rowIndex / ARRAY_SIZE;
int elementIndex = rowIndex % ARRAY_SIZE;
+ // Keep the WAL layout unchanged without materializing lazy all-null value blocks.
+ Object valueArray = columnValues == null ? null : columnValues.get(arrayIndex);
// value
switch (dataTypes.get(columnIndex)) {
case TEXT:
case BLOB:
case STRING:
- Binary valueT = ((Binary[]) columnValues.get(arrayIndex))[elementIndex];
+ Binary valueT = valueArray == null ? null : ((Binary[]) valueArray)[elementIndex];
// In some scenario, the Binary in AlignedTVList will be null if this field is empty in
// current row. We need to handle this scenario to get rid of NPE. See the similar issue
// here: https://github.com/apache/iotdb/pull/9884
@@ -1815,25 +1836,25 @@ public abstract class AlignedTVList extends TVList {
}
break;
case FLOAT:
- float valueF = ((float[]) columnValues.get(arrayIndex))[elementIndex];
+ float valueF = valueArray == null ? 0 : ((float[]) valueArray)[elementIndex];
buffer.putFloat(valueF);
break;
case INT32:
case DATE:
- int valueI = ((int[]) columnValues.get(arrayIndex))[elementIndex];
+ int valueI = valueArray == null ? 0 : ((int[]) valueArray)[elementIndex];
buffer.putInt(valueI);
break;
case INT64:
case TIMESTAMP:
- long valueL = ((long[]) columnValues.get(arrayIndex))[elementIndex];
+ long valueL = valueArray == null ? 0 : ((long[]) valueArray)[elementIndex];
buffer.putLong(valueL);
break;
case DOUBLE:
- double valueD = ((double[]) columnValues.get(arrayIndex))[elementIndex];
+ double valueD = valueArray == null ? 0 : ((double[]) valueArray)[elementIndex];
buffer.putDouble(valueD);
break;
case BOOLEAN:
- boolean valueB = ((boolean[]) columnValues.get(arrayIndex))[elementIndex];
+ boolean valueB = valueArray != null && ((boolean[]) valueArray)[elementIndex];
WALWriteUtils.write(valueB, buffer);
break;
default:
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkTest.java
new file mode 100644
index 00000000000..2d93ab7d4c3
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunkTest.java
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.storageengine.dataregion.memtable;
+
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest;
+import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
+
+import org.apache.tsfile.common.conf.TSFileConfig;
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.utils.Binary;
+import org.apache.tsfile.utils.BitMap;
+import org.apache.tsfile.write.schema.IMeasurementSchema;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+public class AlignedWritableMemChunkTest {
+
+ @Test
+ public void testTabletExtendsColumnsWithReorderingAndBitmaps() throws IOException {
+ IMeasurementSchema existing = new MeasurementSchema("existing", TSDataType.INT64);
+ IMeasurementSchema omitted = new MeasurementSchema("omitted", TSDataType.BOOLEAN);
+ IMeasurementSchema integer = new MeasurementSchema("integer", TSDataType.INT32);
+ IMeasurementSchema text = new MeasurementSchema("text", TSDataType.TEXT);
+ List<IMeasurementSchema> initialSchemas = Arrays.asList(existing, omitted);
+ AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(new ArrayList<>(initialSchemas));
+ chunk.writeAlignedPoints(0, new Object[] {1L, true}, initialSchemas);
+ Binary binary = new Binary("value", TSFileConfig.STRING_CHARSET);
+ BitMap integerBitmap = new BitMap(3);
+ integerBitmap.mark(2);
+ BitMap textBitmap = new BitMap(3);
+ textBitmap.mark(1);
+ BitMap existingBitmap = new BitMap(3);
+ existingBitmap.mark(1);
+ chunk.writeAlignedTablet(
+ new long[] {-1, 1, 2},
+ new Object[] {
+ new long[] {0, 10, 20},
+ new int[] {0, 1, 2},
+ null,
+ new Binary[] {binary, binary, binary},
+ new int[] {0, 11, 12}
+ },
+ new BitMap[] {existingBitmap, null, null, textBitmap, integerBitmap},
+ Arrays.asList(existing, integer, null, text, integer),
+ 1,
+ 3);
+
+ Assert.assertEquals(Arrays.asList(existing, omitted, integer, text), chunk.getSchemaList());
+ Assert.assertEquals(
+ Arrays.asList(2, 3), chunk.buildColumnIndexList(Arrays.asList(integer, text)));
+ AlignedTVList list = chunk.getWorkingTVList();
+ Assert.assertEquals(3, list.rowCount());
+ Assert.assertTrue(list.isNullValue(0, 2));
+ Assert.assertTrue(list.isNullValue(0, 3));
+ Assert.assertTrue(list.isNullValue(1, 1));
+ Assert.assertTrue(list.isNullValue(2, 1));
+ Assert.assertTrue(list.isNullValue(1, 0));
+ Assert.assertEquals(20L, list.getLongByValueIndex(2, 0));
+ Assert.assertEquals(11, list.getIntByValueIndex(1, 2));
+ Assert.assertTrue(list.isNullValue(2, 2));
+ Assert.assertTrue(list.isNullValue(1, 3));
+ Assert.assertEquals(binary, list.getBinaryByValueIndex(2, 3));
+
+ WALByteBufferForTest buffer =
+ new WALByteBufferForTest(ByteBuffer.allocate(chunk.serializedSize()));
+ chunk.serializeToWAL(buffer);
+ AlignedWritableMemChunk restored =
+ AlignedWritableMemChunk.deserialize(
+ new DataInputStream(new ByteArrayInputStream(buffer.getBuffer().array())));
+ Assert.assertEquals(chunk.getSchemaList(), restored.getSchemaList());
+ AlignedTVList restoredList = restored.getWorkingTVList();
+ for (int row = 0; row < list.rowCount(); row++) {
+ Assert.assertEquals(list.getTime(row), restoredList.getTime(row));
+ Assert.assertEquals(
+ list.getAlignedValue(row).toString(), restoredList.getAlignedValue(row).toString());
+ }
+ chunk.release();
+ restored.release();
+ }
+
+ @Test
+ public void testTabletWithoutExtensionReordersExistingColumnsAndBitmaps() {
+ IMeasurementSchema first = new MeasurementSchema("first", TSDataType.INT64);
+ IMeasurementSchema second = new MeasurementSchema("second", TSDataType.INT32);
+ IMeasurementSchema omitted = new MeasurementSchema("omitted", TSDataType.BOOLEAN);
+ List<IMeasurementSchema> schemas = new ArrayList<>(Arrays.asList(first, second, omitted));
+ AlignedWritableMemChunk chunk = new AlignedWritableMemChunk(schemas);
+ chunk.writeAlignedPoints(0, new Object[] {1L, 2, true}, schemas);
+ AlignedTVList list = chunk.getWorkingTVList();
+ List<TSDataType> dataTypes = list.getTsDataTypes();
+ BitMap firstBitmap = new BitMap(3);
+ firstBitmap.mark(1);
+ BitMap secondBitmap = new BitMap(3);
+ secondBitmap.mark(2);
+ chunk.writeAlignedTablet(
+ new long[] {-1, 1, 2},
+ new Object[] {new int[] {0, 3, 4}, null, new long[] {0, 5, 6}, new int[] {0, 7, 8}},
+ new BitMap[] {null, null, firstBitmap, secondBitmap},
+ Arrays.asList(second, null, first, second),
+ 1,
+ 3);
+
+ Assert.assertEquals(Arrays.asList(first, second, omitted), chunk.getSchemaList());
+ Assert.assertSame(dataTypes, list.getTsDataTypes());
+ Assert.assertEquals(3, dataTypes.size());
+ Assert.assertEquals(3, list.rowCount());
+ Assert.assertTrue(list.isNullValue(1, 0));
+ Assert.assertEquals(6L, list.getLongByValueIndex(2, 0));
+ Assert.assertEquals(7, list.getIntByValueIndex(1, 1));
+ Assert.assertTrue(list.isNullValue(2, 1));
+ Assert.assertTrue(list.isNullValue(1, 2));
+ Assert.assertTrue(list.isNullValue(2, 2));
+ chunk.release();
+ }
+
+ @Test
+ public void testPointsExtendColumnsAfterRemovalAndHandover() {
+ IMeasurementSchema original = new MeasurementSchema("original", TSDataType.INT64);
+ IMeasurementSchema added = new MeasurementSchema("added", TSDataType.INT32);
+ AlignedWritableMemChunk chunk =
+ new AlignedWritableMemChunk(new ArrayList<>(Collections.singletonList(original)));
+ chunk.writeAlignedPoints(0, new Object[] {1L}, Collections.singletonList(original));
+ chunk.removeColumn("original");
+ chunk.writeAlignedPoints(1, new Object[] {7, null, 2L}, Arrays.asList(added, null, original));
+
+ Assert.assertEquals(3, chunk.getSchemaList().size());
+ Assert.assertEquals(
+ Arrays.asList(1, 2), chunk.buildColumnIndexList(Arrays.asList(added, original)));
+ AlignedTVList list = chunk.getWorkingTVList();
+ Assert.assertTrue(list.isNullValue(0, 1));
+ Assert.assertTrue(list.isNullValue(0, 2));
+ Assert.assertEquals(7, list.getIntByValueIndex(1, 1));
+ Assert.assertEquals(2L, list.getLongByValueIndex(1, 2));
+ chunk.handoverAlignedTvList();
+
+ IMeasurementSchema afterHandover = new MeasurementSchema("after", TSDataType.DOUBLE);
+ chunk.writeAlignedPoints(
+ 2, new Object[] {3.5, 3L, 8}, Arrays.asList(afterHandover, original, added));
+ Assert.assertEquals(
+ Collections.singletonList(3),
+ chunk.buildColumnIndexList(Collections.singletonList(afterHandover)));
+ Assert.assertEquals(3, chunk.getSortedList().get(0).getTsDataTypes().size());
+ Assert.assertEquals(4, chunk.getWorkingTVList().getTsDataTypes().size());
+ Assert.assertEquals(8, chunk.getWorkingTVList().getIntByValueIndex(0, 1));
+ Assert.assertEquals(3L, chunk.getWorkingTVList().getLongByValueIndex(0, 2));
+ Assert.assertEquals(3.5, chunk.getWorkingTVList().getDoubleByValueIndex(0, 3), 0);
+ chunk.release();
+ }
+
+ @Test
+ public void testWideTabletKeepsUnwrittenColumnsLazy() {
+ int columnCount = 10_000;
+ IMeasurementSchema existing = new MeasurementSchema("existing", TSDataType.INT64);
+ AlignedWritableMemChunk chunk =
+ new AlignedWritableMemChunk(new ArrayList<>(Collections.singletonList(existing)));
+ chunk.writeAlignedPoints(0, new Object[] {1L}, Collections.singletonList(existing));
+ List<IMeasurementSchema> incomingSchemas = new ArrayList<>(columnCount);
+ Object[] values = new Object[columnCount];
+ for (int column = 0; column < columnCount; column++) {
+ incomingSchemas.add(new MeasurementSchema("s" + column, TSDataType.INT64));
+ }
+ values[0] = new long[] {2L};
+ values[columnCount - 1] = new long[] {3L};
+ chunk.writeAlignedTablet(new long[] {1}, values, null, incomingSchemas, 0, 1);
+
+ AlignedTVList list = chunk.getWorkingTVList();
+ Assert.assertEquals(columnCount + 1, list.getTsDataTypes().size());
+ Assert.assertEquals(columnCount + 1, chunk.getSchemaList().size());
+ Assert.assertEquals(2, list.rowCount());
+ Assert.assertEquals(2L, list.getLongByValueIndex(1, 1));
+ Assert.assertEquals(3L, list.getLongByValueIndex(1, columnCount));
+ List<Integer> columnIndices = chunk.buildColumnIndexList(incomingSchemas);
+ for (int column = 1; column <= columnCount; column++) {
+ Assert.assertTrue(list.isNullValue(0, column));
+ Assert.assertEquals(column, columnIndices.get(column - 1).intValue());
+ if (column > 1 && column < columnCount) {
+ Assert.assertNull(list.getValues().get(column).get(0));
+ Assert.assertTrue(list.isNullValue(1, column));
+ }
+ }
+ chunk.writeAlignedPoints(
+ 2,
+ new Object[] {4L, 5L},
+ Arrays.asList(incomingSchemas.get(columnCount - 1), incomingSchemas.get(0)));
+ Assert.assertEquals(columnCount + 1, chunk.getSchemaList().size());
+ Assert.assertEquals(5L, list.getLongByValueIndex(2, 1));
+ Assert.assertEquals(4L, list.getLongByValueIndex(2, columnCount));
+ chunk.release();
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
index f76752894a0..2d94cf24c16 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java
@@ -18,6 +18,8 @@
*/
package org.apache.iotdb.db.utils.datastructure;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest;
+
import org.apache.commons.lang3.ArrayUtils;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.enums.TSDataType;
@@ -27,6 +29,10 @@ import org.apache.tsfile.utils.RamUsageEstimator;
import org.junit.Assert;
import org.junit.Test;
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -41,6 +47,124 @@ import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF;
public class AlignedTVListTest {
+ @Test
+ public void testExtendColumnsPreservesLazyBlocksAndAccounting() {
+ AlignedTVList list =
+ AlignedTVList.newAlignedList(
+ new ArrayList<>(Arrays.asList(TSDataType.INT64, TSDataType.TEXT)));
+ Binary text = new Binary("value", TSFileConfig.STRING_CHARSET);
+ for (int row = 0; row <= ARRAY_SIZE; row++) {
+ list.putAlignedValue(row, new Object[] {row == 1 ? null : (long) row, text});
+ }
+ long ramBefore = list.getRamSize();
+ long binarySizeBefore = list.memoryBinaryChunkSize[1];
+ list.extendColumns(Arrays.asList(TSDataType.INT32, TSDataType.TEXT, TSDataType.DOUBLE));
+
+ Assert.assertEquals(5, list.memoryBinaryChunkSize.length);
+ Assert.assertEquals(binarySizeBefore, list.memoryBinaryChunkSize[1]);
+ for (int column = 2; column < 5; column++) {
+ Assert.assertEquals(0, list.memoryBinaryChunkSize[column]);
+ Assert.assertEquals(2, list.getValues().get(column).size());
+ Assert.assertNull(list.getValues().get(column).get(0));
+ Assert.assertNull(list.getValues().get(column).get(1));
+ Assert.assertNull(list.getBitMaps().get(column));
+ for (int row = 0; row <= ARRAY_SIZE; row++) {
+ Assert.assertTrue(list.isNullValue(row, column));
+ }
+ }
+ Assert.assertTrue(list.getRamSize() > ramBefore);
+ Assert.assertEquals(list.getRamSize(null), list.getRamSize());
+
+ list.putAlignedValue(ARRAY_SIZE + 1L, new Object[] {null, null, 7, text, 3.5});
+ Assert.assertEquals(7, list.getIntByValueIndex(ARRAY_SIZE + 1, 2));
+ Assert.assertEquals(text, list.getBinaryByValueIndex(ARRAY_SIZE + 1, 3));
+ Assert.assertEquals(3.5, list.getDoubleByValueIndex(ARRAY_SIZE + 1, 4), 0);
+ Assert.assertTrue(list.memoryBinaryChunkSize[3] > 0);
+ for (int column = 2; column < 5; column++) {
+ Assert.assertNull(list.getValues().get(column).get(0));
+ Assert.assertTrue(list.isNullValue(ARRAY_SIZE, column));
+ }
+ Assert.assertEquals(list.getRamSize(null), list.getRamSize());
+ AlignedTVList clone = list.clone();
+ Assert.assertEquals(list.getRamSize(), clone.getRamSize());
+ Assert.assertEquals(list.getRamSize(), list.cloneForFlushSort().getRamSize());
+ clone.clear();
+ list.clear();
+ Assert.assertEquals(list.getRamSize(null), list.getRamSize());
+ Assert.assertTrue(list.getRamSize() < ramBefore);
+ list.putAlignedValue(0, new Object[] {1L, text, 2, text, 3.0});
+ Assert.assertEquals(list.getRamSize(null), list.getRamSize());
+ list.clear();
+ }
+
+ @Test
+ public void testExtendColumnsBeforeFirstWriteAndEmptyBatch() {
+ AlignedTVList list = AlignedTVList.newAlignedList(new ArrayList<>());
+ list.extendColumns(Arrays.asList(TSDataType.INT32, TSDataType.INT64));
+ long[] binarySizes = list.memoryBinaryChunkSize;
+ long ramSize = list.getRamSize();
+ list.extendColumns(Collections.emptyList());
+ Assert.assertSame(binarySizes, list.memoryBinaryChunkSize);
+ Assert.assertEquals(ramSize, list.getRamSize());
+ list.putAlignedValue(0, new Object[] {3, 4L});
+ Assert.assertEquals(3, list.getIntByValueIndex(0, 0));
+ Assert.assertEquals(4L, list.getLongByValueIndex(0, 1));
+ Assert.assertEquals(list.getRamSize(null), list.getRamSize());
+ list.clear();
+ }
+
+ @Test
+ public void testExtendedLazyColumnsRoundTripThroughWAL() throws IOException {
+ AlignedTVList list =
+ AlignedTVList.newAlignedList(new ArrayList<>(Collections.singletonList(TSDataType.INT64)));
+ for (int row = 0; row <= ARRAY_SIZE; row++) {
+ list.putAlignedValue(row, new Object[] {(long) row});
+ }
+ list.extendColumns(
+ Arrays.asList(
+ TSDataType.BOOLEAN,
+ TSDataType.INT32,
+ TSDataType.INT64,
+ TSDataType.FLOAT,
+ TSDataType.DOUBLE,
+ TSDataType.TEXT,
+ TSDataType.DATE,
+ TSDataType.TIMESTAMP,
+ TSDataType.BLOB,
+ TSDataType.STRING));
+ Binary text = new Binary("value", TSFileConfig.STRING_CHARSET);
+ list.putAlignedValue(
+ ARRAY_SIZE + 1L,
+ new Object[] {1L, true, 2, 3L, 4.5F, 6.5D, text, 20240101, 7L, text, null});
+
+ int serializedSize = list.serializedSize();
+ long ramSize = list.getRamSize();
+ WALByteBufferForTest buffer = new WALByteBufferForTest(ByteBuffer.allocate(serializedSize));
+ list.serializeToWAL(buffer);
+ Assert.assertEquals(serializedSize, buffer.position());
+ Assert.assertEquals(ramSize, list.getRamSize());
+ AlignedTVList restored =
+ AlignedTVList.deserialize(
+ new DataInputStream(new ByteArrayInputStream(buffer.getBuffer().array())));
+
+ Assert.assertEquals(list.getTsDataTypes(), restored.getTsDataTypes());
+ Assert.assertEquals(list.rowCount(), restored.rowCount());
+ for (int column = 1; column < list.getTsDataTypes().size(); column++) {
+ Assert.assertNull(list.getValues().get(column).get(0));
+ for (int row = 0; row <= ARRAY_SIZE; row++) {
+ Assert.assertTrue(list.isNullValue(row, column));
+ Assert.assertTrue(restored.isNullValue(row, column));
+ }
+ }
+ for (int row = 0; row < list.rowCount(); row++) {
+ Assert.assertEquals(list.getTime(row), restored.getTime(row));
+ Assert.assertEquals(
+ list.getAlignedValue(row).toString(), restored.getAlignedValue(row).toString());
+ }
+ list.clear();
+ restored.clear();
+ }
+
// A null-only column keeps a null array slot while a populated column materializes its array.
@Test
public void testPrimitiveArraysAreMaterializedOnlyForNonNullColumns() {