This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch use_long_backed_bitmap_for_memtable in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e8a1ea7e5eb88a1daa8a8fc3962f174ee9f93b25 Author: Tian Jiang <[email protected]> AuthorDate: Tue Jul 21 18:29:39 2026 +0800 Use DynamicBitMap for MemTable --- .../IoTDBAlignedTVListBitMapPerformanceIT.java | 163 +++++++++++++++++++++ .../db/utils/datastructure/AlignedTVList.java | 56 +++---- .../iotdb/db/utils/datastructure/LazyBitMap.java | 2 +- .../db/utils/datastructure/AlignedTVListTest.java | 31 ++++ pom.xml | 2 +- 5 files changed, 220 insertions(+), 34 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/performance/IoTDBAlignedTVListBitMapPerformanceIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/performance/IoTDBAlignedTVListBitMapPerformanceIT.java new file mode 100644 index 00000000000..d79119e0ebb --- /dev/null +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/performance/IoTDBAlignedTVListBitMapPerformanceIT.java @@ -0,0 +1,163 @@ +/* + * 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.performance; + +import org.apache.iotdb.isession.ISession; +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.utils.BitMap; +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.Assume; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +@RunWith(IoTDBTestRunner.class) +@Category({LocalStandaloneIT.class}) +public class IoTDBAlignedTVListBitMapPerformanceIT { + + private static final Logger LOGGER = + LoggerFactory.getLogger(IoTDBAlignedTVListBitMapPerformanceIT.class); + + private static final String ENABLED_PROPERTY = "iotdb.aligned.bitmap.perf.enabled"; + private static final String MEASUREMENT_COUNT_PROPERTY = + "iotdb.aligned.bitmap.perf.measurement.count"; + private static final String WARMUP_BATCH_COUNT_PROPERTY = + "iotdb.aligned.bitmap.perf.warmup.batch.count"; + private static final String BATCH_COUNT_PROPERTY = "iotdb.aligned.bitmap.perf.batch.count"; + private static final String ROUND_COUNT_PROPERTY = "iotdb.aligned.bitmap.perf.round.count"; + + private static final int ROWS_PER_BATCH = 64; + private static final int MEASUREMENT_COUNT = Integer.getInteger(MEASUREMENT_COUNT_PROPERTY, 256); + private static final int WARMUP_BATCH_COUNT = Integer.getInteger(WARMUP_BATCH_COUNT_PROPERTY, 50); + private static final int BATCH_COUNT = Integer.getInteger(BATCH_COUNT_PROPERTY, 100); + private static final int ROUND_COUNT = Integer.getInteger(ROUND_COUNT_PROPERTY, 5); + private static final String DEVICE = "root.aligned_bitmap_performance.d1"; + + @BeforeClass + public static void setUp() throws Exception { + Assume.assumeTrue(Boolean.getBoolean(ENABLED_PROPERTY)); + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setAutoCreateSchemaEnabled(true) + .setPrimitiveArraySize(ROWS_PER_BATCH) + .setMemtableSizeThreshold(512L * 1024 * 1024); + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() throws Exception { + EnvFactory.getEnv().cleanClusterEnvironment(); + } + + @Test + public void testNullHeavyAlignedTabletWritePerformance() throws Exception { + Assert.assertTrue(MEASUREMENT_COUNT > 0); + Assert.assertTrue(WARMUP_BATCH_COUNT >= 0); + Assert.assertTrue(BATCH_COUNT > 0); + Assert.assertTrue(ROUND_COUNT > 0); + + Tablet tablet = createTablet(); + long nextTimestamp = 0; + try (ISession session = EnvFactory.getEnv().getSessionConnection()) { + nextTimestamp = insertBatches(session, tablet, nextTimestamp, WARMUP_BATCH_COUNT); + + long[] elapsedNanos = new long[ROUND_COUNT]; + for (int round = 0; round < ROUND_COUNT; round++) { + long startTime = System.nanoTime(); + nextTimestamp = insertBatches(session, tablet, nextTimestamp, BATCH_COUNT); + elapsedNanos[round] = System.nanoTime() - startTime; + } + + long expectedBatchCount = WARMUP_BATCH_COUNT + (long) BATCH_COUNT * ROUND_COUNT; + assertFirstMeasurementCount(session, expectedBatchCount * (ROWS_PER_BATCH - 1L)); + Arrays.sort(elapsedNanos); + double medianNanos = elapsedNanos[elapsedNanos.length / 2]; + double pointsPerSecond = + (double) BATCH_COUNT * ROWS_PER_BATCH * MEASUREMENT_COUNT * 1_000_000_000L / medianNanos; + LOGGER.info( + "AlignedTVList bitmap write benchmark: measurements={}, rows/batch={}, warmup batches={}, " + + "batches/round={}, rounds={}, median={} ms/round, throughput={} points/s", + MEASUREMENT_COUNT, + ROWS_PER_BATCH, + WARMUP_BATCH_COUNT, + BATCH_COUNT, + ROUND_COUNT, + String.format("%.3f", medianNanos / 1_000_000.0), + String.format("%.0f", pointsPerSecond)); + } + } + + private static Tablet createTablet() { + List<IMeasurementSchema> schemas = new ArrayList<>(MEASUREMENT_COUNT); + for (int i = 0; i < MEASUREMENT_COUNT; i++) { + schemas.add(new MeasurementSchema("s" + i, TSDataType.INT64)); + } + Tablet tablet = new Tablet(DEVICE, schemas, ROWS_PER_BATCH); + tablet.initBitMaps(); + Object[] values = tablet.getValues(); + BitMap[] bitMaps = tablet.getBitMaps(); + for (int column = 0; column < MEASUREMENT_COUNT; column++) { + long[] columnValues = (long[]) values[column]; + Arrays.fill(columnValues, column); + bitMaps[column] = new BitMap(ROWS_PER_BATCH); + bitMaps[column].mark(column % ROWS_PER_BATCH); + } + tablet.setRowSize(ROWS_PER_BATCH); + return tablet; + } + + private static long insertBatches( + ISession session, Tablet tablet, long nextTimestamp, int batchCount) throws Exception { + long[] timestamps = tablet.getTimestamps(); + for (int batch = 0; batch < batchCount; batch++) { + for (int row = 0; row < ROWS_PER_BATCH; row++) { + timestamps[row] = nextTimestamp++; + } + session.insertAlignedTablet(tablet); + } + return nextTimestamp; + } + + private static void assertFirstMeasurementCount(ISession session, long expectedCount) + throws Exception { + try (org.apache.iotdb.isession.SessionDataSet dataSet = + session.executeQueryStatement("SELECT COUNT(s0) FROM " + DEVICE)) { + Assert.assertTrue(dataSet.hasNext()); + Assert.assertEquals(expectedCount, dataSet.next().getFields().get(0).getLongV()); + Assert.assertFalse(dataSet.hasNext()); + } + } +} 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 0e44a3dbde5..464cbe09995 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 @@ -46,7 +46,6 @@ import org.apache.tsfile.read.filter.basic.Filter; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.BitMap; import org.apache.tsfile.utils.Pair; -import org.apache.tsfile.utils.RamUsageEstimator; import org.apache.tsfile.utils.ReadWriteForEncodingUtils; import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.utils.TsPrimitiveType; @@ -74,9 +73,7 @@ import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_OBJECT_REF; public abstract class AlignedTVList extends TVList { private static final long BITMAP_RAM_COST_PER_BLOCK = - RamUsageEstimator.shallowSizeOfInstance(BitMap.class) - + RamUsageEstimator.sizeOfByteArray(BitMap.getSizeOfBytes(ARRAY_SIZE)) - + NUM_BYTES_OBJECT_REF; + BitMap.createBitMapDynamically(ARRAY_SIZE).ramBytesUsed() + NUM_BYTES_OBJECT_REF; // Data types of this aligned tvList protected List<TSDataType> dataTypes; @@ -426,7 +423,7 @@ public abstract class AlignedTVList extends TVList { default: break; } - BitMap bitMap = new BitMap(ARRAY_SIZE); + 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. @@ -734,13 +731,13 @@ public abstract class AlignedTVList extends TVList { if (bitMaps.get(columnIndex) == null) { List<BitMap> columnBitMaps = new ArrayList<>(values.get(columnIndex).size()); for (int i = 0; i < values.get(columnIndex).size(); i++) { - columnBitMaps.add(new BitMap(ARRAY_SIZE)); + columnBitMaps.add(BitMap.createBitMapDynamically(ARRAY_SIZE)); } bitMaps.set(columnIndex, columnBitMaps); } for (int i = 0; i < bitMaps.get(columnIndex).size(); i++) { if (bitMaps.get(columnIndex).get(i) == null) { - bitMaps.get(columnIndex).set(i, new BitMap(ARRAY_SIZE)); + bitMaps.get(columnIndex).set(i, BitMap.createBitMapDynamically(ARRAY_SIZE)); } bitMaps.get(columnIndex).get(i).markAll(); } @@ -832,12 +829,9 @@ public abstract class AlignedTVList extends TVList { */ private boolean markRowNull(int i) { if (timeColDeletedMap == null) { - timeColDeletedMap = new BitMap(rowCount); + timeColDeletedMap = BitMap.createBitMapDynamically(rowCount); } else if (timeColDeletedMap.getSize() < rowCount) { - byte[] prevBytes = timeColDeletedMap.getByteArray(); - byte[] newBytes = new byte[rowCount / 8 + 1]; - System.arraycopy(prevBytes, 0, newBytes, 0, prevBytes.length); - timeColDeletedMap = new BitMap(rowCount, newBytes); + timeColDeletedMap.extend(rowCount); } // use value index so that sorts will not change the nullability if (timeColDeletedMap.isMarked(getValueIndex(i))) { @@ -930,9 +924,9 @@ public abstract class AlignedTVList extends TVList { int arrayIndex) { /* 1. Build result-level bitmap (1 = failure row) */ - byte[] resultBitMap = + BitMap resultBitMap = results != null && containsFailedStatus(results, idx, len) - ? buildResultBitMapBytes(results, idx, elementIdx, len) + ? buildResultBitMap(results, idx, elementIdx, len) : null; for (int j = 0; j < values.length; j++) { @@ -949,7 +943,7 @@ public abstract class AlignedTVList extends TVList { /* 3. Overlay result bitmap (failure rows) */ if (resultBitMap != null) { - markNullValue(j, arrayIndex, elementIdx, resultBitMap); + getBitMap(j, arrayIndex).merge(resultBitMap, elementIdx & 7, elementIdx, len); } } } @@ -990,13 +984,18 @@ public abstract class AlignedTVList extends TVList { public static byte[] buildResultBitMapBytes( TSStatus[] results, int idx, int elementIdx, int length) { + int totalBits = (elementIdx & 7) + length; + return Arrays.copyOf( + buildResultBitMap(results, idx, elementIdx, length).getByteArray(), (totalBits + 7) >> 3); + } + + private static BitMap buildResultBitMap(TSStatus[] results, int idx, int elementIdx, int length) { int start = elementIdx & 7; int totalBits = start + length; - int size = (totalBits + 7) >> 3; - BitMap bitmap = new BitMap(totalBits, new byte[size]); + BitMap bitmap = BitMap.createBitMapDynamically(totalBits); if (results == null) { - return bitmap.getByteArray(); + return bitmap; } for (int i = 0; i < length; i++) { @@ -1005,7 +1004,7 @@ public abstract class AlignedTVList extends TVList { bitmap.mark(start + i); } } - return bitmap.getByteArray(); + return bitmap; } private void arrayCopy(Object[] value, int idx, int arrayIndex, int elementIndex, int remaining) { @@ -1077,21 +1076,12 @@ public abstract class AlignedTVList extends TVList { // if the bitmap in arrayIndex is null, init the bitmap if (bitMaps.get(columnIndex).get(arrayIndex) == null) { - bitMaps.get(columnIndex).set(arrayIndex, new BitMap(ARRAY_SIZE)); + bitMaps.get(columnIndex).set(arrayIndex, BitMap.createBitMapDynamically(ARRAY_SIZE)); } return bitMaps.get(columnIndex).get(arrayIndex); } - private void markNullValue( - int columnIndex, int arrayIndex, int elementIndex, byte[] resultBitMap) { - byte[] bitMap = getBitMap(columnIndex, arrayIndex).getByteArray(); - int start = elementIndex >>> 3; - for (byte b : resultBitMap) { - bitMap[start++] |= b; - } - } - private boolean markNullValue(int columnIndex, int arrayIndex, int elementIndex) { // mark the null value in the current bitmap BitMap bitMap = getBitMap(columnIndex, arrayIndex); @@ -1548,7 +1538,7 @@ public abstract class AlignedTVList extends TVList { Object[] values = new Object[dataTypeNum]; BitMap[] bitMaps = new BitMap[dataTypeNum]; for (int columnIndex = 0; columnIndex < dataTypeNum; ++columnIndex) { - BitMap bitMap = new BitMap(rowCount); + BitMap bitMap = BitMap.createBitMapDynamically(rowCount); Object valuesOfOneColumn; switch (dataTypes.get(columnIndex)) { case TEXT: @@ -1702,8 +1692,9 @@ public abstract class AlignedTVList extends TVList { } byte bits = (byte) 0X00; + byte[] bitMapBytes = bitMap.getByteArray(); for (int j = 0; j < size; j++) { - rowBitsArr[index] &= bitMap.getByteArray()[j]; + rowBitsArr[index] &= bitMapBytes[j]; bits |= rowBitsArr[index++]; isEnd = false; } @@ -1946,7 +1937,8 @@ public abstract class AlignedTVList extends TVList { valueColumnsDeletionList.get(columnIndex), valueColumnDeleteCursor.get(columnIndex), scanOrder))) { - bitMap = bitMap == null ? new BitMap(dataTypeList.size()) : bitMap; + bitMap = + bitMap == null ? BitMap.createBitMapDynamically(dataTypeList.size()) : bitMap; bitMap.mark(columnIndex); } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LazyBitMap.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LazyBitMap.java index ab35067270d..f977951ecb4 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LazyBitMap.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/LazyBitMap.java @@ -55,7 +55,7 @@ public class LazyBitMap { ensureCapacity(blockIndex); BitMap block = blocks.get(blockIndex); if (block == null) { - block = new BitMap(blockSize); + block = BitMap.createBitMapDynamically(blockSize); blocks.set(blockIndex, block); } block.mark(getInnerIndex(index)); 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 5ab976c1310..c6ccd333942 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 @@ -167,10 +167,41 @@ public class AlignedTVListTest { Assert.assertNotNull(firstColumnBitMaps.get(2)); Assert.assertEquals( BitMap.getSizeOfBytes(ARRAY_SIZE), firstColumnBitMaps.get(2).getByteArray().length); + Assert.assertTrue( + firstColumnBitMaps.get(2).ramBytesUsed() < new BitMap(ARRAY_SIZE).ramBytesUsed()); Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE * 2 + 1, 0)); Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE * 2, 0)); } + @Test + public void testFailedStatusBitmapWithNonByteAlignedOffset() { + AlignedTVList tvList = + AlignedTVList.newAlignedList(Arrays.asList(TSDataType.INT64, TSDataType.INT64)); + long[] times = new long[ARRAY_SIZE + 3]; + long[][] values = new long[2][ARRAY_SIZE + 3]; + TSStatus[] results = new TSStatus[ARRAY_SIZE + 3]; + for (int i = 0; i < times.length; i++) { + times[i] = i; + values[0][i] = i; + values[1][i] = i; + } + results[ARRAY_SIZE - 1] = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); + results[ARRAY_SIZE + 1] = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); + + tvList.putAlignedValues(times, values, null, 0, 3, results); + tvList.putAlignedValues(times, values, null, 3, times.length, results); + + for (int column = 0; column < 2; column++) { + Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE - 2, column)); + Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE - 1, column)); + Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE, column)); + Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE + 1, column)); + Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE + 2, column)); + } + + Assert.assertEquals(1, AlignedTVList.buildResultBitMapBytes(new TSStatus[8], 0, 0, 8).length); + } + @Test public void testEmptyInputBitmapsDoNotMaterializeMemTableBitmaps() { AlignedTVList tvList = AlignedTVList.newAlignedList(List.of(TSDataType.INT64)); diff --git a/pom.xml b/pom.xml index 4b0f44bbad3..79df3c2ebb1 100644 --- a/pom.xml +++ b/pom.xml @@ -145,7 +145,7 @@ <thrift.version>0.23.0</thrift.version> <xz.version>1.9</xz.version> <zstd-jni.version>1.5.6-3</zstd-jni.version> - <tsfile.version>2.3.2-260616-SNAPSHOT</tsfile.version> + <tsfile.version>2.3.2-260721-SNAPSHOT</tsfile.version> <i18n.locale>en</i18n.locale> <tsfile.locale.opt/> </properties>
