This is an automated email from the ASF dual-hosted git repository. jt2594838 pushed a commit to branch add_flat_device in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 80f4a5bc3cc43a30e97edb718db961f0b342e1cc Author: Tian Jiang <[email protected]> AuthorDate: Sat Oct 10 19:02:15 2026 +0800 tmp --- .../operator/source/FlatDevicePageReaderQueue.java | 1 - .../dataregion/memtable/AbstractMemTable.java | 135 +++++++++++++++------ .../memtable/AlignedWritableMemChunk.java | 39 +++++- .../db/utils/datastructure/AlignedTVList.java | 5 + .../datastructure/FlatDeviceAlignedTVList.java | 110 ++++++++++++----- .../db/utils/datastructure/AlignedTVListTest.java | 97 +++++++++++++++ .../datastructure/FlatDeviceAlignedTVListTest.java | 126 +++++++++++++++++++ 7 files changed, 437 insertions(+), 76 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java index 4a8166544d6..0ad96dabc48 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FlatDevicePageReaderQueue.java @@ -128,7 +128,6 @@ final class FlatDevicePageReaderQueue { private class ChunkPages { private final AbstractAlignedChunkMetadata metadata; private final Filter deletionFilter; - private List<IPageReader> readers; private ChunkPages(AbstractAlignedChunkMetadata metadata, Filter deletionFilter) { this.metadata = metadata; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java index 0919a70906a..95deb8b0902 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractMemTable.java @@ -38,6 +38,9 @@ import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; import org.apache.iotdb.db.schemaengine.schemaregion.utils.ResourceByPathUtils; +import org.apache.iotdb.db.service.metrics.WritingMetrics; +import org.apache.iotdb.db.service.metrics.WritingMetrics.FlatDeviceWriteCostTracker; +import org.apache.iotdb.db.service.metrics.WritingMetrics.FlatDeviceWriteStage; import org.apache.iotdb.db.storageengine.dataregion.flush.FlushStatus; import org.apache.iotdb.db.storageengine.dataregion.flush.NotifyFlushMemTable; import org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry; @@ -186,12 +189,13 @@ public abstract class AbstractMemTable implements IMemTable { List<IMeasurementSchema> schemaList, Supplier<List<Integer>> sortColumnIndicesSupplier, boolean useColumnSortedList) { - List<IMeasurementSchema> filteredSchemaList = - schemaList.stream().filter(Objects::nonNull).collect(Collectors.toList()); IWritableMemChunkGroup memChunkGroup = memTableMap.computeIfAbsent( deviceId, k -> { + // Existing groups do not need a schema copy for each logical device write. + List<IMeasurementSchema> filteredSchemaList = + schemaList.stream().filter(Objects::nonNull).collect(Collectors.toList()); seriesNumber += filteredSchemaList.size(); // Evaluate the supplier only for a newly created group. return new AlignedWritableMemChunkGroup( @@ -360,27 +364,49 @@ public abstract class AbstractMemTable implements IMemTable { public int insertAlignedTablet( InsertTabletNode insertTabletNode, int start, int end, TSStatus[] results) throws WriteProcessException { + final FlatDeviceWriteCostTracker tracker = + insertTabletNode.isFlatDevice() + ? WritingMetrics.getInstance().getFlatDeviceWriteCostTracker() + : null; + long stageStart; try { - int nullPointsNumber = - computeTabletNullPointsNumber( - insertTabletNode, start, end, !insertTabletNode.isFlatDevice(), results); + int nullPointsNumber; + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + nullPointsNumber = + computeTabletNullPointsNumber( + insertTabletNode, start, end, !insertTabletNode.isFlatDevice(), results); + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_NULL_COUNT, System.nanoTime() - stageStart); + } + } writeAlignedTablet(insertTabletNode, start, end, results); - // TODO-Table: what is the relation between this and TsFileProcessor.checkMemCost - memSize += MemUtils.getAlignedTabletSize(insertTabletNode, start, end, results); - int validValueCount = - insertTabletNode.getValidMemTableMeasurementNumber() - * computeSuccessfulRowCount(results, start, end); - int pointsInserted = - validValueCount - - (IoTDBDescriptor.getInstance() - .getConfig() - .isIncludeNullValueInWriteThroughputMetric() - ? 0 - : nullPointsNumber); - totalPointsNum += pointsInserted; - totalValueCount += validValueCount; - nullValueCount += nullPointsNumber; - return pointsInserted; + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + // TODO-Table: what is the relation between this and TsFileProcessor.checkMemCost + memSize += MemUtils.getAlignedTabletSize(insertTabletNode, start, end, results); + int validValueCount = + insertTabletNode.getValidMemTableMeasurementNumber() + * computeSuccessfulRowCount(results, start, end); + int pointsInserted = + validValueCount + - (IoTDBDescriptor.getInstance() + .getConfig() + .isIncludeNullValueInWriteThroughputMetric() + ? 0 + : nullPointsNumber); + totalPointsNum += pointsInserted; + totalValueCount += validValueCount; + nullValueCount += nullPointsNumber; + return pointsInserted; + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_STATISTICS, System.nanoTime() - stageStart); + } + } } catch (RuntimeException e) { throw new WriteProcessException(e); } @@ -501,27 +527,39 @@ public abstract class AbstractMemTable implements IMemTable { public void writeAlignedTablet( InsertTabletNode insertTabletNode, int start, int end, TSStatus[] results) { + final FlatDeviceWriteCostTracker tracker = + insertTabletNode.isFlatDevice() + ? WritingMetrics.getInstance().getFlatDeviceWriteCostTracker() + : null; + long stageStart = tracker == null ? 0 : System.nanoTime(); List<IMeasurementSchema> schemaList = new ArrayList<>(); final int measurementSize = insertTabletNode.getMeasurements() == null ? 0 : insertTabletNode.getMeasurements().length; final Object[] writableColumns = new Object[measurementSize]; final BitMap[] writableBitMaps = insertTabletNode.getBitMaps() == null ? null : new BitMap[measurementSize]; - for (int i = 0; i < measurementSize; i++) { - if (!insertTabletNode.isValidMemTableMeasurement(i) - || insertTabletNode.getColumns() == null - || i >= insertTabletNode.getColumns().length - || insertTabletNode.getColumns()[i] == null) { - schemaList.add(null); - } else { - schemaList.add(getMemTableMeasurementSchema(insertTabletNode, i)); - writableColumns[i] = insertTabletNode.getColumns()[i]; - if (writableBitMaps != null - && insertTabletNode.getBitMaps() != null - && i < insertTabletNode.getBitMaps().length) { - writableBitMaps[i] = insertTabletNode.getBitMaps()[i]; + try { + for (int i = 0; i < measurementSize; i++) { + if (!insertTabletNode.isValidMemTableMeasurement(i) + || insertTabletNode.getColumns() == null + || i >= insertTabletNode.getColumns().length + || insertTabletNode.getColumns()[i] == null) { + schemaList.add(null); + } else { + schemaList.add(getMemTableMeasurementSchema(insertTabletNode, i)); + writableColumns[i] = insertTabletNode.getColumns()[i]; + if (writableBitMaps != null + && insertTabletNode.getBitMaps() != null + && i < insertTabletNode.getBitMaps().length) { + writableBitMaps[i] = insertTabletNode.getBitMaps()[i]; + } } } + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_SCHEMA_PREPARE, System.nanoTime() - stageStart); + } } if (schemaList.isEmpty()) { return; @@ -530,15 +568,32 @@ public abstract class AbstractMemTable implements IMemTable { if (insertTabletNode.isFlatDevice()) { sortColumnIndicesSupplier = () -> getTagColumnIndices(insertTabletNode, schemaList); } - final List<Pair<IDeviceID, Integer>> deviceEndOffsetPair = - insertTabletNode.splitByDevice(start, end); + final List<Pair<IDeviceID, Integer>> deviceEndOffsetPair; + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + deviceEndOffsetPair = insertTabletNode.splitByDevice(start, end); + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_DEVICE_SPLIT, System.nanoTime() - stageStart); + } + } int splitStart = start; for (Pair<IDeviceID, Integer> pair : deviceEndOffsetPair) { - final IDeviceID deviceID = insertTabletNode.getMemTableDeviceID(pair.left); int splitEnd = pair.right; - IWritableMemChunkGroup memChunkGroup = - createAlignedMemChunkGroupIfNotExistAndGet( - deviceID, schemaList, sortColumnIndicesSupplier, insertTabletNode.isFlatDevice()); + IWritableMemChunkGroup memChunkGroup; + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + final IDeviceID deviceID = insertTabletNode.getMemTableDeviceID(pair.left); + memChunkGroup = + createAlignedMemChunkGroupIfNotExistAndGet( + deviceID, schemaList, sortColumnIndicesSupplier, insertTabletNode.isFlatDevice()); + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_CHUNK_LOOKUP, System.nanoTime() - stageStart); + } + } if (memChunkGroup instanceof AlignedWritableMemChunkGroup alignedMemChunkGroup) { alignedMemChunkGroup.writeTablet( insertTabletNode.getTimes(), diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java index 0eee722067d..50c6698973c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedWritableMemChunk.java @@ -27,6 +27,9 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.DataTypeInconsistentException; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.service.metrics.WritingMetrics; +import org.apache.iotdb.db.service.metrics.WritingMetrics.FlatDeviceWriteCostTracker; +import org.apache.iotdb.db.service.metrics.WritingMetrics.FlatDeviceWriteStage; import org.apache.iotdb.db.storageengine.dataregion.modification.FlatDeviceDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; import org.apache.iotdb.db.storageengine.dataregion.tsfile.FlatDeviceAlignedChunkWriter; @@ -411,13 +414,41 @@ public class AlignedWritableMemChunk extends AbstractWritableMemChunk { int start, int end, TSStatus[] results) { - Pair<Object[], BitMap[]> pair = - checkAndReorderColumnValuesInInsertPlan(schemaList, valueList, bitMaps, columnCategories); + final FlatDeviceWriteCostTracker tracker = + useColumnSortedList ? WritingMetrics.getInstance().getFlatDeviceWriteCostTracker() : null; + long stageStart; + Pair<Object[], BitMap[]> pair; + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + pair = + checkAndReorderColumnValuesInInsertPlan(schemaList, valueList, bitMaps, columnCategories); + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_COLUMN_REORDER, System.nanoTime() - stageStart); + } + } Object[] reorderedColumnValues = pair.left; BitMap[] reorderedBitMaps = pair.right; - putAlignedTablet(times, reorderedColumnValues, reorderedBitMaps, start, end, results); + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + putAlignedTablet(times, reorderedColumnValues, reorderedBitMaps, start, end, results); + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_TVLIST_APPEND, System.nanoTime() - stageStart); + } + } if (TVLIST_SORT_THRESHOLD > 0 && list.rowCount() >= TVLIST_SORT_THRESHOLD) { - handoverAlignedTvList(); + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + handoverAlignedTvList(); + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_TVLIST_HANDOVER, System.nanoTime() - stageStart); + } + } } } 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 ff26cce7e9a..96349e2535b 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 @@ -1339,6 +1339,11 @@ public abstract class AlignedTVList extends TVList { @Override public synchronized void putAlignedValues( long[] time, Object[] value, BitMap[] bitMaps, int start, int end, TSStatus[] results) { + // Row statuses apply to every column. Resolve the all-success case once per input range + // so lazy column allocation does not repeatedly scan the same statuses. + if (results != null && !containsFailedStatus(results, start, end - start)) { + results = null; + } checkExpansion(); int idx = start; diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java index 9bb277fa90d..186c42fc56e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVList.java @@ -20,6 +20,9 @@ package org.apache.iotdb.db.utils.datastructure; import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.db.i18n.DataNodeMiscMessages; +import org.apache.iotdb.db.service.metrics.WritingMetrics; +import org.apache.iotdb.db.service.metrics.WritingMetrics.FlatDeviceWriteCostTracker; +import org.apache.iotdb.db.service.metrics.WritingMetrics.FlatDeviceWriteStage; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALWriteUtils; import org.apache.iotdb.rpc.TSStatusCode; @@ -64,6 +67,9 @@ public class FlatDeviceAlignedTVList extends AlignedTVList { private final List<Integer> sortColumns = new ArrayList<>(); private Map<IDeviceID, DeviceInfo> deviceInfoMap = new TreeMap<>(FlatDeviceAlignedTVList::compareDevices); + // Includes TreeMap entries and DeviceInfo payloads. Device keys/TAGs are immutable; only + // row-index capacity grows, so writes need not revisit every previously inserted device. + private long deviceMetadataRamCost; private AlignedTVList fields; private int[] fieldColumns; private int[] tagOrdinals; @@ -233,33 +239,60 @@ public class FlatDeviceAlignedTVList extends AlignedTVList { if (start >= end) { return; } - for (int row = start; row < end; row++) { - boolean failed = - results != null - && results[row] != null - && results[row].code != TSStatusCode.SUCCESS_STATUS.getStatusCode(); - for (int tag = 0; tag < tagColumns.size(); tag++) { - int column = tagColumns.get(tag); - lookupTags[tag] = - failed - || input[column] == null - || (nulls != null && nulls[column] != null && nulls[column].isMarked(row)) - ? null - : ((Binary[]) input[column])[row]; + final FlatDeviceWriteCostTracker tracker = + WritingMetrics.getInstance().getFlatDeviceWriteCostTracker(); + long stageStart; + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + for (int row = start; row < end; row++) { + boolean failed = + results != null + && results[row] != null + && results[row].code != TSStatusCode.SUCCESS_STATUS.getStatusCode(); + for (int tag = 0; tag < tagColumns.size(); tag++) { + int column = tagColumns.get(tag); + lookupTags[tag] = + failed + || input[column] == null + || (nulls != null && nulls[column] != null && nulls[column].isMarked(row)) + ? null + : ((Binary[]) input[column])[row]; + } + appendDevice(rowCount + row - start, times[row]); + } + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_DEVICE_INDEX, System.nanoTime() - stageStart); } - appendDevice(rowCount + row - start, times[row]); } - for (int column = 0; column < fieldColumns.length; column++) { - int field = fieldColumns[column]; - if (field >= 0) { - fieldInput[field] = input[column]; - fieldBitMaps[field] = nulls == null ? null : nulls[column]; + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + for (int column = 0; column < fieldColumns.length; column++) { + int field = fieldColumns[column]; + if (field >= 0) { + fieldInput[field] = input[column]; + fieldBitMaps[field] = nulls == null ? null : nulls[column]; + } + } + fields.putAlignedValues(times, fieldInput, fieldBitMaps, start, end, results); + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_FIELD_APPEND, System.nanoTime() - stageStart); + } + } + stageStart = tracker == null ? 0 : System.nanoTime(); + try { + Arrays.fill(fieldInput, null); + Arrays.fill(fieldBitMaps, null); + finishAppend(); + } finally { + if (tracker != null) { + tracker.recordCost( + FlatDeviceWriteStage.MEMTABLE_APPEND_FINISH, System.nanoTime() - stageStart); } } - fields.putAlignedValues(times, fieldInput, fieldBitMaps, start, end, results); - Arrays.fill(fieldInput, null); - Arrays.fill(fieldBitMaps, null); - finishAppend(); } private void finishAppend() { @@ -299,9 +332,14 @@ public class FlatDeviceAlignedTVList extends AlignedTVList { : lookupTags[i].getStringValue(TSFileConfig.STRING_CHARSET); } IDeviceID id = IDeviceID.Factory.DEFAULT_FACTORY.create(segments); - info = deviceInfoMap.computeIfAbsent(id, DeviceInfo::new); + info = deviceInfoMap.get(id); + if (info == null) { + info = new DeviceInfo(id); + deviceInfoMap.put(id, info); + deviceMetadataRamCost += 40 + info.ramBytesUsed(); + } } - info.addRow(row, time); + deviceMetadataRamCost += info.addRow(row, time); lastLookup = info; Arrays.fill(lookupTags, null); } @@ -679,6 +717,8 @@ public class FlatDeviceAlignedTVList extends AlignedTVList { for (DeviceInfo info : deviceInfoMap.values()) { copy.deviceInfoMap.put(info.id, info.copy()); } + // Copies preserve device keys and row-index capacities, including partial/flush snapshots. + copy.deviceMetadataRamCost = deviceMetadataRamCost; copy.valuePositions = valuePositions.clone(); for (DeviceInfo info : devicesInOrder) { copy.devicesInOrder.add(copy.deviceInfoMap.get(info.id)); @@ -798,6 +838,7 @@ public class FlatDeviceAlignedTVList extends AlignedTVList { protected void clearValue() { fields.clear(); deviceInfoMap = new TreeMap<>(FlatDeviceAlignedTVList::compareDevices); + deviceMetadataRamCost = 0; valuePositions = new int[0]; devicesInOrder = new ArrayList<>(); lastLookup = null; @@ -828,7 +869,10 @@ public class FlatDeviceAlignedTVList extends AlignedTVList { @Override public synchronized long getRamSize(Set<Integer> columns) { long size = - fields.getRamSize(fieldSet(columns)) + // The no-argument FIELD path uses incrementally maintained value/bitmap costs. + // Keep selected-column accounting for query ownership transfers. + (columns == null ? fields.getRamSize() : fields.getRamSize(fieldSet(columns))) + + deviceMetadataRamCost + RamUsageEstimator.shallowSizeOf(this) + RamUsageEstimator.shallowSizeOf(deviceInfoMap) + RamUsageEstimator.sizeOfIntArray(valuePositions.length) @@ -844,9 +888,6 @@ public class FlatDeviceAlignedTVList extends AlignedTVList { if (timeColDeletedMap != null) { size += timeColDeletedMap.ramBytesUsed(); } - for (DeviceInfo info : deviceInfoMap.values()) { - size += 40 + info.ramBytesUsed(); - } return size; } @@ -1053,13 +1094,20 @@ public class FlatDeviceAlignedTVList extends AlignedTVList { return true; } - void addRow(int row, long time) { + // Return only newly retained bytes; unchanged row-index capacity has no accounting work. + long addRow(int row, long time) { + long memoryIncrement = 0; if (rowCount == rowIndices.length) { - rowIndices = Arrays.copyOf(rowIndices, rowIndices.length * 2); + int previousCapacity = rowIndices.length; + rowIndices = Arrays.copyOf(rowIndices, previousCapacity * 2); + memoryIncrement = + RamUsageEstimator.sizeOfIntArray(rowIndices.length) + - RamUsageEstimator.sizeOfIntArray(previousCapacity); } rowIndices[rowCount++] = row; startTime = Math.min(startTime, time); endTime = Math.max(endTime, time); + return memoryIncrement; } boolean containsUnsortedRow(int row) { 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 df80d0913b6..b826048c3aa 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 @@ -257,6 +257,103 @@ public class AlignedTVListTest { Assert.assertNull(tvList.getBitMaps()); } + @Test + public void testSuccessfulTabletSubrangeWithFailuresOutsideRange() { + // Failures outside the selected range must not affect successful rows across array boundaries. + // Null and explicit success statuses are equivalent, and the caller's array must stay intact. + AlignedTVList tvList = + AlignedTVListFactory.createAlignedList(List.of(TSDataType.INT64, TSDataType.INT64)); + int inputSize = ARRAY_SIZE + 9; + int start = 3; + int end = inputSize - 2; + long[] times = new long[inputSize]; + long[][] values = new long[2][inputSize]; + TSStatus[] results = new TSStatus[inputSize]; + for (int rowIndex = 0; rowIndex < inputSize; rowIndex++) { + times[rowIndex] = rowIndex; + values[0][rowIndex] = rowIndex; + values[1][rowIndex] = rowIndex * 10L; + if ((rowIndex & 1) != 0) { + results[rowIndex] = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + } + results[start - 1] = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); + results[end] = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); + TSStatus[] originalResults = results.clone(); + + tvList.putAlignedValues( + times, values, new BitMap[] {new BitMap(inputSize), null}, start, end, results); + + Assert.assertEquals(end - start, tvList.rowCount()); + Assert.assertNull(tvList.getBitMaps()); + Assert.assertArrayEquals(originalResults, results); + for (int rowIndex = 0; rowIndex < end - start; rowIndex++) { + int inputRowIndex = start + rowIndex; + Assert.assertEquals(inputRowIndex, tvList.getTime(rowIndex)); + Assert.assertEquals(inputRowIndex, tvList.getLongByValueIndex(rowIndex, 0)); + Assert.assertEquals(inputRowIndex * 10L, tvList.getLongByValueIndex(rowIndex, 1)); + } + } + + @Test + public void testFailureAndNullBitmapUnionAcrossArrayBoundaries() { + // Start inside a populated block: failure statuses and input nulls must be merged at the + // correct offsets, while subsequent entirely null blocks retain lazy value allocation. + AlignedTVList tvList = + AlignedTVListFactory.createAlignedList( + List.of(TSDataType.INT64, TSDataType.INT64, TSDataType.INT64)); + tvList.putAlignedValue(-1, new Object[] {-1L, -1L, null}); + int inputSize = ARRAY_SIZE * 2 + 9; + int start = 3; + int end = inputSize - 2; + long[] times = new long[inputSize]; + long[] firstValues = new long[inputSize]; + long[] secondValues = new long[inputSize]; + BitMap[] bitMaps = {new BitMap(inputSize), new BitMap(inputSize), null}; + TSStatus[] results = new TSStatus[inputSize]; + for (int rowIndex = 0; rowIndex < inputSize; rowIndex++) { + times[rowIndex] = rowIndex; + firstValues[rowIndex] = rowIndex; + secondValues[rowIndex] = rowIndex * 10L; + if ((rowIndex & 1) == 0) { + results[rowIndex] = new TSStatus(TSStatusCode.INTERNAL_SERVER_ERROR.getStatusCode()); + } else { + bitMaps[0].mark(rowIndex); + if (rowIndex % 4 == 1) { + results[rowIndex] = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + } + if (rowIndex % 3 == 0) { + bitMaps[1].mark(rowIndex); + } + } + + tvList.putAlignedValues( + times, new Object[] {firstValues, secondValues, null}, bitMaps, start, end, results); + + Assert.assertEquals(1 + end - start, tvList.rowCount()); + Assert.assertFalse(tvList.isNullValue(0, 0)); + Assert.assertFalse(tvList.isNullValue(0, 1)); + Assert.assertEquals(-1L, tvList.getLongByValueIndex(0, 0)); + for (int inputRowIndex = start; inputRowIndex < end; inputRowIndex++) { + int rowIndex = 1 + inputRowIndex - start; + boolean secondValueIsNull = (inputRowIndex & 1) == 0 || inputRowIndex % 3 == 0; + Assert.assertEquals(inputRowIndex, tvList.getTime(rowIndex)); + Assert.assertTrue(tvList.isNullValue(rowIndex, 0)); + Assert.assertEquals(secondValueIsNull, tvList.isNullValue(rowIndex, 1)); + Assert.assertTrue(tvList.isNullValue(rowIndex, 2)); + if (!secondValueIsNull) { + Assert.assertEquals(inputRowIndex * 10L, tvList.getLongByValueIndex(rowIndex, 1)); + } + } + for (int arrayIndex = 1; arrayIndex < tvList.getValues().get(0).size(); arrayIndex++) { + Assert.assertNull(tvList.getValues().get(0).get(arrayIndex)); + } + for (Object valueArray : tvList.getValues().get(2)) { + Assert.assertNull(valueArray); + } + } + @Test public void testPrimitiveArraysAreAllocatedOnFirstWrite() { AlignedTVList tvList = diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVListTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVListTest.java index d9c8c12aadf..c65af55204a 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVListTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/utils/datastructure/FlatDeviceAlignedTVListTest.java @@ -36,8 +36,10 @@ import org.junit.Test; import java.io.ByteArrayInputStream; import java.io.DataInputStream; +import java.lang.reflect.Field; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; @@ -60,6 +62,130 @@ public class FlatDeviceAlignedTVListTest { return new Binary(text, TSFileConfig.STRING_CHARSET); } + // Compare incremental totals against the former full traversal, independently of write deltas. + private static void assertRamAccounting(FlatDeviceAlignedTVList list) throws Exception { + Field deviceCost = FlatDeviceAlignedTVList.class.getDeclaredField("deviceMetadataRamCost"); + deviceCost.setAccessible(true); + long expectedDeviceCost = 0; + for (FlatDeviceAlignedTVList.DeviceInfo info : list.getDeviceInfoMap().values()) { + expectedDeviceCost += 40 + info.ramBytesUsed(); + } + Assert.assertEquals(expectedDeviceCost, deviceCost.getLong(list)); + + // Selecting every column still scans FIELD arrays/bitmaps and is an oracle for the cached path. + Set<Integer> columns = new HashSet<>(); + for (int columnIndex = 0; columnIndex < list.getTsDataTypes().size(); columnIndex++) { + columns.add(columnIndex); + } + Assert.assertEquals(list.getRamSize(columns), list.getRamSize()); + Assert.assertEquals(list.getRamSize(), list.calculateRamSize().getRamSize()); + } + + @Test + public void testIncrementalRamAccountingAcrossDeviceGrowthAndSchemaChanges() throws Exception { + FlatDeviceAlignedTVList list = list(); + assertRamAccounting(list); + // Mix last-device cache hits, map lookups, null TAGs and repeated row-index doublings. + for (int rowIndex = 0; rowIndex < 257; rowIndex++) { + Binary tag = rowIndex % 3 == 0 ? binary("hot") : binary("device" + rowIndex % 7); + list.putAlignedValue( + 257 - rowIndex, new Object[] {tag, rowIndex % 4 == 0 ? null : (long) rowIndex, null}); + assertRamAccounting(list); + } + list.sort(); + assertRamAccounting(list); + list.addTagColumn(3, TSDataType.STRING); + list.putAlignedValue(300, new Object[] {binary("hot"), null, null, null}); + list.putAlignedValue(301, new Object[] {binary("hot"), 1L, null, binary("new")}); + list.insertColumn(1, TSDataType.DOUBLE); + list.putAlignedValue(302, new Object[] {binary("hot"), 2.0, 3L, null, null}); + assertRamAccounting(list); + list.delete(0, 257, 2); + list.deleteTime(300, 300); + assertRamAccounting(list); + } + + @Test + public void testSparseWideFieldRamAccountingAndOwnershipTransfer() throws Exception { + List<TSDataType> types = new ArrayList<>(); + types.add(TSDataType.STRING); + for (int columnIndex = 0; columnIndex < 500; columnIndex++) { + types.add(TSDataType.DOUBLE); + } + FlatDeviceAlignedTVList list = + (FlatDeviceAlignedTVList) + AlignedTVListFactory.createFlatDeviceAlignedList(types, List.of(0), PHYSICAL_DEVICE); + // Reproduce 100 populated FIELDs out of 500; bitmap accounting crosses multiple blocks. + for (int rowIndex = 0; rowIndex < 257; rowIndex++) { + Object[] row = new Object[501]; + row[0] = binary("device" + rowIndex % 3); + for (int columnIndex = 1; columnIndex <= 100; columnIndex++) { + row[100 * (rowIndex % 5) + columnIndex] = (double) rowIndex; + } + list.putAlignedValue(rowIndex, row); + } + assertRamAccounting(list); + for (Set<Integer> retained : List.of(Set.<Integer>of(), Set.of(1, 101), Set.of(0))) { + FlatDeviceAlignedTVList source = (FlatDeviceAlignedTVList) list.clone(); + source.setSortColumnIndices(List.of(201)); + source.sort(); + long reserved = source.getRamSize(retained); + AlignedTVList.PartialClonePlan plan = source.preparePartialClone(retained); + FlatDeviceAlignedTVList working = (FlatDeviceAlignedTVList) plan.getCloneList(); + plan.commit(); + Assert.assertEquals(reserved, source.getRamSize()); + assertRamAccounting(source); + assertRamAccounting(working); + source.clear(); + assertRamAccounting(source); + assertRamAccounting(working); + } + list.releaseNonQueryColumns(Set.of(1, 101)); + assertRamAccounting(list); + } + + @Test + public void testDeviceRamAccountingSurvivesSnapshotsWalAndReuse() throws Exception { + FlatDeviceAlignedTVList list = list(); + // A tablet, rather than individual inserts, must maintain the same device capacity accounting. + int rowCount = 129; + long[] times = new long[rowCount]; + long[] values = new long[rowCount]; + Binary[] tags = new Binary[rowCount]; + for (int rowIndex = 0; rowIndex < rowCount; rowIndex++) { + times[rowIndex] = rowCount - rowIndex; + tags[rowIndex] = rowIndex % 3 == 0 ? null : binary("hot"); + } + list.putAlignedValues(times, new Object[] {tags, values, null}, null, 0, rowCount, null); + assertRamAccounting(list); + list.sort(); + FlatDeviceAlignedTVList full = (FlatDeviceAlignedTVList) list.clone(); + FlatDeviceAlignedTVList flush = (FlatDeviceAlignedTVList) list.cloneForFlushSort(); + Assert.assertEquals(list.getRamSize(), full.getRamSize()); + Assert.assertEquals(list.getRamSize(), flush.getRamSize()); + assertRamAccounting(full); + assertRamAccounting(flush); + + ByteBuffer buffer = ByteBuffer.allocate(list.serializedSize()); + list.serializeToWAL(new WALByteBufferForTest(buffer)); + FlatDeviceAlignedTVList restored = + (FlatDeviceAlignedTVList) + AlignedTVListFactory.deserialize( + new DataInputStream(new ByteArrayInputStream(buffer.array())), PHYSICAL_DEVICE); + assertRamAccounting(restored); + long snapshotSize = full.getRamSize(); + for (int rowIndex = 0; rowIndex < 129; rowIndex++) { + restored.putAlignedValue(rowIndex, new Object[] {binary("hot"), 1L, null}); + } + restored.putAlignedValue(0, new Object[] {binary("new"), null, null}); + assertRamAccounting(restored); + Assert.assertEquals(snapshotSize, full.getRamSize()); + restored.clear(); + assertRamAccounting(restored); + restored.putAlignedValue(1, new Object[] {binary("reused"), 1L, binary("tag")}); + assertRamAccounting(restored); + } + @Test public void testDeviceOrderAndFieldThenTimeOrder() { FlatDeviceAlignedTVList list = list();
