You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/12 06:51:12 UTC
(iotdb) 37/38: tmp
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 <jt...@163.com>
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();