This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new 9e2074d26bd [to dev/1.3] Optimize aligned TVList partial-column clone
memory usage (#18454)
9e2074d26bd is described below
commit 9e2074d26bd67c7afe7de7e810e6ed7f2cf62cc2
Author: shuwenwei <[email protected]>
AuthorDate: Thu Aug 13 18:14:16 2026 +0800
[to dev/1.3] Optimize aligned TVList partial-column clone memory usage
(#18454)
---
.../execution/MemoryEstimationHelper.java | 2 +-
.../fragment/FragmentInstanceContext.java | 4 +-
.../db/utils/datastructure/AlignedTVList.java | 124 ++++++++----------
.../db/utils/datastructure/BackAlignedTVList.java | 6 +-
.../db/utils/datastructure/QuickAlignedTVList.java | 6 +-
.../iotdb/db/utils/datastructure/TVList.java | 5 -
.../db/utils/datastructure/TimAlignedTVList.java | 6 +-
.../fragment/FragmentInstanceExecutionTest.java | 145 +++++++++++++++++++++
.../dataregion/memtable/TsFileProcessorTest.java | 28 ++--
.../db/utils/datastructure/AlignedTVListTest.java | 43 ++++++
10 files changed, 272 insertions(+), 97 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/MemoryEstimationHelper.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/MemoryEstimationHelper.java
index ba6660f3d6a..1737b48cf18 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/MemoryEstimationHelper.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/MemoryEstimationHelper.java
@@ -44,7 +44,7 @@ public class MemoryEstimationHelper {
private static final long MEASUREMENT_PATH_INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(AlignedPath.class);
- private static final long ARRAY_LIST_INSTANCE_SIZE =
+ public static final long ARRAY_LIST_INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(ArrayList.class);
private static final long INTEGER_INSTANCE_SIZE =
RamUsageEstimator.shallowSizeOfInstance(Integer.class);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
index 24b233efcdf..14a90159dac 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java
@@ -165,7 +165,7 @@ public class FragmentInstanceContext extends QueryContext {
private boolean highestPriority = false;
// accessed value columns on each referenced AlignedTVList.
- private final Map<TVList, Set<Integer>> alignedTVListColumnAccessMap = new
ConcurrentHashMap<>();
+ private Map<TVList, Set<Integer>> alignedTVListColumnAccessMap = new
ConcurrentHashMap<>();
public static FragmentInstanceContext createFragmentInstanceContext(
FragmentInstanceId id,
@@ -1027,7 +1027,7 @@ public class FragmentInstanceContext extends QueryContext
{
// release TVList/AlignedTVList owned by current query
releaseTVListOwnedByQuery();
- alignedTVListColumnAccessMap.clear();
+ alignedTVListColumnAccessMap = null;
fileModCache = null;
nonExistentModFiles = null;
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 d8b16b3eb84..a3cb492c9a1 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
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.utils.datastructure;
+import org.apache.iotdb.db.queryengine.execution.MemoryEstimationHelper;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import org.apache.iotdb.db.queryengine.plan.statement.component.Ordering;
@@ -92,10 +93,9 @@ public abstract class AlignedTVList extends TVList {
public static final class PartialClonePlan {
private final AlignedTVList sourceList;
private final AlignedTVList cloneList;
- private final List<Object>[] valueColumnsToMove;
- private final List<BitMap>[] bitmapColumnsToMove;
- private final long sourceArrayMemCostWithoutIndex;
- private final long cloneArrayMemCostWithoutIndex;
+ // The columns retained by the source. commit() derives the moved columns
from this set, so
+ // no O(N) move-plan arrays need to be allocated during preparation.
+ private final Set<Integer> retainedColumns;
private final long sourceBitmapMemoryCost;
private final long cloneBitmapMemoryCost;
@@ -104,18 +104,12 @@ public abstract class AlignedTVList extends TVList {
private PartialClonePlan(
AlignedTVList sourceList,
AlignedTVList cloneList,
- List<Object>[] valueColumnsToMove,
- List<BitMap>[] bitmapColumnsToMove,
- long sourceArrayMemCostWithoutIndex,
- long cloneArrayMemCostWithoutIndex,
+ Set<Integer> retainedColumns,
long sourceBitmapMemoryCost,
long cloneBitmapMemoryCost) {
this.sourceList = sourceList;
this.cloneList = cloneList;
- this.valueColumnsToMove = valueColumnsToMove;
- this.bitmapColumnsToMove = bitmapColumnsToMove;
- this.sourceArrayMemCostWithoutIndex = sourceArrayMemCostWithoutIndex;
- this.cloneArrayMemCostWithoutIndex = cloneArrayMemCostWithoutIndex;
+ this.retainedColumns = retainedColumns;
this.sourceBitmapMemoryCost = sourceBitmapMemoryCost;
this.cloneBitmapMemoryCost = cloneBitmapMemoryCost;
}
@@ -149,13 +143,17 @@ public abstract class AlignedTVList extends TVList {
private final AlignedTVList outer = this;
AlignedTVList(List<TSDataType> types) {
+ this(types, true);
+ }
+
+ AlignedTVList(List<TSDataType> types, boolean initializeValueColumns) {
super();
dataTypes = types;
memoryBinaryChunkSize = new long[dataTypes.size()];
materializedValueArrayCounts = new int[dataTypes.size()];
values = new ArrayList<>(types.size());
for (int i = 0; i < types.size(); i++) {
- values.add(new ArrayList<>());
+ values.add(initializeValueColumns ? new ArrayList<>() : null);
}
// arrayMemCostWithoutIndex depends on per-column value arrays, so values
must be
// initialized before computing it
@@ -163,48 +161,19 @@ public abstract class AlignedTVList extends TVList {
}
public static AlignedTVList newAlignedList(List<TSDataType> dataTypes) {
+ return newAlignedList(dataTypes, true);
+ }
+
+ public static AlignedTVList newAlignedList(
+ List<TSDataType> dataTypes, boolean initializeValueColumns) {
switch (TVLIST_SORT_ALGORITHM) {
case QUICK:
- return new QuickAlignedTVList(dataTypes);
+ return new QuickAlignedTVList(dataTypes, initializeValueColumns);
case BACKWARD:
- return new BackAlignedTVList(dataTypes);
+ return new BackAlignedTVList(dataTypes, initializeValueColumns);
default:
- return new TimAlignedTVList(dataTypes);
- }
- }
-
- @Override
- public TVList getTvListByColumnIndex(List<Integer> columnIndex,
List<TSDataType> dataTypeList) {
- List<List<Object>> values = new ArrayList<>();
- List<List<BitMap>> bitMaps = null;
- for (int i = 0; i < columnIndex.size(); i++) {
- // columnIndex == -1 means querying a non-exist column, add null column
here
- if (columnIndex.get(i) == -1) {
- values.add(null);
- } else {
- values.add(this.values.get(columnIndex.get(i)));
- if (this.bitMaps != null && this.bitMaps.get(columnIndex.get(i)) !=
null) {
- if (bitMaps == null) {
- bitMaps = new ArrayList<>(columnIndex.size());
- for (int j = 0; j < columnIndex.size(); j++) {
- bitMaps.add(null);
- }
- }
- bitMaps.set(i, this.bitMaps.get(columnIndex.get(i)));
- }
- }
+ return new TimAlignedTVList(dataTypes, initializeValueColumns);
}
- AlignedTVList alignedTvList = AlignedTVList.newAlignedList(new
ArrayList<>(dataTypeList));
- alignedTvList.timestamps = this.timestamps;
- alignedTvList.indices = this.indices;
- alignedTvList.values = values;
- alignedTvList.bitMaps = bitMaps;
- alignedTvList.rowCount = this.rowCount;
- alignedTvList.allValueColDeletedMap = getAllValueColDeletedMap();
- alignedTvList.materializedBitmapMemoryCost =
calculateBitmapRamCost(bitMaps, null);
- alignedTvList.refreshArrayMemCostWithoutIndex();
- alignedTvList.refreshMaterializedValueArrayMemoryCost();
- return alignedTvList;
}
@Override
@@ -236,13 +205,19 @@ public abstract class AlignedTVList extends TVList {
public synchronized PartialClonePlan preparePartialClone(Set<Integer>
columnsToClone) {
Set<Integer> retainedColumns =
new HashSet<>(Objects.requireNonNull(columnsToClone, "columnsToClone
cannot be null"));
- AlignedTVList cloneList = AlignedTVList.newAlignedList(new
ArrayList<>(dataTypes));
+ AlignedTVList cloneList = AlignedTVList.newAlignedList(new
ArrayList<>(dataTypes), false);
+ // Pre-create the inner value lists for the retained columns; the other
slots stay null until
+ // the ownership transfer moves the source columns into place.
+ for (int i = 0; i < values.size(); i++) {
+ if (retainedColumns.contains(i)) {
+ cloneList.values.set(i, new ArrayList<>(values.get(i).size()));
+ }
+ }
cloneAs(cloneList);
cloneColumnDataTo(cloneList, retainedColumns);
return prepareMovePlan(cloneList, retainedColumns);
}
- @SuppressWarnings("unchecked")
private PartialClonePlan prepareMovePlan(AlignedTVList cloneList,
Set<Integer> retainedColumns) {
Objects.requireNonNull(cloneList, "cloneList cannot be null");
int columnCount = values.size();
@@ -251,23 +226,21 @@ public abstract class AlignedTVList extends TVList {
throw new IllegalStateException("Target AlignedTVList has incompatible
column containers");
}
- List<Object>[] valueColumnsToMove = (List<Object>[]) new
List<?>[columnCount];
- List<BitMap>[] bitmapColumnsToMove = (List<BitMap>[]) new
List<?>[columnCount];
+ // Validate the move without allocating any O(N) move-plan arrays;
commit() re-derives the
+ // moved columns from the retained set, which only needs this validation
to be complete.
for (int i = 0; i < columnCount; i++) {
if (retainedColumns.contains(i)) {
continue;
}
- List<Object> columnValues = values.get(i);
- if (columnValues == null) {
+ if (values.get(i) == null) {
throw new IllegalStateException(
String.format("Missing value arrays for aligned column index %d
during move", i));
}
- if (cloneList.values.get(i) == null ||
!cloneList.values.get(i).isEmpty()) {
+ if (cloneList.values.get(i) != null) {
throw new IllegalStateException(
String.format("Target value column index %d is not ready for
move", i));
}
- valueColumnsToMove[i] = columnValues;
if (bitMaps != null && bitMaps.get(i) != null) {
if (cloneList.bitMaps == null
@@ -276,31 +249,32 @@ public abstract class AlignedTVList extends TVList {
throw new IllegalStateException(
String.format("Target bitmap column index %d is not ready for
move", i));
}
- bitmapColumnsToMove[i] = bitMaps.get(i);
}
}
return new PartialClonePlan(
this,
cloneList,
- valueColumnsToMove,
- bitmapColumnsToMove,
- calculateArrayMemCostWithoutIndex(retainedColumns),
- cloneList.calculateArrayMemCostWithoutIndex(null),
+ retainedColumns,
calculateBitmapRamCost(bitMaps, retainedColumns),
calculateBitmapRamCost(bitMaps, null));
}
private synchronized void commitPartialClone(PartialClonePlan plan) {
- for (int i = 0; i < plan.valueColumnsToMove.length; i++) {
- List<Object> columnValues = plan.valueColumnsToMove[i];
+ Set<Integer> retainedColumns = plan.retainedColumns;
+ for (int i = 0; i < dataTypes.size(); i++) {
+ if (retainedColumns.contains(i)) {
+ continue;
+ }
+ List<Object> columnValues = values.get(i);
if (columnValues == null) {
+ // Defensive: prepareMovePlan already validated that every moved
column is materialized.
continue;
}
plan.cloneList.values.set(i, columnValues);
values.set(i, null);
- List<BitMap> columnBitMaps = plan.bitmapColumnsToMove[i];
+ List<BitMap> columnBitMaps = bitMaps == null ? null : bitMaps.get(i);
if (columnBitMaps != null) {
plan.cloneList.bitMaps.set(i, columnBitMaps);
bitMaps.set(i, null);
@@ -308,8 +282,8 @@ public abstract class AlignedTVList extends TVList {
memoryBinaryChunkSize[i] = 0;
}
- arrayMemCostWithoutIndex = plan.sourceArrayMemCostWithoutIndex;
- plan.cloneList.arrayMemCostWithoutIndex =
plan.cloneArrayMemCostWithoutIndex;
+ refreshArrayMemCostWithoutIndex();
+ plan.cloneList.refreshArrayMemCostWithoutIndex();
materializedBitmapMemoryCost = plan.sourceBitmapMemoryCost;
plan.cloneList.materializedBitmapMemoryCost = plan.cloneBitmapMemoryCost;
refreshMaterializedValueArrayMemoryCost();
@@ -1252,6 +1226,7 @@ public abstract class AlignedTVList extends TVList {
size += listRamCostWithReferences(dataTypes);
size += RamUsageEstimator.sizeOfLongArray(memoryBinaryChunkSize.length);
+ size +=
RamUsageEstimator.sizeOfIntArray(materializedValueArrayCounts.length);
size += listRamCostWithoutReferences(timestamps);
if (indices != null) {
size += listRamCostWithoutReferences(indices);
@@ -1284,11 +1259,12 @@ public abstract class AlignedTVList extends TVList {
}
private static long listRamCostWithReferences(List<?> list) {
- return RamUsageEstimator.shallowSizeOf(list) +
RamUsageEstimator.sizeOfObjectArray(list.size());
+ return MemoryEstimationHelper.ARRAY_LIST_INSTANCE_SIZE
+ + RamUsageEstimator.sizeOfObjectArray(list.size());
}
private static long listRamCostWithoutReferences(List<?> list) {
- return RamUsageEstimator.shallowSizeOf(list)
+ return MemoryEstimationHelper.ARRAY_LIST_INSTANCE_SIZE
+ (list.isEmpty() ? 0 : RamUsageEstimator.sizeOfObjectArray(0));
}
@@ -1347,11 +1323,12 @@ public abstract class AlignedTVList extends TVList {
/** Initial list-container memory before the first aligned row is written. */
public static long alignedTvListInitialMemCost(int measurementColumnCount) {
- long arrayListShallowSize =
RamUsageEstimator.shallowSizeOfInstance(ArrayList.class);
+ long arrayListShallowSize =
MemoryEstimationHelper.ARRAY_LIST_INSTANCE_SIZE;
long listWithReferencesSize =
arrayListShallowSize +
RamUsageEstimator.sizeOfObjectArray(measurementColumnCount);
return 2 * listWithReferencesSize
+ RamUsageEstimator.sizeOfLongArray(measurementColumnCount)
+ + RamUsageEstimator.sizeOfIntArray(measurementColumnCount)
+ arrayListShallowSize
+ (long) measurementColumnCount * arrayListShallowSize;
}
@@ -1553,6 +1530,9 @@ public abstract class AlignedTVList extends TVList {
size +=
RamUsageEstimator.sizeOfLongArray(newTotalColumnCount)
- RamUsageEstimator.sizeOfLongArray(oldColumnCount);
+ size +=
+ RamUsageEstimator.sizeOfIntArray(newTotalColumnCount)
+ - RamUsageEstimator.sizeOfIntArray(oldColumnCount);
size +=
RamUsageEstimator.sizeOfObjectArray(newTotalColumnCount)
- RamUsageEstimator.sizeOfObjectArray(oldColumnCount);
@@ -1562,7 +1542,7 @@ public abstract class AlignedTVList extends TVList {
- RamUsageEstimator.sizeOfObjectArray(oldColumnCount);
}
long newColumnContainerCost =
- RamUsageEstimator.shallowSizeOf(new ArrayList<>())
+ MemoryEstimationHelper.ARRAY_LIST_INSTANCE_SIZE
+ (timestamps.isEmpty() ? 0 :
RamUsageEstimator.sizeOfObjectArray(0));
return size + newColumnCount * newColumnContainerCost;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackAlignedTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackAlignedTVList.java
index 2603790bd66..c4f5174d195 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackAlignedTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/BackAlignedTVList.java
@@ -27,7 +27,11 @@ public class BackAlignedTVList extends QuickAlignedTVList {
private final BackwardSort policy;
BackAlignedTVList(List<TSDataType> types) {
- super(types);
+ this(types, true);
+ }
+
+ BackAlignedTVList(List<TSDataType> types, boolean initializeValueColumns) {
+ super(types, initializeValueColumns);
policy = new BackwardSort(this);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
index c5bd5550056..9ea50a06246 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/QuickAlignedTVList.java
@@ -26,7 +26,11 @@ public class QuickAlignedTVList extends AlignedTVList {
private final QuickSort policy;
QuickAlignedTVList(List<TSDataType> types) {
- super(types);
+ this(types, true);
+ }
+
+ QuickAlignedTVList(List<TSDataType> types, boolean initializeValueColumns) {
+ super(types, initializeValueColumns);
policy = new QuickSort(this);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
index 60e421c9af5..2536836069c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TVList.java
@@ -540,11 +540,6 @@ public abstract class TVList implements WALEntryValue {
throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
}
- public TVList getTvListByColumnIndex(
- List<Integer> columnIndexList, List<TSDataType> dataTypeList) {
- throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
- }
-
public long getMaxTime() {
return maxTime;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
index 61d5c75096e..c39297d59b6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/TimAlignedTVList.java
@@ -26,7 +26,11 @@ public class TimAlignedTVList extends AlignedTVList {
private final TimSort policy;
TimAlignedTVList(List<TSDataType> types) {
- super(types);
+ this(types, true);
+ }
+
+ TimAlignedTVList(List<TSDataType> types, boolean initializeValueColumns) {
+ super(types, initializeValueColumns);
policy = new TimSort(this);
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java
index 0f1b1c7d253..8bde6b302d5 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceExecutionTest.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.queryengine.execution.fragment;
import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
import org.apache.iotdb.commons.exception.IllegalPathException;
import org.apache.iotdb.commons.exception.MetadataException;
+import org.apache.iotdb.commons.path.AlignedPath;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -48,9 +49,12 @@ import com.google.common.collect.ImmutableSet;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
+import org.apache.tsfile.read.TimeValuePair;
import org.apache.tsfile.read.reader.IPointReader;
+import org.apache.tsfile.utils.TsPrimitiveType;
import org.apache.tsfile.write.schema.IMeasurementSchema;
import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Mockito;
@@ -58,7 +62,9 @@ import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintStream;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
@@ -67,11 +73,23 @@ import static
org.apache.iotdb.db.queryengine.common.QueryId.MOCK_QUERY_ID;
import static
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext.createFragmentInstanceContext;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
public class FragmentInstanceExecutionTest {
+ @BeforeClass
+ public static void setUpClass() {
+ // Initialize DataNodeId before any test to avoid
ExceptionInInitializerError when
+ // Coordinator.<clinit> is triggered indirectly by async state-change
listeners
+ // (e.g., via QueryRelatedResourceMetricSet -> Coordinator ->
QueryIdGenerator).
+ IoTDBDescriptor.getInstance().getConfig().setDataNodeId(1);
+ }
+
@Test
public void testFragmentInstanceExecution() {
ExecutorService instanceNotificationExecutor =
@@ -239,6 +257,133 @@ public class FragmentInstanceExecutionTest {
}
}
+ @Test
+ public void testAlignedTVListPartialColumnCloneEndToEnd() {
+ ExecutorService instanceNotificationExecutor =
+ IoTDBThreadPoolFactory.newFixedThreadPool(2,
"test-aligned-partial-clone");
+ try {
+ // Create an unsorted aligned MemTable with 5 columns. Row i (0..99) has
+ // value i * 100 + j in column j, and rows are inserted in reverse order.
+ List<IMeasurementSchema> schemaList = new ArrayList<>();
+ for (int i = 0; i < 5; i++) {
+ schemaList.add(new MeasurementSchema("sensor_" + i, TSDataType.INT64));
+ }
+ String deviceId = "d1";
+ IMemTable memTable = createMemTable(deviceId, schemaList);
+
+ assertEquals(1, memTable.getMemTableMap().size());
+ IWritableMemChunkGroup memChunkGroup =
memTable.getMemTableMap().values().iterator().next();
+ assertEquals(1, memChunkGroup.getMemChunkMap().size());
+ IWritableMemChunk memChunk =
memChunkGroup.getMemChunkMap().values().iterator().next();
+ AlignedTVList workingTvList = (AlignedTVList)
memChunk.getWorkingTVList();
+ assertFalse(workingTvList.isSorted());
+ assertEquals(100, workingTvList.rowCount());
+
+ // Two concurrent query contexts.
+ FragmentInstanceId id1 = new FragmentInstanceId(new
PlanFragmentId(MOCK_QUERY_ID, 1), "1");
+ FragmentInstanceStateMachine stateMachine1 =
+ new FragmentInstanceStateMachine(id1, instanceNotificationExecutor);
+ FragmentInstanceContext context1 = createFragmentInstanceContext(id1,
stateMachine1);
+
+ FragmentInstanceId id2 = new FragmentInstanceId(new
PlanFragmentId(MOCK_QUERY_ID, 2), "2");
+ FragmentInstanceStateMachine stateMachine2 =
+ new FragmentInstanceStateMachine(id2, instanceNotificationExecutor);
+ FragmentInstanceContext context2 = createFragmentInstanceContext(id2,
stateMachine2);
+
+ // Query 1: sensor_2 and sensor_0. It stays active on the unsorted
working TVList
+ // (no point reader is opened yet, so the list is not sorted).
+ List<String> measurements1 = Arrays.asList("sensor_2", "sensor_0");
+ List<IMeasurementSchema> schemas1 = Arrays.asList(schemaList.get(2),
schemaList.get(0));
+ AlignedPath fullPath1 = new AlignedPath(deviceId, measurements1,
schemas1);
+ ReadOnlyMemChunk readOnlyMemChunk1 =
+ memTable.query(context1, fullPath1, Long.MIN_VALUE, null, null);
+
+ // Query 2: sensor_1 and sensor_3. Because Query 1 is still active on
the unsorted
+ // working TVList, this triggers clone-and-swap of the working TVList.
+ List<String> measurements2 = Arrays.asList("sensor_1", "sensor_3");
+ List<IMeasurementSchema> schemas2 = Arrays.asList(schemaList.get(1),
schemaList.get(3));
+ AlignedPath fullPath2 = new AlignedPath(deviceId, measurements2,
schemas2);
+ ReadOnlyMemChunk readOnlyMemChunk2 =
+ memTable.query(context2, fullPath2, Long.MIN_VALUE, null, null);
+
+ // Query 1's columns (0 and 2) stay in the old working TVList; the other
columns are
+ // moved to the clone.
+ assertEquals(
+ new HashSet<>(Arrays.asList(0, 2)),
context1.getAccessedAlignedColumns(workingTvList));
+ assertNotNull(workingTvList.getValues().get(0));
+ assertNull(workingTvList.getValues().get(1));
+ assertNotNull(workingTvList.getValues().get(2));
+ assertNull(workingTvList.getValues().get(3));
+ assertNull(workingTvList.getValues().get(4));
+
+ // The memChunk now points to the clone, which owns all 5 columns.
+ AlignedTVList cloneTvList = (AlignedTVList) memChunk.getWorkingTVList();
+ assertNotSame(workingTvList, cloneTvList);
+ assertEquals(100, cloneTvList.rowCount());
+ assertNotNull(cloneTvList.getValues().get(0));
+ assertNotNull(cloneTvList.getValues().get(1));
+ assertNotNull(cloneTvList.getValues().get(2));
+ assertNotNull(cloneTvList.getValues().get(3));
+ assertNotNull(cloneTvList.getValues().get(4));
+
+ // The old working TVList is owned by Query 1, which also reserved its
memory.
+ assertSame(context1, workingTvList.getOwnerQuery());
+ assertTrue(workingTvList.getReservedMemoryBytes() > 0);
+
+ // Both queries must still read all 100 rows with correct values.
+ IPointReader pointReader1 = readOnlyMemChunk1.getPointReader();
+ IPointReader pointReader2 = readOnlyMemChunk2.getPointReader();
+ long row = 0;
+ while (pointReader1.hasNextTimeValuePair() &&
pointReader2.hasNextTimeValuePair()) {
+ TimeValuePair tvPair1 = pointReader1.nextTimeValuePair();
+ TimeValuePair tvPair2 = pointReader2.nextTimeValuePair();
+
+ assertEquals(row, tvPair1.getTimestamp());
+ assertEquals(row, tvPair2.getTimestamp());
+
+ // Query 1 reads [sensor_2, sensor_0] in query order.
+ TsPrimitiveType[] values1 = tvPair1.getValue().getVector();
+ assertEquals(2, values1.length);
+ assertEquals(row * 100 + 2, values1[0].getLong());
+ assertEquals(row * 100 + 0, values1[1].getLong());
+
+ // Query 2 reads [sensor_1, sensor_3] in query order.
+ TsPrimitiveType[] values2 = tvPair2.getValue().getVector();
+ assertEquals(2, values2.length);
+ assertEquals(row * 100 + 1, values2[0].getLong());
+ assertEquals(row * 100 + 3, values2[1].getLong());
+ row++;
+ }
+ assertEquals(100, row);
+ assertFalse(pointReader1.hasNextTimeValuePair());
+ assertFalse(pointReader2.hasNextTimeValuePair());
+
+ // Before ending the queries, the retained source's reserved memory must
equal the value
+ // recalculated at cleanup time - this is exactly the accounting
releaseTVListOwnedByQuery
+ // validates when the owning query ends (a mismatch would emit a WARN
during release).
+ long reservedBeforeRelease = workingTvList.getReservedMemoryBytes();
+ assertTrue(reservedBeforeRelease > 0);
+ assertEquals(workingTvList.calculateRamSize().getRamSize(),
reservedBeforeRelease);
+
+ // End Query 1, the owner of the retained source. No other query uses
the source any more,
+ // so the final release must return its reserved memory and clear the
list.
+ context1.releaseResource();
+ assertTrue(workingTvList.getQueryContextSet().isEmpty());
+ assertEquals(0, workingTvList.rowCount());
+
+ // End Query 2. The clone is now the memTable's working TVList, so
cleanup only detaches the
+ // query from it; the clone itself must remain intact as the working
list.
+ context2.releaseResource();
+ assertTrue(cloneTvList.getQueryContextSet().isEmpty());
+ assertEquals(100, cloneTvList.rowCount());
+ assertSame(cloneTvList, memChunk.getWorkingTVList());
+ } catch (Exception e) {
+ fail(e.getMessage());
+ } finally {
+ instanceNotificationExecutor.shutdown();
+ }
+ }
+
private FragmentInstanceExecution createFragmentInstanceExecution(int id,
Executor executor)
throws CpuNotEnoughException {
IDriverScheduler scheduler = Mockito.mock(IDriverScheduler.class);
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 c878b9109d9..3ec333d264e 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
@@ -691,11 +691,11 @@ public class TsFileProcessorTest {
// Test Tablet
processor.insertTablet(genInsertTableNode(0, true), 0, 10, new
TSStatus[10]);
IMemTable memTable = processor.getWorkMemTable();
- Assert.assertEquals(1764688, memTable.getTVListsRamCost());
+ Assert.assertEquals(1776704, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNode(100, true), 0, 10, new
TSStatus[10]);
- Assert.assertEquals(1764688, memTable.getTVListsRamCost());
+ Assert.assertEquals(1776704, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNode(200, true), 0, 10, new
TSStatus[10]);
- Assert.assertEquals(1764688, memTable.getTVListsRamCost());
+ Assert.assertEquals(1776704, memTable.getTVListsRamCost());
Assert.assertEquals(90000, memTable.getTotalPointsNum());
Assert.assertEquals(720360, memTable.memSize());
// Test records
@@ -704,7 +704,7 @@ public class TsFileProcessorTest {
record.addTuple(DataPoint.getDataPoint(dataType, measurementId,
String.valueOf(i)));
processor.insert(buildInsertRowNodeByTSRecord(record), new long[4]);
}
- Assert.assertEquals(1766304, memTable.getTVListsRamCost());
+ Assert.assertEquals(1778320, memTable.getTVListsRamCost());
Assert.assertEquals(90100, memTable.getTotalPointsNum());
Assert.assertEquals(721560, memTable.memSize());
}
@@ -727,21 +727,21 @@ public class TsFileProcessorTest {
// Test Tablet
processor.insertTablet(genInsertTableNode(0, true), 0, 10, new
TSStatus[10]);
IMemTable memTable = processor.getWorkMemTable();
- Assert.assertEquals(1764688, memTable.getTVListsRamCost());
+ Assert.assertEquals(1776704, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNodeFors3000ToS6000(0, true), 0, 10,
new TSStatus[10]);
- Assert.assertEquals(4152728, memTable.getTVListsRamCost());
+ Assert.assertEquals(4176744, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNode(100, true), 0, 10, new
TSStatus[10]);
- Assert.assertEquals(4152728, memTable.getTVListsRamCost());
+ Assert.assertEquals(4176744, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNodeFors3000ToS6000(100, true), 0,
10, new TSStatus[10]);
- Assert.assertEquals(4152728, memTable.getTVListsRamCost());
+ Assert.assertEquals(4176744, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNode(200, true), 0, 10, new
TSStatus[10]);
- Assert.assertEquals(4152728, memTable.getTVListsRamCost());
+ Assert.assertEquals(4176744, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNodeFors3000ToS6000(200, true), 0,
10, new TSStatus[10]);
- Assert.assertEquals(4152728, memTable.getTVListsRamCost());
+ Assert.assertEquals(4176744, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNode(300, true), 0, 10, new
TSStatus[10]);
- Assert.assertEquals(5953280, memTable.getTVListsRamCost());
+ Assert.assertEquals(5977296, memTable.getTVListsRamCost());
processor.insertTablet(genInsertTableNodeFors3000ToS6000(300, true), 0,
10, new TSStatus[10]);
- Assert.assertEquals(7705280, memTable.getTVListsRamCost());
+ Assert.assertEquals(7729296, memTable.getTVListsRamCost());
Assert.assertEquals(240000, memTable.getTotalPointsNum());
Assert.assertEquals(1920960, memTable.memSize());
@@ -751,14 +751,14 @@ public class TsFileProcessorTest {
record.addTuple(DataPoint.getDataPoint(dataType, measurementId,
String.valueOf(i)));
processor.insert(buildInsertRowNodeByTSRecord(record), new long[4]);
}
- Assert.assertEquals(7706896, memTable.getTVListsRamCost());
+ Assert.assertEquals(7730912, memTable.getTVListsRamCost());
// Test records
for (int i = 1; i <= 100; i++) {
TSRecord record = new TSRecord(i, deviceId);
record.addTuple(DataPoint.getDataPoint(dataType, "s1",
String.valueOf(i)));
processor.insert(buildInsertRowNodeByTSRecord(record), new long[4]);
}
- Assert.assertEquals(7708512, memTable.getTVListsRamCost());
+ Assert.assertEquals(7732528, memTable.getTVListsRamCost());
Assert.assertEquals(240200, memTable.getTotalPointsNum());
Assert.assertEquals(1923360, memTable.memSize());
}
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 3f8f1811997..f76752894a0 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
@@ -258,6 +258,8 @@ public class AlignedTVListTest {
- RamUsageEstimator.sizeOfObjectArray(oldColumnCount))
+ RamUsageEstimator.sizeOfLongArray(oldColumnCount + 1)
- RamUsageEstimator.sizeOfLongArray(oldColumnCount)
+ + RamUsageEstimator.sizeOfIntArray(oldColumnCount + 1)
+ - RamUsageEstimator.sizeOfIntArray(oldColumnCount)
+ RamUsageEstimator.shallowSizeOf(new ArrayList<>())
+ RamUsageEstimator.sizeOfObjectArray(0);
tvList.extendColumn(TSDataType.INT32);
@@ -425,6 +427,47 @@ public class AlignedTVListTest {
Assert.assertEquals(retainedRamSize,
tvList.calculateRamSize().getRamSize());
}
+ // After a partial ownership transfer the new working list keeps the full
pre-clone RAM
+ // (retained columns are copied, remaining columns are moved), and its
write-cost estimator must
+ // match the actual RAM delta when a new block is written.
+ @Test
+ public void testPartialCloneKeepsWorkingListRamAndWriteCostMatchesActual() {
+ List<TSDataType> dataTypes =
+ Arrays.asList(TSDataType.INT64, TSDataType.INT32, TSDataType.DOUBLE);
+ AlignedTVList tvList = AlignedTVList.newAlignedList(dataTypes);
+
+ // Fill exactly one block so the next write crosses a block boundary.
+ for (int i = 0; i < ARRAY_SIZE; i++) {
+ tvList.putAlignedValue(i, new Object[] {(long) i, i, (double) i});
+ }
+
+ long fullRamBeforeClone = tvList.getRamSize();
+
+ AlignedTVList.PartialClonePlan plan =
tvList.preparePartialClone(Collections.singleton(0));
+ AlignedTVList workingClone = plan.getCloneList();
+ plan.commit();
+
+ // The new working list still carries all columns, so its complete RAM
must equal the
+ // pre-clone working-list RAM.
+ Assert.assertEquals(fullRamBeforeClone, workingClone.getRamSize());
+
+ // Write a new block into the working list. The incremental write estimate
must equal the
+ // actual RAM increase.
+ long ramBeforeWrite = workingClone.getRamSize();
+ long nextTime = ARRAY_SIZE;
+ Object[] rowValues = {nextTime, (int) nextTime, (double) nextTime};
+ long estimatedWriteCost =
+ workingClone.alignedWriteRowMemCost(
+ new int[] {0, 1, 2},
+ new TSDataType[] {TSDataType.INT64, TSDataType.INT32,
TSDataType.DOUBLE},
+ rowValues,
+ 0);
+ workingClone.putAlignedValue(nextTime, rowValues);
+ long ramAfterWrite = workingClone.getRamSize();
+
+ Assert.assertEquals(estimatedWriteCost, ramAfterWrite - ramBeforeWrite);
+ }
+
@Test
public void testPartialRamSizeIncludesWideColumnContainers() {
int columnCount = 256;
