This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch allocate_primitive_array_lazily_pr in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 566781937d31e0baaaada5c616c86f3977a0975c Author: Tian Jiang <[email protected]> AuthorDate: Wed Jul 22 11:57:35 2026 +0800 Initialize primitive array lazily --- .../db/it/IoTDBAlignedTVListLazyAllocationIT.java | 142 +++++++++++++++++ .../dataregion/memtable/TsFileProcessor.java | 48 ++++-- .../db/utils/datastructure/AlignedTVList.java | 174 ++++++++++++--------- .../db/utils/datastructure/AlignedTVListTest.java | 90 +++++++++++ 4 files changed, 365 insertions(+), 89 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java new file mode 100644 index 00000000000..375f560c9ee --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBAlignedTVListLazyAllocationIT.java @@ -0,0 +1,142 @@ +/* + * 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.it; + +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; +import org.apache.iotdb.isession.ISession; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.it.env.EnvFactory; +import org.apache.iotdb.it.framework.IoTDBTestRunner; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.AfterClass; +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; + +@RunWith(IoTDBTestRunner.class) +@Category({LocalStandaloneIT.class}) +public class IoTDBAlignedTVListLazyAllocationIT { + + private static final String DEVICE = "root.aligned_lazy_allocation.d1"; + private static final int DATANODE_MAX_HEAP_SIZE_IN_MB = 256; + private static final int INITIAL_ROW_COUNT = 40_000; + private static final int NEW_COLUMN_COUNT = 1_200; + + @BeforeClass + public static void setUp() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getDataNodeJVMConfig() + .setMaxHeapSize(DATANODE_MAX_HEAP_SIZE_IN_MB); + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setAutoCreateSchemaEnabled(true) + .setEnableMemControl(true) + .setPrimitiveArraySize(64) + .setMemtableSizeThreshold(512L * 1024 * 1024) + .setDatanodeMemoryProportion("6:1:1:1:1:1") + .setWriteMemoryProportion("100:1"); + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testAddingManyColumnsAfterManyRowsDoesNotExhaustWriteMemory() throws Exception { + int historicalBlockCount = (INITIAL_ROW_COUNT + ARRAY_SIZE - 1) / ARRAY_SIZE; + long eagerAllocationCost = + (long) historicalBlockCount + * NEW_COLUMN_COUNT + * AlignedTVList.valueListArrayMemCost(TSDataType.INT64); + long lazyAllocationCost = + (long) historicalBlockCount + * NEW_COLUMN_COUNT + * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray() + + (long) NEW_COLUMN_COUNT * AlignedTVList.primitiveArrayMemCost(TSDataType.INT64); + long dataNodeMaxHeapSize = DATANODE_MAX_HEAP_SIZE_IN_MB * 1024L * 1024L; + + Assert.assertTrue( + "The eager implementation must exceed the entire DataNode heap in this scenario", + eagerAllocationCost > dataNodeMaxHeapSize); + Assert.assertTrue( + "The lazy implementation must fit comfortably within the DataNode heap", + lazyAllocationCost < dataNodeMaxHeapSize / 2); + + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + insertInitialRows(session); + + List<String> measurements = new ArrayList<>(NEW_COLUMN_COUNT); + List<TSDataType> dataTypes = new ArrayList<>(NEW_COLUMN_COUNT); + List<Object> values = new ArrayList<>(NEW_COLUMN_COUNT); + for (int i = 1; i <= NEW_COLUMN_COUNT; i++) { + measurements.add("s" + i); + dataTypes.add(TSDataType.INT64); + values.add((long) i); + } + session.insertAlignedRecord(DEVICE, INITIAL_ROW_COUNT, measurements, dataTypes, values); + + try (SessionDataSet dataSet = + session.executeQueryStatement( + "SELECT COUNT(s0), COUNT(s" + NEW_COLUMN_COUNT + ") FROM " + DEVICE)) { + Assert.assertTrue(dataSet.hasNext()); + List<org.apache.tsfile.read.common.Field> fields = dataSet.next().getFields(); + Assert.assertEquals(INITIAL_ROW_COUNT, fields.get(0).getLongV()); + Assert.assertEquals(1, fields.get(1).getLongV()); + Assert.assertFalse(dataSet.hasNext()); + } + } + } + + private static void insertInitialRows(ISession session) throws Exception { + List<IMeasurementSchema> schemas = + Collections.singletonList(new MeasurementSchema("s0", TSDataType.INT64)); + Tablet tablet = new Tablet(DEVICE, schemas); + for (int i = 0; i < INITIAL_ROW_COUNT; i++) { + int rowIndex = tablet.getRowSize(); + if (rowIndex == tablet.getMaxRowNumber()) { + session.insertAlignedTablet(tablet); + tablet.reset(); + rowIndex = 0; + } + tablet.addTimestamp(rowIndex, i); + tablet.addValue("s0", rowIndex, (long) i); + } + if (tablet.getRowSize() > 0) { + session.insertAlignedTablet(tablet); + } + } +} 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 4528b44f598..b2996d775ab 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 @@ -104,9 +104,11 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.CopyOnWriteArrayList; @@ -827,7 +829,6 @@ public class TsFileProcessor { } else { // For existed device of this mem table AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk; - List<TSDataType> dataTypesInTVList = new ArrayList<>(); for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { // Skip failed Measurements if (!isWritableFieldMeasurement(measurements, dataTypes, values, columnCategories, i)) { @@ -841,16 +842,14 @@ public class TsFileProcessor { + (alignedMemChunk.alignedListSize() % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); - memTableIncrement += currentArrayNum * AlignedTVList.valueListArrayMemCost(dataTypes[i]); - dataTypesInTVList.add(dataTypes[i]); + memTableIncrement += + currentArrayNum * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray() + + AlignedTVList.primitiveArrayMemCost(dataTypes[i]); } } // this insertion will result in a new array if ((alignedMemChunk.alignedListSize() % PrimitiveArrayManager.ARRAY_SIZE) == 0) { memTableIncrement += alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost(); - for (TSDataType dataType : dataTypesInTVList) { - memTableIncrement += AlignedTVList.valueListArrayMemCost(dataType); - } } } @@ -908,7 +907,7 @@ public class TsFileProcessor { // For existed device of this mem table AlignedWritableMemChunk alignedMemChunk = (AlignedWritableMemChunk) memChunk; int currentChunkPointNum = alignedMemChunk == null ? 0 : alignedMemChunk.alignedListSize(); - List<TSDataType> dataTypesInTVList = new ArrayList<>(); + Set<String> measurementsAddedInCurrentRow = new HashSet<>(); Pair<Map<String, TSDataType>, Integer> addingPointNumInfo = increasingMemTableInfo.computeIfAbsent(deviceId, k -> new Pair<>(new HashMap<>(), 0)); for (int i = 0; dataTypes != null && i < dataTypes.length; i++) { @@ -925,6 +924,7 @@ public class TsFileProcessor { if (!currentMemChunkContainsMeasurement && !addingPointNumInfo.left.containsKey(measurements[i])) { addingPointNumInfo.left.put(measurements[i], dataTypes[i]); + measurementsAddedInCurrentRow.add(measurements[i]); int currentArrayNum = (currentChunkPointNum + addingPointNum) / PrimitiveArrayManager.ARRAY_SIZE + ((currentChunkPointNum + addingPointNum) % PrimitiveArrayManager.ARRAY_SIZE @@ -932,21 +932,28 @@ public class TsFileProcessor { ? 1 : 0); memTableIncrement += - currentArrayNum * AlignedTVList.valueListArrayMemCost(dataTypes[i]); + currentArrayNum * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray() + + AlignedTVList.primitiveArrayMemCost(dataTypes[i]); } } int addingPointNum = addingPointNumInfo.right; // Here currentChunkPointNum + addingPointNum >= 1 if (((currentChunkPointNum + addingPointNum) % PrimitiveArrayManager.ARRAY_SIZE) == 0) { - dataTypesInTVList.addAll(addingPointNumInfo.left.values()); + List<TSDataType> existingDataTypesInTVList = new ArrayList<>(); + addingPointNumInfo.left.forEach( + (measurement, dataType) -> { + if (!measurementsAddedInCurrentRow.contains(measurement)) { + existingDataTypesInTVList.add(dataType); + } + }); memTableIncrement += alignedMemChunk != null ? alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost() - + dataTypesInTVList.stream() + + existingDataTypesInTVList.stream() .mapToLong(AlignedTVList::valueListArrayMemCost) .sum() : AlignedTVList.alignedTvListArrayMemCost( - dataTypesInTVList.toArray(new TSDataType[0]), null); + existingDataTypesInTVList.toArray(new TSDataType[0]), null); } addingPointNumInfo.setRight(addingPointNum + 1); } @@ -1112,8 +1119,9 @@ public class TsFileProcessor { } if (!alignedMemChunk.containsMeasurement(measurementIds[i])) { - // add a new column in the TVList, the new column should be as long as existing ones - memIncrements[0] += currentArrayCnt * AlignedTVList.valueListArrayMemCost(dataType); + // Historical blocks only add null placeholders and bitmaps for a new column. + memIncrements[0] += + currentArrayCnt * AlignedTVList.valueListArrayMemCostWithoutPrimitiveArray(); dataTypesInTVList.add(dataType); } } @@ -1124,13 +1132,21 @@ public class TsFileProcessor { + (newPointNum % PrimitiveArrayManager.ARRAY_SIZE > 0 ? 1 : 0); long acquireArray = newArrayCnt - currentArrayCnt; + boolean writesExistingLastBlock = + currentPointNum % PrimitiveArrayManager.ARRAY_SIZE != 0 && incomingPointNum > 0; + for (TSDataType dataType : dataTypesInTVList) { + if (writesExistingLastBlock) { + memIncrements[0] += AlignedTVList.primitiveArrayMemCost(dataType); + } + // Reserve a bitmap as well as a value array for new blocks because the tablet may contain + // null or failed rows in those blocks. + memIncrements[0] += acquireArray * AlignedTVList.valueListArrayMemCost(dataType); + } + if (acquireArray != 0) { // memory of extending the TVList memIncrements[0] += acquireArray * alignedMemChunk.getWorkingTVList().alignedTvListArrayMemCost(); - for (TSDataType dataType : dataTypesInTVList) { - memIncrements[0] += acquireArray * AlignedTVList.valueListArrayMemCost(dataType); - } } } 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 464cbe09995..83ea1d1217d 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 @@ -125,7 +125,9 @@ public abstract class AlignedTVList extends TVList { if (!to.isCompatible(from)) { return null; } - return originalValues.stream().map(o -> to.castFromArray(from, o)).collect(Collectors.toList()); + return originalValues.stream() + .map(o -> o == null ? null : to.castFromArray(from, o)) + .collect(Collectors.toList()); } @Override @@ -229,43 +231,35 @@ public abstract class AlignedTVList extends TVList { timestamps.get(arrayIndex)[elementIndex] = timestamp; for (int i = 0; i < values.size(); i++) { Object columnValue = value[i]; - List<Object> columnValues = values.get(i); if (columnValue == null) { markNullValue(i, arrayIndex, elementIndex); + continue; } + Object valueArray = getOrCreateValueArray(i, arrayIndex); switch (dataTypes.get(i)) { case TEXT: case BLOB: case STRING: case OBJECT: - ((Binary[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (Binary) columnValue : Binary.EMPTY_VALUE; - memoryBinaryChunkSize[i] += - columnValue != null - ? getBinarySize((Binary) columnValue) - : getBinarySize(Binary.EMPTY_VALUE); + ((Binary[]) valueArray)[elementIndex] = (Binary) columnValue; + memoryBinaryChunkSize[i] += getBinarySize((Binary) columnValue); break; case FLOAT: - ((float[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (float) columnValue : Float.MIN_VALUE; + ((float[]) valueArray)[elementIndex] = (float) columnValue; break; case INT32: case DATE: - ((int[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (int) columnValue : Integer.MIN_VALUE; + ((int[]) valueArray)[elementIndex] = (int) columnValue; break; case INT64: case TIMESTAMP: - ((long[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (long) columnValue : Long.MIN_VALUE; + ((long[]) valueArray)[elementIndex] = (long) columnValue; break; case DOUBLE: - ((double[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null ? (double) columnValue : Double.MIN_VALUE; + ((double[]) valueArray)[elementIndex] = (double) columnValue; break; case BOOLEAN: - ((boolean[]) columnValues.get(arrayIndex))[elementIndex] = - columnValue != null && (boolean) columnValue; + ((boolean[]) valueArray)[elementIndex] = (boolean) columnValue; break; default: break; @@ -396,33 +390,7 @@ public abstract class AlignedTVList extends TVList { List<Object> columnValue = new ArrayList<>(timestamps.size()); List<BitMap> columnBitMaps = new ArrayList<>(timestamps.size()); for (int i = 0; i < timestamps.size(); i++) { - switch (dataType) { - case TEXT: - case STRING: - case BLOB: - case OBJECT: - columnValue.add(getPrimitiveArraysByType(TSDataType.TEXT)); - break; - case FLOAT: - columnValue.add(getPrimitiveArraysByType(TSDataType.FLOAT)); - break; - case INT32: - case DATE: - columnValue.add(getPrimitiveArraysByType(TSDataType.INT32)); - break; - case INT64: - case TIMESTAMP: - columnValue.add(getPrimitiveArraysByType(TSDataType.INT64)); - break; - case DOUBLE: - columnValue.add(getPrimitiveArraysByType(TSDataType.DOUBLE)); - break; - case BOOLEAN: - columnValue.add(getPrimitiveArraysByType(TSDataType.BOOLEAN)); - break; - default: - break; - } + columnValue.add(null); BitMap bitMap = BitMap.createBitMapDynamically(ARRAY_SIZE); // The following code is for these 2 kinds of scenarios. @@ -621,12 +589,15 @@ public abstract class AlignedTVList extends TVList { if (columnIndex < 0 || columnIndex >= values.size() || values.get(columnIndex) == null) { return true; } + int arrayIndex = unsortedRowIndex / ARRAY_SIZE; + if (values.get(columnIndex).get(arrayIndex) == null) { + return true; + } if (bitMaps == null || bitMaps.get(columnIndex) == null - || bitMaps.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); @@ -744,6 +715,9 @@ public abstract class AlignedTVList extends TVList { } protected Object cloneValue(TSDataType type, Object value) { + if (value == null) { + return null; + } switch (type) { case TEXT: case BLOB: @@ -791,7 +765,9 @@ public abstract class AlignedTVList extends TVList { List<Object> columnValues = values.get(i); if (columnValues != null) { for (Object dataArray : columnValues) { - PrimitiveArrayManager.release(dataArray); + if (dataArray != null) { + PrimitiveArrayManager.release(dataArray); + } } columnValues.clear(); } @@ -817,7 +793,7 @@ public abstract class AlignedTVList extends TVList { indices.add((int[]) getPrimitiveArraysByType(TSDataType.INT32)); } for (int i = 0; i < dataTypes.size(); i++) { - values.get(i).add(getPrimitiveArraysByType(dataTypes.get(i))); + values.get(i).add(null); if (bitMaps != null && bitMaps.get(i) != null) { bitMaps.get(i).add(null); } @@ -887,7 +863,7 @@ public abstract class AlignedTVList extends TVList { if (internalRemaining >= inputRemaining) { // the remaining inputs can fit the last array, copy all remaining inputs into last array System.arraycopy(time, idx, timestamps.get(arrayIdx), elementIdx, inputRemaining); - arrayCopy(value, idx, arrayIdx, elementIdx, inputRemaining); + arrayCopy(value, bitMaps, results, idx, arrayIdx, elementIdx, inputRemaining); for (int i = 0; i < inputRemaining; i++) { if (indices != null) { indices.get(arrayIdx)[elementIdx + i] = rowCount; @@ -900,7 +876,7 @@ public abstract class AlignedTVList extends TVList { // the remaining inputs cannot fit the last array, fill the last array and create a new // one and enter the next loop System.arraycopy(time, idx, timestamps.get(arrayIdx), elementIdx, internalRemaining); - arrayCopy(value, idx, arrayIdx, elementIdx, internalRemaining); + arrayCopy(value, bitMaps, results, idx, arrayIdx, elementIdx, internalRemaining); for (int i = 0; i < internalRemaining; i++) { if (indices != null) { indices.get(arrayIdx)[elementIdx + i] = rowCount; @@ -1007,18 +983,25 @@ public abstract class AlignedTVList extends TVList { return bitmap; } - private void arrayCopy(Object[] value, int idx, int arrayIndex, int elementIndex, int remaining) { + private void arrayCopy( + Object[] value, + BitMap[] bitMaps, + TSStatus[] results, + int idx, + int arrayIndex, + int elementIndex, + int remaining) { for (int i = 0; i < values.size(); i++) { - if (value[i] == null) { + if (value[i] == null || !containsNonNullValue(bitMaps, results, i, idx, remaining)) { continue; } - List<Object> columnValues = values.get(i); + Object valueArray = getOrCreateValueArray(i, arrayIndex); switch (dataTypes.get(i)) { case TEXT: case BLOB: case STRING: case OBJECT: - Binary[] arrayT = ((Binary[]) columnValues.get(arrayIndex)); + Binary[] arrayT = (Binary[]) valueArray; System.arraycopy(value[i], idx, arrayT, elementIndex, remaining); // update raw size of Text chunk @@ -1028,25 +1011,25 @@ public abstract class AlignedTVList extends TVList { } break; case FLOAT: - float[] arrayF = ((float[]) columnValues.get(arrayIndex)); + float[] arrayF = (float[]) valueArray; System.arraycopy(value[i], idx, arrayF, elementIndex, remaining); break; case INT32: case DATE: - int[] arrayI = ((int[]) columnValues.get(arrayIndex)); + int[] arrayI = (int[]) valueArray; System.arraycopy(value[i], idx, arrayI, elementIndex, remaining); break; case INT64: case TIMESTAMP: - long[] arrayL = ((long[]) columnValues.get(arrayIndex)); + long[] arrayL = (long[]) valueArray; System.arraycopy(value[i], idx, arrayL, elementIndex, remaining); break; case DOUBLE: - double[] arrayD = ((double[]) columnValues.get(arrayIndex)); + double[] arrayD = (double[]) valueArray; System.arraycopy(value[i], idx, arrayD, elementIndex, remaining); break; case BOOLEAN: - boolean[] arrayB = ((boolean[]) columnValues.get(arrayIndex)); + boolean[] arrayB = (boolean[]) valueArray; System.arraycopy(value[i], idx, arrayB, elementIndex, remaining); break; default: @@ -1055,6 +1038,37 @@ public abstract class AlignedTVList extends TVList { } } + private static boolean containsNonNullValue( + BitMap[] bitMaps, TSStatus[] results, int columnIndex, int start, int length) { + BitMap bitMap = bitMaps == null ? null : bitMaps[columnIndex]; + boolean containsNull = bitMap != null && containsMarkedBit(bitMap, start, length); + boolean containsFailure = results != null && containsFailedStatus(results, start, length); + if (!containsNull && !containsFailure) { + return true; + } + for (int i = start; i < start + length; i++) { + boolean isNull = + (bitMap != null && bitMap.isMarked(i)) + || (results != null + && results[i] != null + && results[i].code != TSStatusCode.SUCCESS_STATUS.getStatusCode()); + if (!isNull) { + return true; + } + } + return false; + } + + private Object getOrCreateValueArray(int columnIndex, int arrayIndex) { + List<Object> columnValues = values.get(columnIndex); + Object valueArray = columnValues.get(arrayIndex); + if (valueArray == null) { + valueArray = getPrimitiveArraysByType(dataTypes.get(columnIndex)); + columnValues.set(arrayIndex, valueArray); + } + return valueArray; + } + private BitMap getBitMap(int columnIndex, int arrayIndex) { // init BitMaps if doesn't have if (bitMaps == null) { @@ -1175,14 +1189,20 @@ public abstract class AlignedTVList extends TVList { * @return valueListArrayMemCost */ public static long valueListArrayMemCost(TSDataType type) { + return primitiveArrayMemCost(type) + valueListArrayMemCostWithoutPrimitiveArray(); + } + + public static long primitiveArrayMemCost(TSDataType type) { + // value array payload and header + return (long) PrimitiveArrayManager.ARRAY_SIZE * (long) type.getDataTypeSize() + + NUM_BYTES_ARRAY_HEADER; + } + + public static long valueListArrayMemCostWithoutPrimitiveArray() { long size = 0; - // value array mem size - size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long) type.getDataTypeSize(); // bitmap object, byte array, and reference in the bitmap list size += BITMAP_RAM_COST_PER_BLOCK; - // array headers mem size - size += NUM_BYTES_ARRAY_HEADER; - // Object references size in ArrayList + // null placeholder reference in the value ArrayList size += NUM_BYTES_OBJECT_REF; return size; } @@ -1404,7 +1424,13 @@ public abstract class AlignedTVList extends TVList { case STRING: case OBJECT: for (int rowIdx = 0; rowIdx < rowCount; ++rowIdx) { - size += ReadWriteIOUtils.sizeToWrite(getBinaryByValueIndex(rowIdx, columnIndex)); + int arrayIndex = rowIdx / ARRAY_SIZE; + Object valueArray = values.get(columnIndex).get(arrayIndex); + Binary value = + valueArray == null + ? Binary.EMPTY_VALUE + : ((Binary[]) valueArray)[rowIdx % ARRAY_SIZE]; + size += ReadWriteIOUtils.sizeToWrite(value == null ? Binary.EMPTY_VALUE : value); } break; case FLOAT: @@ -1458,13 +1484,15 @@ public abstract class AlignedTVList extends TVList { for (int rowIndex = 0; rowIndex < rowCount; ++rowIndex) { int arrayIndex = rowIndex / ARRAY_SIZE; int elementIndex = rowIndex % ARRAY_SIZE; + Object valueArray = columnValues.get(arrayIndex); // value switch (dataTypes.get(columnIndex)) { case TEXT: case BLOB: case STRING: case OBJECT: - Binary valueT = ((Binary[]) columnValues.get(arrayIndex))[elementIndex]; + Binary valueT = + valueArray == null ? Binary.EMPTY_VALUE : ((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 @@ -1474,29 +1502,29 @@ public abstract class AlignedTVList extends TVList { if (valueT != null) { WALWriteUtils.write(valueT, buffer); } else { - WALWriteUtils.write(new Binary(new byte[0]), buffer); + WALWriteUtils.write(Binary.EMPTY_VALUE, buffer); } 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/utils/datastructure/AlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedTVListTest.java index c6ccd333942..5cc4e2a288b 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 @@ -19,6 +19,7 @@ package org.apache.iotdb.db.utils.datastructure; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; import org.apache.iotdb.rpc.TSStatusCode; import org.apache.tsfile.common.conf.TSFileConfig; @@ -29,6 +30,10 @@ import org.apache.tsfile.utils.BitMap; 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.List; @@ -224,6 +229,91 @@ public class AlignedTVListTest { Assert.assertNull(tvList.getBitMaps()); } + @Test + public void testPrimitiveArraysAreAllocatedOnFirstWrite() { + AlignedTVList tvList = + AlignedTVList.newAlignedList( + new ArrayList<>(Arrays.asList(TSDataType.INT64, TSDataType.INT64))); + for (int i = 0; i <= ARRAY_SIZE; i++) { + tvList.putAlignedValue(i, new Object[] {(long) i, null}); + } + + Assert.assertNotNull(tvList.getValues().get(0).get(0)); + Assert.assertNotNull(tvList.getValues().get(0).get(1)); + Assert.assertNull(tvList.getValues().get(1).get(0)); + Assert.assertNull(tvList.getValues().get(1).get(1)); + + tvList.putAlignedValue(ARRAY_SIZE + 1L, new Object[] {null, 1L}); + + Assert.assertNull(tvList.getValues().get(1).get(0)); + Assert.assertNotNull(tvList.getValues().get(1).get(1)); + Assert.assertTrue(tvList.isNullValue(0, 1)); + Assert.assertEquals(1, tvList.getLongByValueIndex(ARRAY_SIZE + 1, 1)); + + tvList.extendColumn(TSDataType.INT32); + + Assert.assertNull(tvList.getValues().get(2).get(0)); + Assert.assertNull(tvList.getValues().get(2).get(1)); + + tvList.putAlignedValue(ARRAY_SIZE + 2L, new Object[] {null, null, 2}); + + Assert.assertNull(tvList.getValues().get(2).get(0)); + Assert.assertNotNull(tvList.getValues().get(2).get(1)); + Assert.assertTrue(tvList.isNullValue(0, 2)); + Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE + 2, 2)); + Assert.assertEquals(2, tvList.getIntByValueIndex(ARRAY_SIZE + 2, 2)); + } + + @Test + public void testBatchDoesNotAllocateAllNullPrimitiveArray() { + AlignedTVList tvList = + AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64, TSDataType.INT64)); + long[] times = new long[ARRAY_SIZE]; + long[][] values = new long[2][ARRAY_SIZE]; + BitMap[] bitMaps = new BitMap[] {null, new BitMap(ARRAY_SIZE)}; + bitMaps[1].markAll(); + for (int i = 0; i < ARRAY_SIZE; i++) { + times[i] = i; + values[0][i] = i; + values[1][i] = i; + } + + tvList.putAlignedValues(times, values, bitMaps, 0, ARRAY_SIZE, null); + + Assert.assertNotNull(tvList.getValues().get(0).get(0)); + Assert.assertNull(tvList.getValues().get(1).get(0)); + Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE - 1, 1)); + } + + @Test + public void testNullPrimitiveArrayCanBeClonedAndSerialized() throws IOException { + AlignedTVList tvList = + AlignedTVList.newAlignedList(Arrays.asList(TSDataType.TEXT, TSDataType.INT64)); + for (int i = 0; i <= ARRAY_SIZE; i++) { + tvList.putAlignedValue(i, new Object[] {null, null}); + } + tvList.putAlignedValue( + ARRAY_SIZE + 1L, new Object[] {new Binary("value", TSFileConfig.STRING_CHARSET), 1L}); + + AlignedTVList clonedTvList = tvList.clone(); + Assert.assertNull(clonedTvList.getValues().get(0).get(0)); + Assert.assertNull(clonedTvList.getValues().get(1).get(0)); + Assert.assertEquals("[null, null]", clonedTvList.getAlignedValue(0).toString()); + Assert.assertEquals("[value, 1]", clonedTvList.getAlignedValue(ARRAY_SIZE + 1).toString()); + + WALByteBufferForTest walBuffer = + new WALByteBufferForTest(ByteBuffer.allocate(tvList.serializedSize())); + tvList.serializeToWAL(walBuffer); + AlignedTVList deserializedTvList = + AlignedTVList.deserialize( + new DataInputStream(new ByteArrayInputStream(walBuffer.getBuffer().array()))); + + Assert.assertEquals(tvList.rowCount(), deserializedTvList.rowCount()); + Assert.assertEquals("[null, null]", deserializedTvList.getAlignedValue(0).toString()); + Assert.assertEquals( + "[value, 1]", deserializedTvList.getAlignedValue(ARRAY_SIZE + 1).toString()); + } + @Test public void testClone() { List<TSDataType> dataTypes = new ArrayList<>();
