From ef44ea50cdef7baa35c83b600cc798c1a363a0a8 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 30 Jul 2026 11:06:09 +0800 Subject: [PATCH 1/3] [Performance] Avoid bitmap copies for aligned tablet prefixes --- .../db/utils/datastructure/AlignedTVList.java | 8 +- ...lignedBitmapRangeCheckPerformanceTest.java | 201 ++++++++++++++++++ .../datastructure/AlignedTVListTest.java | 36 ++++ 3 files changed, 242 insertions(+), 3 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedBitmapRangeCheckPerformanceTest.java 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 a589e1ea97a6..0c60af7cd369 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 @@ -970,9 +970,11 @@ private static boolean containsFailedStatus(TSStatus[] results, int start, int l return false; } - private static boolean containsMarkedBit(BitMap bitMap, int start, int length) { - if (length <= 0) { - return false; + static boolean containsMarkedBit(BitMap bitMap, int start, int length) { + // Avoid materializing a byte-array copy on the common aligned-tablet path, which starts at + // offset 0 and can be inspected directly by BitMap. + if (start == 0) { + return length > 0 && !bitMap.isAllUnmarked(length); } byte[] bytes = bitMap.getByteArray(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedBitmapRangeCheckPerformanceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedBitmapRangeCheckPerformanceTest.java new file mode 100644 index 000000000000..9247c8672025 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedBitmapRangeCheckPerformanceTest.java @@ -0,0 +1,201 @@ +/* + * 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.utils.datastructure; + +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils; +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement; +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary; + +import org.apache.tsfile.utils.BitMap; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.Test; + +import java.util.Locale; + +public class AlignedBitmapRangeCheckPerformanceTest { + + private static final String ENABLED_PROPERTY = "iotdb.aligned.bitmap.range-check.perf.enabled"; + private static final String ITERATIONS_PROPERTY = + "iotdb.aligned.bitmap.range-check.perf.iterations"; + private static final String ROUNDS_PROPERTY = "iotdb.aligned.bitmap.range-check.perf.rounds"; + private static final int REPETITIONS = 2048; + private static final int BITMAP_COUNT = 64; + private static final int BITMAP_MASK = BITMAP_COUNT - 1; + + private static volatile long benchmarkBlackhole; + + @Test + public void bitmapRangeCheckBenchmark() { + Assume.assumeTrue( + String.format( + Locale.ROOT, + "Manual performance UT. Enable with -D%s=true; optionally tune -D%s and -D%s.", + ENABLED_PROPERTY, + ITERATIONS_PROPERTY, + ROUNDS_PROPERTY), + Boolean.getBoolean(ENABLED_PROPERTY)); + Assume.assumeTrue( + "Current-thread CPU time and allocation metrics are required.", + ManualPerformanceTestUtils.enableThreadMetrics()); + + int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 4000); + int rounds = Integer.getInteger(ROUNDS_PROPERTY, 5); + Assert.assertTrue(iterations > 0); + Assert.assertTrue(rounds > 0); + + runScenario("prefix", createBitMaps(), 0, Long.SIZE, iterations, rounds); + runScenario("partial", createBitMaps(), 1, Long.SIZE - 1, iterations, rounds); + } + + private static void runScenario( + String label, BitMap[] bitMaps, int start, int length, int iterations, int rounds) { + int operations = iterations * REPETITIONS; + runLegacy(bitMaps, start, length, REPETITIONS); + runOptimized(bitMaps, start, length, REPETITIONS); + + Measurement[] legacyMeasurements = new Measurement[rounds]; + Measurement[] optimizedMeasurements = new Measurement[rounds]; + for (int i = 0; i < rounds; i++) { + if ((i & 1) == 0) { + legacyMeasurements[i] = measureLegacy(bitMaps, start, length, operations); + optimizedMeasurements[i] = measureOptimized(bitMaps, start, length, operations); + } else { + optimizedMeasurements[i] = measureOptimized(bitMaps, start, length, operations); + legacyMeasurements[i] = measureLegacy(bitMaps, start, length, operations); + } + } + + Summary legacySummary = ManualPerformanceTestUtils.summarize(legacyMeasurements, operations); + Summary optimizedSummary = + ManualPerformanceTestUtils.summarize(optimizedMeasurements, operations); + printResult(label, start, length, operations, rounds, legacySummary, optimizedSummary); + } + + private static Measurement measureLegacy( + BitMap[] bitMaps, int start, int length, int operations) { + return ManualPerformanceTestUtils.measure( + 1, () -> runLegacy(bitMaps, start, length, operations)); + } + + private static Measurement measureOptimized( + BitMap[] bitMaps, int start, int length, int operations) { + return ManualPerformanceTestUtils.measure( + 1, () -> runOptimized(bitMaps, start, length, operations)); + } + + private static void runLegacy(BitMap[] bitMaps, int start, int length, int operations) { + long markedCount = 0; + for (int i = 0; i < operations; i++) { + if (legacyContainsMarkedBit(bitMaps[i & BITMAP_MASK], start, length)) { + markedCount++; + } + } + benchmarkBlackhole = markedCount; + } + + private static void runOptimized(BitMap[] bitMaps, int start, int length, int operations) { + long markedCount = 0; + for (int i = 0; i < operations; i++) { + if (AlignedTVList.containsMarkedBit(bitMaps[i & BITMAP_MASK], start, length)) { + markedCount++; + } + } + benchmarkBlackhole = markedCount; + } + + private static boolean legacyContainsMarkedBit(BitMap bitMap, int start, int length) { + byte[] bytes = bitMap.getByteArray(); + int end = start + length - 1; + int firstByteIndex = start >>> 3; + int lastByteIndex = end >>> 3; + if (firstByteIndex == lastByteIndex) { + int mask = (0xFF << (start & 7)) & (0xFF >>> (7 - (end & 7))); + return (bytes[firstByteIndex] & mask) != 0; + } + if ((bytes[firstByteIndex] & (0xFF << (start & 7))) != 0) { + return true; + } + for (int i = firstByteIndex + 1; i < lastByteIndex; i++) { + if (bytes[i] != 0) { + return true; + } + } + return (bytes[lastByteIndex] & (0xFF >>> (7 - (end & 7)))) != 0; + } + + private static BitMap[] createBitMaps() { + BitMap[] bitMaps = new BitMap[BITMAP_COUNT]; + for (int i = 0; i < BITMAP_COUNT; i++) { + bitMaps[i] = BitMap.createBitMapDynamically(Long.SIZE); + if ((i & 1) != 0) { + bitMaps[i].mark(Long.SIZE - 1); + } + } + return bitMaps; + } + + private static void printResult( + String label, + int start, + int length, + int operations, + int rounds, + Summary legacySummary, + Summary optimizedSummary) { + System.out.printf( + Locale.ROOT, + "Aligned bitmap range-check benchmark (%s): start=%d, length=%d, operations/round=%d, rounds=%d%n", + label, + start, + length, + operations, + rounds); + printSummary("legacy", legacySummary); + printSummary("optimized", optimizedSummary); + System.out.printf( + Locale.ROOT, + " optimized/legacy CPU ratio=%.2f%%, allocation ratio=%.2f%%%n", + percentage( + optimizedSummary.getCpuNanosPerOperation(), legacySummary.getCpuNanosPerOperation()), + percentage( + optimizedSummary.getAllocatedBytesPerOperation(), + legacySummary.getAllocatedBytesPerOperation())); + System.out.printf( + Locale.ROOT, + " optimized-legacy CPU delta=%+.3f ns/check, allocation delta=%+.1f bytes/check%n", + optimizedSummary.getCpuNanosPerOperation() - legacySummary.getCpuNanosPerOperation(), + optimizedSummary.getAllocatedBytesPerOperation() + - legacySummary.getAllocatedBytesPerOperation()); + } + + private static void printSummary(String label, Summary summary) { + System.out.printf( + Locale.ROOT, + " %-10s CPU=%.3f ns/check, allocated=%.1f bytes/check%n", + label, + summary.getCpuNanosPerOperation(), + summary.getAllocatedBytesPerOperation()); + } + + private static double percentage(double numerator, double denominator) { + return denominator == 0 ? 0 : numerator * 100.0 / denominator; + } +} 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 381f733bf4dd..afc984c4f8c1 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 @@ -244,6 +244,42 @@ public void testEmptyInputBitmapsDoNotMaterializeMemTableBitmaps() { Assert.assertNull(tvList.getBitMaps()); } + @Test + public void testContainsMarkedBitForLongAndArrayBackedBitmaps() { + assertContainsMarkedBitRanges(new BitMap(Long.SIZE)); + assertContainsMarkedBitRanges(BitMap.createBitMapDynamically(Long.SIZE)); + + BitMap largeBitMap = BitMap.createBitMapDynamically(Long.SIZE * 2 + 2); + Assert.assertFalse(AlignedTVList.containsMarkedBit(largeBitMap, 1, Long.SIZE)); + largeBitMap.mark(Long.SIZE); + Assert.assertTrue(AlignedTVList.containsMarkedBit(largeBitMap, 1, Long.SIZE)); + Assert.assertFalse(AlignedTVList.containsMarkedBit(largeBitMap, Long.SIZE + 1, Long.SIZE)); + largeBitMap.mark(Long.SIZE * 2 + 1); + Assert.assertTrue(AlignedTVList.containsMarkedBit(largeBitMap, Long.SIZE + 1, Long.SIZE + 1)); + } + + private static void assertContainsMarkedBitRanges(BitMap bitMap) { + Assert.assertFalse(AlignedTVList.containsMarkedBit(bitMap, 0, 0)); + Assert.assertFalse(AlignedTVList.containsMarkedBit(bitMap, 0, Long.SIZE)); + Assert.assertFalse(AlignedTVList.containsMarkedBit(bitMap, 1, Long.SIZE - 1)); + + bitMap.mark(0); + Assert.assertTrue(AlignedTVList.containsMarkedBit(bitMap, 0, 1)); + Assert.assertFalse(AlignedTVList.containsMarkedBit(bitMap, 1, Long.SIZE - 1)); + + bitMap.reset(); + bitMap.mark(Long.SIZE / 2); + Assert.assertFalse(AlignedTVList.containsMarkedBit(bitMap, 0, Long.SIZE / 2)); + Assert.assertTrue(AlignedTVList.containsMarkedBit(bitMap, 0, Long.SIZE / 2 + 1)); + Assert.assertTrue(AlignedTVList.containsMarkedBit(bitMap, Long.SIZE / 2 - 1, 3)); + Assert.assertFalse(AlignedTVList.containsMarkedBit(bitMap, Long.SIZE / 2 + 1, 1)); + + bitMap.reset(); + bitMap.mark(Long.SIZE - 1); + Assert.assertFalse(AlignedTVList.containsMarkedBit(bitMap, 1, Long.SIZE - 2)); + Assert.assertTrue(AlignedTVList.containsMarkedBit(bitMap, 1, Long.SIZE - 1)); + } + @Test public void testPrimitiveArraysAreAllocatedOnFirstWrite() { AlignedTVList tvList = From cde4ef77eb6768cc8b824627567891c1d291ef45 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 30 Jul 2026 15:17:36 +0800 Subject: [PATCH 2/3] [Performance] Defer bitmap allocation for sparse aligned writes --- .../db/utils/datastructure/AlignedTVList.java | 179 ++++------ .../memtable/TsFileProcessorTest.java | 29 +- .../AlignedSparseWritePerformanceTest.java | 314 ++++++++++++++++++ .../datastructure/AlignedTVListTest.java | 109 +++++- 4 files changed, 494 insertions(+), 137 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedSparseWritePerformanceTest.java 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 0c60af7cd369..1d85426b6821 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 @@ -87,7 +87,8 @@ public abstract class AlignedTVList extends TVList { private long materializedBitmapMemoryCost; private long arrayMemCostWithoutPrimitiveArraysAndIndex; - // Data type list -> list of TVList, add 1 when expanded -> primitive array of basic type + // Data type list -> list of TVList, add 1 when expanded -> primitive array of basic type. + // A null primitive array means all existing rows in that block are null for the column. // Index relation: columnIndex(dataTypeIndex) -> arrayIndex -> elementIndex protected List> values; @@ -261,7 +262,7 @@ public synchronized void putAlignedValue(long timestamp, Object[] value) { markNullValue(i, arrayIndex, elementIndex); continue; } - Object valueArray = getOrCreateValueArray(i, arrayIndex); + Object valueArray = getOrCreateValueArray(i, arrayIndex, elementIndex); switch (dataTypes.get(i)) { case TEXT: case BLOB: @@ -406,37 +407,13 @@ private TsPrimitiveType getAlignedValueByValueIndex( } public void extendColumn(TSDataType dataType) { - if (bitMaps == null) { - List> localBitMaps = new ArrayList<>(values.size()); - for (int i = 0; i < values.size(); i++) { - localBitMaps.add(null); - } - bitMaps = localBitMaps; - } List columnValue = new ArrayList<>(timestamps.size()); - List columnBitMaps = new ArrayList<>(timestamps.size()); for (int i = 0; i < timestamps.size(); i++) { columnValue.add(null); - BitMap bitMap = BitMap.createBitMapDynamically(ARRAY_SIZE); - // The following code is for these 2 kinds of scenarios. - - // Eg1: If rowCount=5 and ARRAY_SIZE=2, we need to supply 3 bitmaps for the extending column. - // The first 2 bitmaps should mark all bits to represent 4 nulls and the 3rd bitmap should - // mark - // the 1st bit to represent 1 null value. - - // Eg2: If rowCount=4 and ARRAY_SIZE=2, we need to supply 2 bitmaps for the extending column. - // These 2 bitmaps should mark all bits to represent 4 nulls. - if (i == timestamps.size() - 1 && rowCount % ARRAY_SIZE != 0) { - bitMap.markRange(0, rowCount % ARRAY_SIZE); - } else { - bitMap.markAll(); - } - columnBitMaps.add(bitMap); } - materializedBitmapMemoryCost += - (long) timestamps.size() * (bitmapReferenceRamCost() + bitmapRamCost()); - this.bitMaps.add(columnBitMaps); + if (bitMaps != null) { + bitMaps.add(null); + } this.values.add(columnValue); this.dataTypes.add(dataType); refreshArrayMemCostWithoutPrimitiveArrays(); @@ -722,28 +699,14 @@ public Pair delete(long lowerBound, long upperBound, int colum } public void deleteColumn(int columnIndex) { - if (bitMaps == null) { - List> localBitMaps = new ArrayList<>(dataTypes.size()); - for (int j = 0; j < dataTypes.size(); j++) { - localBitMaps.add(null); - } - bitMaps = localBitMaps; - } - if (bitMaps.get(columnIndex) == null) { - List columnBitMaps = new ArrayList<>(values.get(columnIndex).size()); - for (int i = 0; i < values.get(columnIndex).size(); i++) { - columnBitMaps.add(BitMap.createBitMapDynamically(ARRAY_SIZE)); - } - bitMaps.set(columnIndex, columnBitMaps); - materializedBitmapMemoryCost += - (long) columnBitMaps.size() * (bitmapReferenceRamCost() + bitmapRamCost()); + List columnValues = values.get(columnIndex); + if (columnValues == null) { + return; } - for (int i = 0; i < bitMaps.get(columnIndex).size(); i++) { - if (bitMaps.get(columnIndex).get(i) == null) { - bitMaps.get(columnIndex).set(i, BitMap.createBitMapDynamically(ARRAY_SIZE)); - materializedBitmapMemoryCost += bitmapRamCost(); + for (int arrayIndex = 0; arrayIndex < columnValues.size(); arrayIndex++) { + if (columnValues.get(arrayIndex) != null) { + getBitMap(columnIndex, arrayIndex).markAll(); } - bitMaps.get(columnIndex).get(i).markAll(); } } @@ -928,7 +891,7 @@ public synchronized void putAlignedValues( } private void markNullBitmapRange( - Object[] values, + Object[] inputValues, BitMap[] bitMaps, TSStatus[] results, int idx, @@ -942,9 +905,15 @@ private void markNullBitmapRange( ? buildResultBitMap(results, idx, elementIdx, len) : null; - for (int j = 0; j < values.length; j++) { + for (int j = 0; j < inputValues.length; j++) { + // A null value array represents an entirely null block, so no bitmap is needed until the + // block receives its first non-null value. + if (values.get(j).get(arrayIndex) == null) { + continue; + } + /* Fast-path: column is entirely null */ - if (values[j] == null) { + if (inputValues[j] == null) { getBitMap(j, arrayIndex).markRange(elementIdx, len); continue; } @@ -1034,7 +1003,7 @@ private void arrayCopy( if (value[i] == null || !containsNonNullValue(bitMaps, results, i, idx, remaining)) { continue; } - Object valueArray = getOrCreateValueArray(i, arrayIndex); + Object valueArray = getOrCreateValueArray(i, arrayIndex, elementIndex); switch (dataTypes.get(i)) { case TEXT: case BLOB: @@ -1098,7 +1067,7 @@ private static boolean containsNonNullValue( return false; } - private Object getOrCreateValueArray(int columnIndex, int arrayIndex) { + private Object getOrCreateValueArray(int columnIndex, int arrayIndex, int elementIndex) { List columnValues = values.get(columnIndex); Object valueArray = columnValues.get(arrayIndex); if (valueArray == null) { @@ -1106,6 +1075,9 @@ private Object getOrCreateValueArray(int columnIndex, int arrayIndex) { columnValues.set(arrayIndex, valueArray); materializedValueArrayCounts[columnIndex]++; materializedValueArrayMemCost += valueListArrayMemCost(dataTypes.get(columnIndex)); + if (elementIndex > 0) { + getBitMap(columnIndex, arrayIndex).markRange(0, elementIndex); + } } return valueArray; } @@ -1140,6 +1112,10 @@ private BitMap getBitMap(int columnIndex, int arrayIndex) { } private boolean markNullValue(int columnIndex, int arrayIndex, int elementIndex) { + List columnValues = values.get(columnIndex); + if (columnValues == null || columnValues.get(arrayIndex) == null) { + return false; + } // mark the null value in the current bitmap BitMap bitMap = getBitMap(columnIndex, arrayIndex); if (bitMap.isMarked(elementIndex)) { @@ -1781,60 +1757,55 @@ public BitMap getAllValueColDeletedMap() { } public BitMap getAllValueColDeletedMap(int rowCount) { - // row exists when any column value exists - if (bitMaps == null) { + if (rowCount <= 0) { return null; } + + int blockCount = (rowCount + ARRAY_SIZE - 1) / ARRAY_SIZE; + // Preserve the dense fast path: one column without null blocks or bitmaps proves that every row + // has at least one value. for (int columnIndex = 0; columnIndex < values.size(); columnIndex++) { - if (values.get(columnIndex) != null && bitMaps.get(columnIndex) == null) { + List columnValues = values.get(columnIndex); + if (columnValues == null) { + continue; + } + List columnBitMaps = bitMaps == null ? null : bitMaps.get(columnIndex); + boolean columnHasNoNull = true; + for (int blockIndex = 0; blockIndex < blockCount; blockIndex++) { + if (columnValues.get(blockIndex) == null + || (columnBitMaps != null && columnBitMaps.get(blockIndex) != null)) { + columnHasNoNull = false; + break; + } + } + if (columnHasNoNull) { return null; } } - byte[] rowBitsArr = new byte[rowCount / Byte.SIZE + 1]; - int bitsMapSize = - rowCount % ARRAY_SIZE == 0 ? rowCount / ARRAY_SIZE : rowCount / ARRAY_SIZE + 1; - boolean[] allNotNullArray = new boolean[bitsMapSize]; + byte[] rowBitsArr = new byte[(rowCount + Byte.SIZE - 1) / Byte.SIZE]; Arrays.fill(rowBitsArr, (byte) 0xFF); - for (int columnIndex = 0; columnIndex < values.size(); columnIndex++) { - List columnBitMaps = bitMaps.get(columnIndex); - if (columnBitMaps == null) { - Arrays.fill(rowBitsArr, (byte) 0x00); - break; - } else if (values.get(columnIndex) != null) { - int row = 0; - boolean isEnd = true; - for (int i = 0; i < bitsMapSize; i++) { - if (allNotNullArray[i]) { - row += ARRAY_SIZE; - continue; - } - - BitMap bitMap = columnBitMaps.get(i); - int index = row / Byte.SIZE; - int size = ((Math.min((rowCount - row), ARRAY_SIZE)) + 7) >>> 3; - row += ARRAY_SIZE; - - if (bitMap == null) { - Arrays.fill(rowBitsArr, index, index + size, (byte) 0x00); - allNotNullArray[i] = true; - continue; - } - - byte bits = (byte) 0X00; - byte[] bitMapBytes = bitMap.getByteArray(); - for (int j = 0; j < size; j++) { - rowBitsArr[index] &= bitMapBytes[j]; - bits |= rowBitsArr[index++]; - isEnd = false; - } - - allNotNullArray[i] = bits == (byte) 0; + for (int blockIndex = 0; blockIndex < blockCount; blockIndex++) { + int row = blockIndex * ARRAY_SIZE; + int byteIndex = row / Byte.SIZE; + int byteCount = (Math.min(rowCount - row, ARRAY_SIZE) + Byte.SIZE - 1) / Byte.SIZE; + for (int columnIndex = 0; columnIndex < values.size(); columnIndex++) { + List columnValues = values.get(columnIndex); + if (columnValues == null || columnValues.get(blockIndex) == null) { + continue; } - if (isEnd) { + List columnBitMaps = bitMaps == null ? null : bitMaps.get(columnIndex); + BitMap bitMap = columnBitMaps == null ? null : columnBitMaps.get(blockIndex); + if (bitMap == null) { + Arrays.fill(rowBitsArr, byteIndex, byteIndex + byteCount, (byte) 0x00); break; } + + byte[] bitMapBytes = bitMap.getByteArray(); + for (int i = 0; i < byteCount; i++) { + rowBitsArr[byteIndex + i] &= bitMapBytes[i]; + } } } return new BitMap(rowCount, rowBitsArr); @@ -1863,19 +1834,9 @@ public int getAvgPointSizeOfLargestColumn() { private int getColumnValueCnt(int columnIndex) { int pointNum = 0; - if (bitMaps == null || bitMaps.get(columnIndex) == null) { - pointNum = rowCount; - } else { - for (int i = 0; i < rowCount; i++) { - int arrayIndex = i / ARRAY_SIZE; - if (bitMaps.get(columnIndex).get(arrayIndex) == null) { - pointNum++; - } else { - int elementIndex = i % ARRAY_SIZE; - if (!bitMaps.get(columnIndex).get(arrayIndex).isMarked(elementIndex)) { - pointNum++; - } - } + for (int i = 0; i < rowCount; i++) { + if (!isNullValue(i, columnIndex)) { + pointNum++; } } return pointNum; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index a8193a9c0b28..8beb960ee4e3 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -589,9 +589,7 @@ public void alignedTabletDoesNotChargePrimitiveArrayForFailedOnlyBlock() Assert.assertEquals( expectedProcessor.getWorkMemTable().getTVListsRamCost() - - AlignedTVList.valueListArrayMemCost(dataType) - + 2 * AlignedTVList.bitmapReferenceRamCost() - + AlignedTVList.bitmapRamCost(), + - AlignedTVList.valueListArrayMemCost(dataType), actualProcessor.getWorkMemTable().getTVListsRamCost()); Assert.assertEquals( TSStatusCode.OUT_OF_TTL.getStatusCode(), actualResults[failedIndex].getCode()); @@ -620,10 +618,7 @@ public void alignedTabletOnlyChargesMaterializedPrimitiveArrays() AlignedTVList.alignedTvListArrayMemCost( new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null); Assert.assertEquals( - denseBlockCost - - AlignedTVList.valueListArrayMemCost(TSDataType.INT32) - + 2 * AlignedTVList.bitmapReferenceRamCost() - + AlignedTVList.bitmapRamCost(), + denseBlockCost - AlignedTVList.valueListArrayMemCost(TSDataType.INT32), processor.getWorkMemTable().getTVListsRamCost() - ramCostBeforeNewBlock); } @@ -649,10 +644,7 @@ public void alignedRowOnlyChargesMaterializedPrimitiveArrays() AlignedTVList.alignedTvListArrayMemCost( new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null); Assert.assertEquals( - denseBlockCost - - AlignedTVList.valueListArrayMemCost(TSDataType.INT32) - + 2 * AlignedTVList.bitmapReferenceRamCost() - + AlignedTVList.bitmapRamCost(), + denseBlockCost - AlignedTVList.valueListArrayMemCost(TSDataType.INT32), processor.getWorkMemTable().getTVListsRamCost() - ramCostBeforeNewBlock); } @@ -700,9 +692,11 @@ public void alignedBitmapMemoryAccountingMatchesActualAllocations() alignedMemChunk.getWorkingTVList().getValues().get(extendedColumnIndex).get(0)); Assert.assertNull( alignedMemChunk.getWorkingTVList().getValues().get(extendedColumnIndex).get(1)); - for (BitMap bitMap : alignedMemChunk.getWorkingTVList().getBitMaps().get(extendedColumnIndex)) { - Assert.assertNotNull(bitMap); - } + List extendedColumnBitMaps = + alignedMemChunk.getWorkingTVList().getBitMaps().get(extendedColumnIndex); + Assert.assertNull(extendedColumnBitMaps.get(0)); + Assert.assertNull(extendedColumnBitMaps.get(1)); + Assert.assertNotNull(extendedColumnBitMaps.get(2)); assertAlignedTvListRamCostMatchesActual(deviceId); } @@ -771,7 +765,8 @@ public void alignedTvListRamCostTest2() new TSStatus[10], true, new long[5]); - Assert.assertEquals(5269104, memTable.getTVListsRamCost()); + Assert.assertEquals( + 5269104 - 3000L * AlignedTVList.bitmapRamCost(), memTable.getTVListsRamCost()); processor.insertTablet( genInsertTableNodeFors3000ToS6000(300, true), Collections.singletonList(new int[] {0, 10}), @@ -1069,9 +1064,7 @@ public void testAlignedSparseRowDoesNotChargeUnallocatedPrimitiveArrayOnNewBlock long denseRowRamIncrement = memTable.getTVListsRamCost() - ramCostBeforeDenseRow; Assert.assertEquals( - AlignedTVList.valueListArrayMemCost(dataType) - - 2 * AlignedTVList.bitmapReferenceRamCost() - - AlignedTVList.bitmapRamCost(), + AlignedTVList.valueListArrayMemCost(dataType), denseRowRamIncrement - sparseRowRamIncrement); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedSparseWritePerformanceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedSparseWritePerformanceTest.java new file mode 100644 index 000000000000..8a759a6792b5 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/AlignedSparseWritePerformanceTest.java @@ -0,0 +1,314 @@ +/* + * 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.utils.datastructure; + +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils; +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement; +import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary; + +import org.apache.tsfile.enums.TSDataType; +import org.junit.Assert; +import org.junit.Assume; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; + +import static org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE; + +public class AlignedSparseWritePerformanceTest { + + private static final String PREFIX = "iotdb.aligned.sparse-write.perf."; + private static final String ENABLED_PROPERTY = PREFIX + "enabled"; + private static final String COLUMNS_PROPERTY = PREFIX + "columns"; + private static final String WRITTEN_COLUMNS_PROPERTY = PREFIX + "written-columns"; + private static final String ROWS_PROPERTY = PREFIX + "rows"; + private static final String HISTORICAL_ROWS_PROPERTY = PREFIX + "historical-rows"; + private static final String WARMUPS_PROPERTY = PREFIX + "warmup.iterations"; + private static final String ITERATIONS_PROPERTY = PREFIX + "iterations"; + private static final String ROUNDS_PROPERTY = PREFIX + "rounds"; + + private static volatile long benchmarkBlackhole; + + @Test + public void sparseAlignedWriteBenchmark() { + Assume.assumeTrue( + String.format( + Locale.ROOT, + "Manual performance UT. Enable with -D%s=true and tune properties under -D%s*.", + ENABLED_PROPERTY, + PREFIX), + Boolean.getBoolean(ENABLED_PROPERTY)); + Assume.assumeTrue( + "Current-thread CPU time and allocation metrics are required.", + ManualPerformanceTestUtils.enableThreadMetrics()); + + int columns = Integer.getInteger(COLUMNS_PROPERTY, 64); + int writtenColumns = Integer.getInteger(WRITTEN_COLUMNS_PROPERTY, 1); + int rows = Integer.getInteger(ROWS_PROPERTY, ARRAY_SIZE); + int historicalRows = Integer.getInteger(HISTORICAL_ROWS_PROPERTY, ARRAY_SIZE * 1024); + int warmups = Integer.getInteger(WARMUPS_PROPERTY, 200); + int iterations = Integer.getInteger(ITERATIONS_PROPERTY, 5000); + int rounds = Integer.getInteger(ROUNDS_PROPERTY, 5); + Assert.assertTrue(columns > 0); + Assert.assertTrue(writtenColumns > 0 && writtenColumns <= columns); + Assert.assertTrue(rows > 0); + Assert.assertTrue(historicalRows > 0); + Assert.assertTrue(warmups > 0); + Assert.assertTrue(iterations > 0); + Assert.assertTrue(rounds > 0); + + List dataTypes = createDataTypes(columns); + Summary dense = + runWriteScenario( + "dense", + dataTypes, + Scenario.batch(columns, columns, rows), + warmups, + iterations, + rounds); + Summary sparse = + runWriteScenario( + "sparse", + dataTypes, + Scenario.batch(columns, writtenColumns, rows), + warmups, + iterations, + rounds); + Summary allNull = + runWriteScenario( + "all-null", dataTypes, Scenario.batch(columns, 0, rows), warmups, iterations, rounds); + Summary nullPrefix = + runWriteScenario( + "null-prefix-first-non-null", + dataTypes, + Scenario.nullPrefix(columns, writtenColumns, rows), + warmups, + iterations, + rounds); + runExtensionScenario(historicalRows, warmups, iterations, rounds); + + printComparison("sparse/dense", sparse, dense); + printComparison("all-null/dense", allNull, dense); + printComparison("null-prefix/sparse", nullPrefix, sparse); + printAvoidedBitmapAccounting(columns, writtenColumns, rows, historicalRows); + } + + private static Summary runWriteScenario( + String label, + List dataTypes, + Scenario scenario, + int warmups, + int iterations, + int rounds) { + runWrite(AlignedTVList.newAlignedList(new ArrayList<>(dataTypes)), scenario, warmups); + Measurement[] measurements = new Measurement[rounds]; + for (int round = 0; round < rounds; round++) { + AlignedTVList target = AlignedTVList.newAlignedList(new ArrayList<>(dataTypes)); + measurements[round] = + ManualPerformanceTestUtils.measure(1, () -> runWrite(target, scenario, iterations)); + } + Summary summary = ManualPerformanceTestUtils.summarize(measurements, iterations); + System.out.printf( + Locale.ROOT, + "Aligned sparse-write benchmark (%s): columns=%d, written columns=%d, rows=%d, batches/round=%d, rounds=%d%n", + label, + dataTypes.size(), + scenario.writtenColumnCount, + scenario.rowCount, + iterations, + rounds); + printSummary(summary, "batch"); + return summary; + } + + private static void runWrite(AlignedTVList target, Scenario scenario, int iterations) { + for (int i = 0; i < iterations; i++) { + scenario.writeTo(target); + } + benchmarkBlackhole = target.getRamSize() + target.rowCount(); + } + + private static List createDataTypes(int columnCount) { + List dataTypes = new ArrayList<>(columnCount); + for (int column = 0; column < columnCount; column++) { + dataTypes.add(TSDataType.INT32); + } + return dataTypes; + } + + private static long[] createTimes(int rowCount, int startTime) { + long[] times = new long[rowCount]; + for (int row = 0; row < rowCount; row++) { + times[row] = startTime + row; + } + return times; + } + + private static Object[] createColumns(int columnCount, int writtenColumnCount, int rowCount) { + Object[] columns = new Object[columnCount]; + for (int column = 0; column < writtenColumnCount; column++) { + int[] columnValues = new int[rowCount]; + for (int row = 0; row < rowCount; row++) { + columnValues[row] = row; + } + columns[column] = columnValues; + } + return columns; + } + + private static void runExtensionScenario( + int historicalRows, int warmups, int iterations, int rounds) { + runExtensions(createHistoricalTarget(historicalRows), warmups); + Measurement[] measurements = new Measurement[rounds]; + for (int round = 0; round < rounds; round++) { + AlignedTVList target = createHistoricalTarget(historicalRows); + measurements[round] = + ManualPerformanceTestUtils.measure(1, () -> runExtensions(target, iterations)); + } + Summary summary = ManualPerformanceTestUtils.summarize(measurements, iterations); + System.out.printf( + Locale.ROOT, + "Aligned sparse-write benchmark (extend-column): historical rows=%d, historical blocks=%d, extensions/round=%d, rounds=%d%n", + historicalRows, + (historicalRows + ARRAY_SIZE - 1) / ARRAY_SIZE, + iterations, + rounds); + printSummary(summary, "column"); + } + + private static AlignedTVList createHistoricalTarget(int historicalRows) { + long[] times = new long[historicalRows]; + int[] values = new int[historicalRows]; + for (int row = 0; row < historicalRows; row++) { + times[row] = row; + values[row] = row; + } + AlignedTVList target = AlignedTVList.newAlignedList(new ArrayList<>(List.of(TSDataType.INT32))); + target.putAlignedValues(times, new Object[] {values}, null, 0, historicalRows, null); + return target; + } + + private static void runExtensions(AlignedTVList target, int count) { + for (int i = 0; i < count; i++) { + target.extendColumn(TSDataType.INT32); + } + benchmarkBlackhole = target.getRamSize() + target.getTsDataTypes().size(); + } + + private static void printSummary(Summary summary, String operation) { + System.out.printf( + Locale.ROOT, + " CPU=%.3f us/%s, allocated=%.1f bytes/%s, peak heap delta=%.3f MiB/round%n", + summary.getCpuNanosPerOperation() / 1_000.0, + operation, + summary.getAllocatedBytesPerOperation(), + operation, + summary.getPeakHeapDeltaBytes() / 1024.0 / 1024.0); + if (summary.getCpuNanosPerOperation() == 0) { + System.out.printf( + " CPU sample is below the platform timer resolution; increase -D%s.%n", + ITERATIONS_PROPERTY); + } + } + + private static void printComparison(String label, Summary numerator, Summary denominator) { + System.out.printf( + Locale.ROOT, + " %s CPU ratio=%.2f%%, allocation ratio=%.2f%%%n", + label, + percentage(numerator.getCpuNanosPerOperation(), denominator.getCpuNanosPerOperation()), + percentage( + numerator.getAllocatedBytesPerOperation(), + denominator.getAllocatedBytesPerOperation())); + } + + private static double percentage(double numerator, double denominator) { + return denominator == 0 ? 0 : numerator * 100.0 / denominator; + } + + private static void printAvoidedBitmapAccounting( + int columns, int writtenColumns, int rows, int historicalRows) { + long bitmapCost = AlignedTVList.bitmapReferenceRamCost() + AlignedTVList.bitmapRamCost(); + long blocks = (rows + ARRAY_SIZE - 1L) / ARRAY_SIZE; + long sparseBytes = (columns - writtenColumns) * blocks * bitmapCost; + long allNullBytes = columns * blocks * bitmapCost; + long historicalBlocks = (historicalRows + ARRAY_SIZE - 1L) / ARRAY_SIZE; + System.out.printf( + Locale.ROOT, + " avoided bitmap RAM accounting: sparse=%d bytes/batch, all-null=%d bytes/batch, extension=%d bytes/column%n", + sparseBytes, + allNullBytes, + historicalBlocks * bitmapCost); + } + + private static final class Scenario { + + private final long[] firstTimes; + private final Object[] firstColumns; + private final long[] secondTimes; + private final Object[] secondColumns; + private final int rowCount; + private final int writtenColumnCount; + + private Scenario( + long[] firstTimes, + Object[] firstColumns, + long[] secondTimes, + Object[] secondColumns, + int writtenColumnCount) { + this.firstTimes = firstTimes; + this.firstColumns = firstColumns; + this.secondTimes = secondTimes; + this.secondColumns = secondColumns; + rowCount = firstTimes.length + (secondTimes == null ? 0 : secondTimes.length); + this.writtenColumnCount = writtenColumnCount; + } + + private static Scenario batch(int columns, int writtenColumns, int rows) { + return new Scenario( + createTimes(rows, 0), + createColumns(columns, writtenColumns, rows), + null, + null, + writtenColumns); + } + + private static Scenario nullPrefix(int columns, int writtenColumns, int rows) { + int prefixRows = rows - 1; + return new Scenario( + createTimes(prefixRows, 0), + new Object[columns], + createTimes(1, prefixRows), + createColumns(columns, writtenColumns, 1), + writtenColumns); + } + + private void writeTo(AlignedTVList target) { + if (firstTimes.length > 0) { + target.putAlignedValues(firstTimes, firstColumns, null, 0, firstTimes.length, null); + } + if (secondTimes != null) { + target.putAlignedValues(secondTimes, secondColumns, null, 0, secondTimes.length, null); + } + } + } +} 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 afc984c4f8c1..133bf794424f 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 @@ -306,20 +306,26 @@ public void testPrimitiveArraysAreAllocatedOnFirstWrite() { Assert.assertNull(tvList.getValues().get(2).get(0)); Assert.assertNull(tvList.getValues().get(2).get(1)); - Assert.assertEquals( - 2L * (AlignedTVList.bitmapReferenceRamCost() + AlignedTVList.bitmapRamCost()), - tvList.calculateRamSize().getRamSize() - ramSizeBeforeExtension); + Assert.assertNull(tvList.getBitMaps().get(2)); + Assert.assertEquals(ramSizeBeforeExtension, tvList.calculateRamSize().getRamSize()); long ramSizeBeforeExtendedColumnMaterialization = tvList.calculateRamSize().getRamSize(); 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.assertNull(tvList.getBitMaps().get(2).get(0)); + Assert.assertNotNull(tvList.getBitMaps().get(2).get(1)); + Assert.assertTrue(tvList.getBitMaps().get(2).get(1).isMarked(0)); + Assert.assertTrue(tvList.getBitMaps().get(2).get(1).isMarked(1)); + Assert.assertFalse(tvList.getBitMaps().get(2).get(1).isMarked(2)); Assert.assertTrue(tvList.isNullValue(0, 2)); Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE + 2, 2)); Assert.assertEquals(2, tvList.getIntByValueIndex(ARRAY_SIZE + 2, 2)); Assert.assertEquals( - AlignedTVList.valueListArrayMemCost(TSDataType.INT32), + AlignedTVList.valueListArrayMemCost(TSDataType.INT32) + + 2L * AlignedTVList.bitmapReferenceRamCost() + + AlignedTVList.bitmapRamCost(), tvList.calculateRamSize().getRamSize() - ramSizeBeforeExtendedColumnMaterialization); } @@ -335,7 +341,9 @@ public void testCalculateRamSizeCountsMaterializedPrimitiveArrays() { tvList.putAlignedValue(ARRAY_SIZE + 1L, new Object[] {1L, 1L}); Assert.assertEquals( - AlignedTVList.valueListArrayMemCost(TSDataType.INT64), + AlignedTVList.valueListArrayMemCost(TSDataType.INT64) + + 2L * AlignedTVList.bitmapReferenceRamCost() + + AlignedTVList.bitmapRamCost(), tvList.calculateRamSize().getRamSize() - ramSizeBeforeMaterialization); Assert.assertEquals( @@ -351,7 +359,8 @@ public void testCalculateRamSizeCountsMaterializedPrimitiveArrays() { * projectedTvList.alignedTvListArrayMemCostWithoutPrimitiveArrays() + AlignedTVList.valueListArrayMemCost(TSDataType.INT64) + (long) projectedTvList.getBitMaps().get(0).size() - * (AlignedTVList.bitmapReferenceRamCost() + AlignedTVList.bitmapRamCost()), + * AlignedTVList.bitmapReferenceRamCost() + + AlignedTVList.bitmapRamCost(), projectedTvList.calculateRamSize().getRamSize()); tvList.clear(); @@ -369,12 +378,10 @@ public void testCalculateRamSizeExcludesUnallocatedPrimitiveArrays() { int blockCount = tvList.getValues().get(0).size(); long denseRamSize = blockCount * tvList.alignedTvListArrayMemCost(); long expectedRamSize = - denseRamSize - - blockCount * AlignedTVList.valueListArrayMemCost(TSDataType.INT64) - + (long) blockCount - * (AlignedTVList.bitmapReferenceRamCost() + AlignedTVList.bitmapRamCost()); + denseRamSize - blockCount * AlignedTVList.valueListArrayMemCost(TSDataType.INT64); Assert.assertEquals(expectedRamSize, tvList.calculateRamSize().getRamSize()); + Assert.assertNull(tvList.getBitMaps()); } @Test @@ -395,9 +402,91 @@ public void testBatchDoesNotAllocateAllNullPrimitiveArray() { Assert.assertNotNull(tvList.getValues().get(0).get(0)); Assert.assertNull(tvList.getValues().get(1).get(0)); + Assert.assertNull(tvList.getBitMaps()); Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE - 1, 1)); } + @Test + public void testImplicitNullBlocksAndAllValueDeletedMap() { + AlignedTVList tvList = + AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64, TSDataType.INT64)); + for (int i = 0; i <= ARRAY_SIZE; i++) { + tvList.putAlignedValue(i, new Object[] {null, null}); + } + + Assert.assertNull(tvList.getBitMaps()); + Assert.assertNull(tvList.getValues().get(0).get(0)); + Assert.assertNull(tvList.getValues().get(0).get(1)); + Assert.assertNull(tvList.getValues().get(1).get(0)); + Assert.assertNull(tvList.getValues().get(1).get(1)); + BitMap allValueDeletedMap = tvList.getAllValueColDeletedMap(); + for (int i = 0; i <= ARRAY_SIZE; i++) { + Assert.assertTrue(allValueDeletedMap.isMarked(i)); + } + + tvList.putAlignedValue(ARRAY_SIZE + 1L, new Object[] {1L, null}); + tvList.putAlignedValue(ARRAY_SIZE + 2L, new Object[] {null, 2L}); + + Assert.assertNull(tvList.getBitMaps().get(0).get(0)); + Assert.assertNull(tvList.getBitMaps().get(1).get(0)); + Assert.assertNotNull(tvList.getBitMaps().get(0).get(1)); + Assert.assertNotNull(tvList.getBitMaps().get(1).get(1)); + allValueDeletedMap = tvList.getAllValueColDeletedMap(); + for (int i = 0; i <= ARRAY_SIZE; i++) { + Assert.assertTrue(allValueDeletedMap.isMarked(i)); + } + Assert.assertFalse(allValueDeletedMap.isMarked(ARRAY_SIZE + 1)); + Assert.assertFalse(allValueDeletedMap.isMarked(ARRAY_SIZE + 2)); + } + + @Test + public void testDeletingImplicitNullColumnDoesNotMaterializeBitmaps() { + AlignedTVList tvList = + AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64, TSDataType.INT64)); + for (int i = 0; i < ARRAY_SIZE; i++) { + tvList.putAlignedValue(i, new Object[] {(long) i, null}); + } + + Assert.assertEquals(0, (int) tvList.delete(0, ARRAY_SIZE - 1L, 1).left); + Assert.assertNull(tvList.getBitMaps()); + tvList.deleteColumn(1); + Assert.assertNull(tvList.getBitMaps()); + + tvList.deleteColumn(0); + Assert.assertNotNull(tvList.getBitMaps().get(0).get(0)); + Assert.assertNull(tvList.getBitMaps().get(1)); + for (int i = 0; i < ARRAY_SIZE; i++) { + Assert.assertTrue(tvList.isNullValue(i, 0)); + Assert.assertTrue(tvList.isNullValue(i, 1)); + } + } + + @Test + public void testAllValueDeletedMapMatchesPerColumnNullState() { + AlignedTVList tvList = + AlignedTVList.newAlignedList( + Arrays.asList(TSDataType.INT64, TSDataType.INT64, TSDataType.INT64)); + int rowCount = ARRAY_SIZE * 2 + 7; + for (int row = 0; row < rowCount; row++) { + tvList.putAlignedValue( + row, + new Object[] { + row % 3 == 0 ? null : (long) row, + row % 5 == 0 ? null : (long) row, + row < ARRAY_SIZE || row % 7 == 0 ? null : (long) row + }); + } + + BitMap allValueDeletedMap = tvList.getAllValueColDeletedMap(); + for (int row = 0; row < rowCount; row++) { + boolean allNull = true; + for (int column = 0; column < 3; column++) { + allNull &= tvList.isNullValue(row, column); + } + Assert.assertEquals(allNull, allValueDeletedMap != null && allValueDeletedMap.isMarked(row)); + } + } + @Test public void testNullPrimitiveArrayCanBeClonedAndSerialized() throws IOException { AlignedTVList tvList = From e04c71af20655a36c9aa85de9be4b7f4a42823db Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 30 Jul 2026 16:31:18 +0800 Subject: [PATCH 3/3] [Performance] Avoid device collection for single-device aligned tablets --- .../write/RelationalInsertTabletNode.java | 4 + .../dataregion/memtable/TsFileProcessor.java | 38 +++- ...BitmapMemoryAccountingPerformanceTest.java | 214 +++++++++++++++++- .../memtable/TsFileProcessorTest.java | 55 +++++ 4 files changed, 291 insertions(+), 20 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java index d4c373e57d7f..029c55499bf9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/RelationalInsertTabletNode.java @@ -116,6 +116,10 @@ public void setSingleDevice() { this.singleDevice = true; } + public boolean isSingleDevice() { + return singleDevice; + } + public List getObjectColumns() { List objectColumns = new ArrayList<>(); for (int i = 0; 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 ee9b39d2ffe4..759b58d4966a 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 @@ -48,6 +48,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.schemaengine.schemaregion.utils.ResourceByPathUtils; import org.apache.iotdb.db.service.metrics.WritingMetrics; import org.apache.iotdb.db.storageengine.dataregion.DataRegion; @@ -648,19 +649,8 @@ public void insertTablet( ensureMemTable(infoForMetrics); workMemTable.checkDataType(insertTabletNode); - Set alignedDeviceIds = new HashSet<>(); - if (insertTabletNode.isAligned()) { - for (int[] range : rangeList) { - for (Pair deviceEndPosition : - insertTabletNode.splitByDevice(range[0], range[1])) { - alignedDeviceIds.add(deviceEndPosition.getLeft()); - } - } - } AlignedTVListRamCostSnapshot alignedRamCostSnapshot = - alignedDeviceIds.isEmpty() - ? null - : new AlignedTVListRamCostSnapshot(workMemTable, alignedDeviceIds); + takeAlignedTVListRamCostSnapshot(workMemTable, insertTabletNode, rangeList); long[] memIncrements = scheduleMemoryBlock(insertTabletNode, rangeList, results, infoForMetrics); @@ -1394,6 +1384,30 @@ private void reconcileAlignedTVListRamCost( } } + static AlignedTVListRamCostSnapshot takeAlignedTVListRamCostSnapshot( + IMemTable memTable, InsertTabletNode insertTabletNode, List rangeList) { + if (!insertTabletNode.isAligned() || rangeList.isEmpty()) { + return null; + } + + if (!(insertTabletNode instanceof RelationalInsertTabletNode) + || ((RelationalInsertTabletNode) insertTabletNode).isSingleDevice()) { + return new AlignedTVListRamCostSnapshot( + memTable, insertTabletNode.getDeviceID(rangeList.get(0)[0])); + } + + Set alignedDeviceIds = new HashSet<>(); + for (int[] range : rangeList) { + for (Pair deviceEndPosition : + insertTabletNode.splitByDevice(range[0], range[1])) { + alignedDeviceIds.add(deviceEndPosition.getLeft()); + } + } + return alignedDeviceIds.isEmpty() + ? null + : new AlignedTVListRamCostSnapshot(memTable, alignedDeviceIds); + } + static final class AlignedTVListRamCostSnapshot { private final IMemTable memTable; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java index 8a872f6af61e..b18b543a2d53 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedBitmapMemoryAccountingPerformanceTest.java @@ -19,6 +19,10 @@ package org.apache.iotdb.db.storageengine.dataregion.memtable; +import org.apache.iotdb.commons.exception.IllegalPathException; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.utils.ManualPerformanceTestUtils; import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Measurement; import org.apache.iotdb.db.utils.ManualPerformanceTestUtils.Summary; @@ -26,6 +30,7 @@ import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.IDeviceID; import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.utils.Pair; import org.apache.tsfile.write.schema.IMeasurementSchema; import org.apache.tsfile.write.schema.MeasurementSchema; import org.junit.Assert; @@ -33,9 +38,12 @@ import org.junit.Test; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Locale; +import java.util.Set; public class AlignedBitmapMemoryAccountingPerformanceTest { @@ -49,10 +57,12 @@ public class AlignedBitmapMemoryAccountingPerformanceTest { private static final String ROUNDS_PROPERTY = "iotdb.aligned.bitmap.accounting.perf.rounds"; private static final int RECONCILIATION_REPETITIONS = 1024; + private static final TsFileProcessor.AlignedTVListRamCostSnapshot[] SNAPSHOT_BLACKHOLE = + new TsFileProcessor.AlignedTVListRamCostSnapshot[RECONCILIATION_REPETITIONS]; private static volatile long benchmarkBlackhole; @Test - public void alignedTabletBitmapAccountingBenchmark() { + public void alignedTabletBitmapAccountingBenchmark() throws IllegalPathException { Assume.assumeTrue( String.format( "Manual performance UT. Enable with -D%s=true, optionally tune -D%s, -D%s, -D%s, -D%s and -D%s.", @@ -78,12 +88,11 @@ public void alignedTabletBitmapAccountingBenchmark() { Assert.assertTrue(iterations > 0); Assert.assertTrue(rounds > 0); - runScenario( - "dense", - createScenario(columnCount, rowCount, false), - warmupIterations, - iterations, - rounds); + Scenario denseScenario = createScenario(columnCount, rowCount, false); + Summary denseWriteSummary = + runScenario("dense", denseScenario, warmupIterations, iterations, rounds); + runSingleDeviceSnapshotScenario( + denseScenario, warmupIterations, iterations, rounds, denseWriteSummary); runScenario( "null-heavy", createScenario(columnCount, rowCount, true), @@ -92,7 +101,7 @@ public void alignedTabletBitmapAccountingBenchmark() { rounds); } - private static void runScenario( + private static Summary runScenario( String label, Scenario scenario, int warmupIterations, int iterations, int rounds) { runReconciliation( createAccountingTarget(scenario), warmupIterations * RECONCILIATION_REPETITIONS); @@ -116,6 +125,107 @@ private static void runScenario( Summary writeSummary = ManualPerformanceTestUtils.summarize(writeMeasurements, iterations); printResult( label, scenario, warmupIterations, iterations, rounds, reconciliationSummary, writeSummary); + return writeSummary; + } + + private static void runSingleDeviceSnapshotScenario( + Scenario scenario, int warmupIterations, int iterations, int rounds, Summary writeSummary) + throws IllegalPathException { + SnapshotTarget target = createSnapshotTarget(scenario); + int warmupOperations = Math.multiplyExact(warmupIterations, RECONCILIATION_REPETITIONS); + int operations = Math.multiplyExact(iterations, RECONCILIATION_REPETITIONS); + + TsFileProcessor.AlignedTVListRamCostSnapshot legacySnapshot = + takeLegacyAlignedTVListRamCostSnapshot( + target.memTable, target.insertTabletNode, target.rangeList); + TsFileProcessor.AlignedTVListRamCostSnapshot optimizedSnapshot = + TsFileProcessor.takeAlignedTVListRamCostSnapshot( + target.memTable, target.insertTabletNode, target.rangeList); + Assert.assertNotNull(legacySnapshot); + Assert.assertNotNull(optimizedSnapshot); + Assert.assertEquals( + legacySnapshot.getMemoryCorrection(0), optimizedSnapshot.getMemoryCorrection(0)); + + runLegacySnapshotLifecycle(target, warmupOperations); + runOptimizedSnapshotLifecycle(target, warmupOperations); + + Measurement[] legacyMeasurements = new Measurement[rounds]; + Measurement[] optimizedMeasurements = new Measurement[rounds]; + for (int i = 0; i < rounds; i++) { + if ((i & 1) == 0) { + legacyMeasurements[i] = measureLegacySnapshotLifecycle(target, operations); + optimizedMeasurements[i] = measureOptimizedSnapshotLifecycle(target, operations); + } else { + optimizedMeasurements[i] = measureOptimizedSnapshotLifecycle(target, operations); + legacyMeasurements[i] = measureLegacySnapshotLifecycle(target, operations); + } + } + + Summary legacySummary = ManualPerformanceTestUtils.summarize(legacyMeasurements, operations); + Summary optimizedSummary = + ManualPerformanceTestUtils.summarize(optimizedMeasurements, operations); + printSnapshotResult( + scenario, + warmupOperations, + operations, + rounds, + legacySummary, + optimizedSummary, + writeSummary); + } + + private static Measurement measureLegacySnapshotLifecycle(SnapshotTarget target, int operations) { + Arrays.fill(SNAPSHOT_BLACKHOLE, null); + return ManualPerformanceTestUtils.measure( + 1, () -> runLegacySnapshotLifecycle(target, operations)); + } + + private static Measurement measureOptimizedSnapshotLifecycle( + SnapshotTarget target, int operations) { + Arrays.fill(SNAPSHOT_BLACKHOLE, null); + return ManualPerformanceTestUtils.measure( + 1, () -> runOptimizedSnapshotLifecycle(target, operations)); + } + + private static void runLegacySnapshotLifecycle(SnapshotTarget target, int operations) { + long correction = 0; + for (int i = 0; i < operations; i++) { + TsFileProcessor.AlignedTVListRamCostSnapshot snapshot = + takeLegacyAlignedTVListRamCostSnapshot( + target.memTable, target.insertTabletNode, target.rangeList); + correction += snapshot.getMemoryCorrection(0); + SNAPSHOT_BLACKHOLE[i % SNAPSHOT_BLACKHOLE.length] = snapshot; + } + benchmarkBlackhole = correction + operations; + } + + private static void runOptimizedSnapshotLifecycle(SnapshotTarget target, int operations) { + long correction = 0; + for (int i = 0; i < operations; i++) { + TsFileProcessor.AlignedTVListRamCostSnapshot snapshot = + TsFileProcessor.takeAlignedTVListRamCostSnapshot( + target.memTable, target.insertTabletNode, target.rangeList); + correction += snapshot.getMemoryCorrection(0); + SNAPSHOT_BLACKHOLE[i % SNAPSHOT_BLACKHOLE.length] = snapshot; + } + benchmarkBlackhole = correction + operations; + } + + private static TsFileProcessor.AlignedTVListRamCostSnapshot + takeLegacyAlignedTVListRamCostSnapshot( + IMemTable memTable, InsertTabletNode insertTabletNode, List rangeList) { + Set alignedDeviceIds = new HashSet<>(); + if (insertTabletNode.isAligned()) { + for (int[] range : rangeList) { + for (Pair deviceEndPosition : + insertTabletNode.splitByDevice(range[0], range[1])) { + alignedDeviceIds.add(deviceEndPosition.getLeft()); + } + } + } + return alignedDeviceIds.isEmpty() + ? null + : new TsFileProcessor.AlignedTVListRamCostSnapshot(memTable, alignedDeviceIds); } private static Measurement measureReconciliation(Scenario scenario, int iterations) { @@ -184,6 +294,26 @@ private static AccountingTarget createAccountingTarget(Scenario scenario) { return new AccountingTarget(memTable, deviceId); } + private static SnapshotTarget createSnapshotTarget(Scenario scenario) + throws IllegalPathException { + AccountingTarget accountingTarget = createAccountingTarget(scenario); + InsertTabletNode insertTabletNode = + new InsertTabletNode( + new PlanNodeId("snapshot_perf"), + new PartialPath("root.accounting.d0"), + true, + scenario.measurements, + scenario.dataTypes, + scenario.times, + scenario.bitMaps, + scenario.columns, + scenario.times.length); + return new SnapshotTarget( + accountingTarget.memTable, + insertTabletNode, + Collections.singletonList(new int[] {0, scenario.times.length})); + } + private static Scenario createScenario(int columnCount, int rowCount, boolean nullHeavy) { String[] measurements = new String[columnCount]; TSDataType[] dataTypes = new TSDataType[columnCount]; @@ -243,6 +373,60 @@ private static void printResult( writeSummary.getAllocatedBytesPerOperation())); } + private static void printSnapshotResult( + Scenario scenario, + int warmupOperations, + int operations, + int rounds, + Summary legacySummary, + Summary optimizedSummary, + Summary writeSummary) { + System.out.printf( + Locale.ROOT, + "Aligned single-device snapshot lifecycle benchmark (dense tree tablet): columns=%d, rows=%d, warmup operations=%d, operations/round=%d, rounds=%d%n", + scenario.measurements.length, + scenario.times.length, + warmupOperations, + operations, + rounds); + printSnapshotSummary("legacy", legacySummary); + printSnapshotSummary("optimized", optimizedSummary); + System.out.printf( + Locale.ROOT, + " optimized/legacy CPU ratio=%.2f%%, allocation ratio=%.2f%%%n", + percentage( + optimizedSummary.getCpuNanosPerOperation(), legacySummary.getCpuNanosPerOperation()), + percentage( + optimizedSummary.getAllocatedBytesPerOperation(), + legacySummary.getAllocatedBytesPerOperation())); + System.out.printf( + Locale.ROOT, + " optimized-legacy CPU delta=%+.3f ns/tablet, allocation delta=%+.1f bytes/tablet%n", + optimizedSummary.getCpuNanosPerOperation() - legacySummary.getCpuNanosPerOperation(), + optimizedSummary.getAllocatedBytesPerOperation() + - legacySummary.getAllocatedBytesPerOperation()); + System.out.printf( + Locale.ROOT, + " savings/dense-write CPU ratio=%.2f%%, allocation ratio=%.2f%%%n", + percentage( + legacySummary.getCpuNanosPerOperation() - optimizedSummary.getCpuNanosPerOperation(), + writeSummary.getCpuNanosPerOperation()), + percentage( + legacySummary.getAllocatedBytesPerOperation() + - optimizedSummary.getAllocatedBytesPerOperation(), + writeSummary.getAllocatedBytesPerOperation())); + } + + private static void printSnapshotSummary(String label, Summary summary) { + System.out.printf( + Locale.ROOT, + " %-10s CPU=%.3f ns/tablet, allocated=%.1f bytes/tablet, peak heap delta=%.3f MiB%n", + label, + summary.getCpuNanosPerOperation(), + summary.getAllocatedBytesPerOperation(), + summary.getPeakHeapDeltaBytes() / 1024.0 / 1024.0); + } + private static void printSummary(String label, Summary summary) { System.out.printf( Locale.ROOT, @@ -268,6 +452,20 @@ private AccountingTarget(IMemTable memTable, IDeviceID deviceId) { } } + private static final class SnapshotTarget { + + private final IMemTable memTable; + private final InsertTabletNode insertTabletNode; + private final List rangeList; + + private SnapshotTarget( + IMemTable memTable, InsertTabletNode insertTabletNode, List rangeList) { + this.memTable = memTable; + this.insertTabletNode = insertTabletNode; + this.rangeList = rangeList; + } + } + private static final class Scenario { private final String[] measurements; diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java index 8beb960ee4e3..43832097c0ef 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessorTest.java @@ -39,6 +39,7 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; import org.apache.iotdb.db.storageengine.dataregion.DataRegionInfo; import org.apache.iotdb.db.storageengine.dataregion.DataRegionTest; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; @@ -65,6 +66,7 @@ import org.apache.tsfile.read.reader.IPointReader; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.utils.Pair; import org.apache.tsfile.write.record.TSRecord; import org.apache.tsfile.write.record.datapoint.DataPoint; import org.apache.tsfile.write.schema.MeasurementSchema; @@ -700,6 +702,59 @@ public void alignedBitmapMemoryAccountingMatchesActualAllocations() assertAlignedTvListRamCostMatchesActual(deviceId); } + @Test + public void alignedTabletRamCostSnapshotUsesSingleDeviceFastPath() { + IMemTable memTable = new PrimitiveMemTable("root.snapshot", "0"); + IDeviceID firstDevice = IDeviceID.Factory.DEFAULT_FACTORY.create("root.snapshot.d0"); + IDeviceID secondDevice = IDeviceID.Factory.DEFAULT_FACTORY.create("root.snapshot.d1"); + List rangeList = Collections.singletonList(new int[] {0, 2}); + + int[] treeSplitCalls = {0}; + InsertTabletNode treeNode = + new InsertTabletNode(new PlanNodeId("tree")) { + @Override + public IDeviceID getDeviceID(int rowIdx) { + return firstDevice; + } + + @Override + public List> splitByDevice(int start, int end) { + treeSplitCalls[0]++; + return Collections.singletonList(new Pair<>(firstDevice, end)); + } + }; + treeNode.setAligned(true); + + Assert.assertNotNull( + TsFileProcessor.takeAlignedTVListRamCostSnapshot(memTable, treeNode, rangeList)); + Assert.assertEquals(0, treeSplitCalls[0]); + + int[] relationalSplitCalls = {0}; + RelationalInsertTabletNode relationalNode = + new RelationalInsertTabletNode(new PlanNodeId("table")) { + @Override + public IDeviceID getDeviceID(int rowIdx) { + return rowIdx == 0 ? firstDevice : secondDevice; + } + + @Override + public List> splitByDevice(int start, int end) { + relationalSplitCalls[0]++; + return Arrays.asList(new Pair<>(firstDevice, 1), new Pair<>(secondDevice, end)); + } + }; + relationalNode.setAligned(true); + + Assert.assertNotNull( + TsFileProcessor.takeAlignedTVListRamCostSnapshot(memTable, relationalNode, rangeList)); + Assert.assertEquals(1, relationalSplitCalls[0]); + + relationalNode.setSingleDevice(); + Assert.assertNotNull( + TsFileProcessor.takeAlignedTVListRamCostSnapshot(memTable, relationalNode, rangeList)); + Assert.assertEquals(1, relationalSplitCalls[0]); + } + @Test public void alignedTvListRamCostTest2() throws MetadataException, WriteProcessException, IOException {