This is an automated email from the ASF dual-hosted git repository.
JackieTien97 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 17e3b4e9460 clone partial columns of aligned tvlist for query (#18409)
17e3b4e9460 is described below
commit 17e3b4e9460c0417f28be525545289eaeea3ed1f
Author: shuwenwei <[email protected]>
AuthorDate: Tue Aug 11 14:27:10 2026 +0800
clone partial columns of aligned tvlist for query (#18409)
---
.../planner/memory/MemoryReservationManager.java | 6 +
.../rate/GroupedRateAccumulatorMemoryTest.java | 5 +
.../rate/RateAccumulatorFactoryTest.java | 3 +
.../RateFunctionIntermediateStateCodecTest.java | 5 +
.../apache/iotdb/db/i18n/DataNodeMiscMessages.java | 19 +
.../apache/iotdb/db/i18n/DataNodeMiscMessages.java | 19 +
.../fragment/FragmentInstanceContext.java | 51 +-
.../execution/fragment/QueryContext.java | 11 +
.../memory/FakedMemoryReservationManager.java | 3 +
.../NotThreadSafeMemoryReservationManager.java | 16 +-
.../memory/ThreadSafeMemoryReservationManager.java | 5 +
.../schemaregion/utils/ResourceByPathUtils.java | 339 +++++++++----
.../memtable/AbstractWritableMemChunk.java | 26 +-
.../memtable/AlignedReadOnlyMemChunk.java | 4 +-
.../dataregion/memtable/ReadOnlyMemChunk.java | 4 +-
.../db/utils/datastructure/AlignedTVList.java | 536 +++++++++++++++++++--
.../db/utils/datastructure/BackAlignedTVList.java | 6 +-
.../db/utils/datastructure/QuickAlignedTVList.java | 6 +-
.../iotdb/db/utils/datastructure/TVList.java | 11 +
.../db/utils/datastructure/TimAlignedTVList.java | 6 +-
.../fragment/FragmentInstanceExecutionTest.java | 212 +++++++-
.../fragment/QueryModificationLoaderTest.java | 5 +
.../LocalExecutionPlannerOperatorsMemoryTest.java | 46 ++
.../utils/ResourceByPathUtilsTest.java | 113 +++++
.../dataregion/memtable/TsFileProcessorTest.java | 58 ++-
.../db/utils/datastructure/AlignedTVListTest.java | 217 ++++++++-
26 files changed, 1525 insertions(+), 207 deletions(-)
diff --git
a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/memory/MemoryReservationManager.java
b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/memory/MemoryReservationManager.java
index f0420330652..18ad895bb31 100644
---
a/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/memory/MemoryReservationManager.java
+++
b/iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/plan/planner/memory/MemoryReservationManager.java
@@ -48,6 +48,12 @@ public interface MemoryReservationManager {
*/
void releaseMemoryCumulatively(final long size);
+ /**
+ * Release the given size immediately. This is used to roll back a
reservation when the operation
+ * protected by that reservation fails before ownership is published.
+ */
+ void releaseMemoryImmediately(final long size);
+
/**
* Release all reserved memory immediately. Make sure this method is called
when the lifecycle of
* this manager ends, Or the memory to be released in the batch may not be
released correctly.
diff --git
a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedRateAccumulatorMemoryTest.java
b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedRateAccumulatorMemoryTest.java
index 7aa15ebfdea..54c0dc4478a 100644
---
a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedRateAccumulatorMemoryTest.java
+++
b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedRateAccumulatorMemoryTest.java
@@ -74,6 +74,11 @@ public class GroupedRateAccumulatorMemoryTest {
cumulativeRelease += size;
}
+ @Override
+ public void releaseMemoryImmediately(long size) {
+ cumulativeRelease += size;
+ }
+
@Override
public void releaseAllReservedMemory() {}
diff --git
a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/RateAccumulatorFactoryTest.java
b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/RateAccumulatorFactoryTest.java
index 364e4439c26..e2556a3cc03 100644
---
a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/RateAccumulatorFactoryTest.java
+++
b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/RateAccumulatorFactoryTest.java
@@ -161,6 +161,9 @@ public class RateAccumulatorFactoryTest {
@Override
public void releaseMemoryCumulatively(long size) {}
+ @Override
+ public void releaseMemoryImmediately(long size) {}
+
@Override
public void releaseAllReservedMemory() {}
diff --git
a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/RateFunctionIntermediateStateCodecTest.java
b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/RateFunctionIntermediateStateCodecTest.java
index ea535e05c5c..aed2a17d96c 100644
---
a/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/RateFunctionIntermediateStateCodecTest.java
+++
b/iotdb-core/calc-commons/src/test/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/RateFunctionIntermediateStateCodecTest.java
@@ -128,6 +128,11 @@ public class RateFunctionIntermediateStateCodecTest {
outstandingReservation -= size;
}
+ @Override
+ public void releaseMemoryImmediately(long size) {
+ outstandingReservation -= size;
+ }
+
@Override
public void releaseAllReservedMemory() {
outstandingReservation = 0;
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
index 080fabdc546..e76211672ff 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
@@ -1512,4 +1512,23 @@ public final class DataNodeMiscMessages {
public static final String
MISC_EXCEPTION_FAILED_TO_RESOLVE_CANONICAL_PATH_FOR_ACTIVE_LOAD_LISTENING_DIRECTORY_S_ARG_0E6A508E
=
"Failed to resolve canonical path for active load listening directory
%s: %s";
+ public static final String EXCEPTION_COLUMNSTOCLONE_CANNOT_BE_NULL_458FDF37 =
+ "columnsToClone cannot be null";
+ public static final String EXCEPTION_CLONELIST_CANNOT_BE_NULL_47AEEA8F =
+ "cloneList cannot be null";
+ public static final String
EXCEPTION_TARGET_ALIGNEDTVLIST_HAS_INCOMPATIBLE_COLUMN_CONTAINERS_31FAC613 =
+ "Target AlignedTVList has incompatible column containers";
+ public static final String
EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_MOVE_08D46037
=
+ "Missing value arrays for aligned column index %d during move";
+ public static final String
EXCEPTION_TARGET_VALUE_COLUMN_INDEX_ARG_IS_NOT_READY_FOR_MOVE_7889C74F =
+ "Target value column index %d is not ready for move";
+ public static final String
EXCEPTION_TARGET_BITMAP_COLUMN_INDEX_ARG_IS_NOT_READY_FOR_MOVE_AE3B5F88 =
+ "Target bitmap column index %d is not ready for move";
+ public static final String
EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_CLONE_795EB1C5
=
+ "Missing value arrays for aligned column index %d during clone";
+ public static final String
EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_EXPAND_68E0C8B6
=
+ "Missing value arrays for aligned column index %d during expand";
+ public static final String
EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_MARK_NULL_VALUE_2893628E
=
+ "Missing value arrays for aligned column index %d during mark null
value";
+
}
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
index 346dc8dafc1..c9a3f49710f 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
@@ -1489,4 +1489,23 @@ public final class DataNodeMiscMessages {
public static final String
MISC_EXCEPTION_FAILED_TO_RESOLVE_CANONICAL_PATH_FOR_ACTIVE_LOAD_LISTENING_DIRECTORY_S_ARG_0E6A508E
=
"无法解析 Active Load 监听目录 %s 的 canonical 路径：%s";
+ public static final String EXCEPTION_COLUMNSTOCLONE_CANNOT_BE_NULL_458FDF37 =
+ "columnsToClone 不能为 null";
+ public static final String EXCEPTION_CLONELIST_CANNOT_BE_NULL_47AEEA8F =
+ "cloneList 不能为 null";
+ public static final String
EXCEPTION_TARGET_ALIGNEDTVLIST_HAS_INCOMPATIBLE_COLUMN_CONTAINERS_31FAC613 =
+ "目标 AlignedTVList 的列容器不兼容";
+ public static final String
EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_MOVE_08D46037
=
+ "移动过程中缺少对齐列索引 %d 的值数组";
+ public static final String
EXCEPTION_TARGET_VALUE_COLUMN_INDEX_ARG_IS_NOT_READY_FOR_MOVE_7889C74F =
+ "目标值列索引 %d 尚未准备好进行移动";
+ public static final String
EXCEPTION_TARGET_BITMAP_COLUMN_INDEX_ARG_IS_NOT_READY_FOR_MOVE_AE3B5F88 =
+ "目标 bitmap 列索引 %d 尚未准备好进行移动";
+ public static final String
EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_CLONE_795EB1C5
=
+ "克隆过程中缺少对齐列索引 %d 的值数组";
+ public static final String
EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_EXPAND_68E0C8B6
=
+ "扩容过程中缺少对齐列索引 %d 的值数组";
+ public static final String
EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_MARK_NULL_VALUE_2893628E
=
+ "标记空值过程中缺少对齐列索引 %d 的值数组";
+
}
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 e29e2a2141a..b62e6a50990 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
@@ -78,8 +78,10 @@ import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
@@ -184,6 +186,9 @@ public class FragmentInstanceContext extends QueryContext {
private long closedUnseqFileNum = 0;
private boolean highestPriority = false;
+ // accessed value columns on each referenced AlignedTVList.
+ private Map<TVList, Set<Integer>> alignedTVListColumnAccessMap = new
ConcurrentHashMap<>();
+
public static FragmentInstanceContext createFragmentInstanceContext(
FragmentInstanceId id,
FragmentInstanceStateMachine stateMachine,
@@ -253,6 +258,49 @@ public class FragmentInstanceContext extends QueryContext {
return queryDataSourceType == QueryDataSourceType.EXTERNAL_TSFILE_SCAN;
}
+ /**
+ * Record columns of the AlignedTVList accessed by the query. This method is
called from
+ * prepareTvListMapForQuery with tvList.lockQueryList() held. Even though
the HashSet inside
+ * alignedTVListColumnAccessMap is not thread-safe, the calling pattern
guarantees thread safety
+ * without requiring additional synchronization.
+ *
+ * @param tvList the TVList being accessed
+ * @param columnIndexList list of column indices being accessed
+ */
+ public void putAccessedColumns(TVList tvList, List<Integer> columnIndexList)
{
+ Set<Integer> accessedColumns =
+ alignedTVListColumnAccessMap.computeIfAbsent(tvList, ignored -> new
HashSet<>());
+ columnIndexList.stream()
+ .filter(Objects::nonNull)
+ .forEach(
+ index -> {
+ if (index >= 0) {
+ accessedColumns.add(index);
+ }
+ });
+ }
+
+ /** Remove column-access metadata for an unpublished TVList when clone
preparation fails. */
+ public void removeAccessedColumns(TVList tvList) {
+ alignedTVListColumnAccessMap.remove(tvList);
+ }
+
+ /**
+ * Get columns of the AlignedTVList accessed by the query. This method is
called from
+ * prepareTvListMapForQuery with tvList.lockQueryList() held, ensuring that
no other thread can
+ * change accessed columns for the same TVList concurrently.
+ *
+ * @param tvList the TVList being accessed
+ * @return set of column indices being accessed, or null if the TVList is
not tracked by this
+ * query. An empty (non-null) set means the TVList is tracked but only
the time column is
+ * accessed (e.g. a time-only scan), which is different from being
untracked.
+ */
+ @Override
+ public Set<Integer> getAccessedAlignedColumns(TVList tvList) {
+ Set<Integer> accessedColumns = alignedTVListColumnAccessMap.get(tvList);
+ return accessedColumns == null ? null :
Collections.unmodifiableSet(accessedColumns);
+ }
+
@TestOnly
public static FragmentInstanceContext createFragmentInstanceContext(
FragmentInstanceId id, FragmentInstanceStateMachine stateMachine) {
@@ -1041,12 +1089,12 @@ public class FragmentInstanceContext extends
QueryContext {
*/
private void releaseTVListOwnedByQuery() {
for (TVList tvList : tvListSet) {
- long tvListRamSize = tvList.calculateRamSize().getRamSize();
tvList.lockQueryList();
Set<QueryContext> queryContextSet = tvList.getQueryContextSet();
try {
queryContextSet.remove(this);
if (tvList.getOwnerQuery() == this) {
+ long tvListRamSize = tvList.calculateRamSize().getRamSize();
if (tvList.getReservedMemoryBytes() != tvListRamSize) {
LOGGER.warn(
DataNodeQueryMessages
@@ -1131,6 +1179,7 @@ public class FragmentInstanceContext extends QueryContext
{
// release TVList/AlignedTVList owned by current query
releaseTVListOwnedByQuery();
+ alignedTVListColumnAccessMap = null;
fileModCache = null;
tables = null;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryContext.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryContext.java
index 83b567b0650..a9cea56e96e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryContext.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryContext.java
@@ -282,6 +282,17 @@ public class QueryContext {
tvListSet.addAll(set);
}
+ /**
+ * Get columns of the TVList accessed by this query, or null if the query
does not track
+ * column-level access for the TVList (e.g. a non-FragmentInstanceContext
query, or a TVList not
+ * tracked by a FragmentInstanceContext). An empty (non-null) set means the
TVList is tracked but
+ * only the time column is accessed (e.g. a time-only scan), which is
different from being
+ * untracked.
+ */
+ public Set<Integer> getAccessedAlignedColumns(TVList tvList) {
+ return null;
+ }
+
public void addRowLevelFilteredCount(long count) {
throw new UnsupportedOperationException(
DataNodeQueryMessages
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/FakedMemoryReservationManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/FakedMemoryReservationManager.java
index d1c34e365ef..8b45582f138 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/FakedMemoryReservationManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/FakedMemoryReservationManager.java
@@ -37,6 +37,9 @@ public class FakedMemoryReservationManager implements
MemoryReservationManager {
@Override
public void releaseMemoryCumulatively(long size) {}
+ @Override
+ public void releaseMemoryImmediately(long size) {}
+
@Override
public void releaseAllReservedMemory() {}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/NotThreadSafeMemoryReservationManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/NotThreadSafeMemoryReservationManager.java
index 71924894c7c..39e987a4b2f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/NotThreadSafeMemoryReservationManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/NotThreadSafeMemoryReservationManager.java
@@ -80,7 +80,14 @@ public class NotThreadSafeMemoryReservationManager
implements MemoryReservationM
public void reserveMemoryCumulatively(final long size) {
bytesToBeReserved += size;
if (bytesToBeReserved >= MEMORY_BATCH_THRESHOLD) {
- reserveMemoryImmediately();
+ try {
+ reserveMemoryImmediately();
+ } catch (RuntimeException | Error failure) {
+ // reserveMemoryImmediately can fail only while asking the planner for
memory, before it
+ // updates this manager's counters. Keep the caller-visible
reservation operation atomic.
+ bytesToBeReserved -= size;
+ throw failure;
+ }
}
}
@@ -129,6 +136,13 @@ public class NotThreadSafeMemoryReservationManager
implements MemoryReservationM
}
}
+ @Override
+ public void releaseMemoryImmediately(final long size) {
+ if (size > 0) {
+ releaseBytesImmediately(size);
+ }
+ }
+
private void releaseBytesImmediately(final long size) {
long poolBytes = deductReleaseAccounting(size);
if (poolBytes > 0) {
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/ThreadSafeMemoryReservationManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/ThreadSafeMemoryReservationManager.java
index 0a1c6eee418..71676e5b77f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/ThreadSafeMemoryReservationManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/memory/ThreadSafeMemoryReservationManager.java
@@ -51,6 +51,11 @@ public class ThreadSafeMemoryReservationManager extends
NotThreadSafeMemoryReser
super.releaseMemoryCumulatively(size);
}
+ @Override
+ public synchronized void releaseMemoryImmediately(long size) {
+ super.releaseMemoryImmediately(size);
+ }
+
@Override
public synchronized void releaseAllReservedMemory() {
super.releaseAllReservedMemory();
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
index 17d5dfd2995..f95fcfda473 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java
@@ -40,6 +40,7 @@ import
org.apache.iotdb.db.storageengine.dataregion.modification.ModEntry;
import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import org.apache.iotdb.db.utils.ModificationUtils;
import org.apache.iotdb.db.utils.SchemaUtils;
+import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.db.utils.datastructure.TVList;
import org.apache.tsfile.enums.TSDataType;
@@ -73,6 +74,7 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.iotdb.commons.path.AlignedPath.VECTOR_PLACEHOLDER;
@@ -129,7 +131,8 @@ public abstract class ResourceByPathUtils {
QueryContext context,
IWritableMemChunk memChunk,
boolean isWorkMemTable,
- Filter globalTimeFilter) {
+ Filter globalTimeFilter,
+ List<Integer> columnIndexList) {
// should copy globalTimeFilter because GroupByMonthFilter is stateful
Filter copyTimeFilter = null;
if (globalTimeFilter != null) {
@@ -150,121 +153,249 @@ public abstract class ResourceByPathUtils {
.SCHEMA_LOG_FLUSHING_WORKING_MEMTABLE_ADD_CURRENT_QUERY_CONTEXT_TO_IMMUTABLE_7B7CD373);
tvList.getQueryContextSet().add(context);
tvListQueryMap.put(tvList, tvList.rowCount());
+ // columnIndexList is to track column-level access for AlignedTVList.
+ // For TVList (primitive time series), it remains null and column
tracking is not needed.
+ if (columnIndexList != null && context instanceof
FragmentInstanceContext) {
+ ((FragmentInstanceContext) context).putAccessedColumns(tvList,
columnIndexList);
+ }
} finally {
tvList.unlockQueryList();
}
}
- // mutable tvlist
- TVList list = memChunk.getWorkingTVList();
- TVList cloneList = null;
- TVList.RamInfo listRamInfo = list.calculateRamSize();
- list.lockQueryList();
- try {
- if (copyTimeFilter != null
- && !copyTimeFilter.satisfyStartEndTime(list.getMinTime(),
list.getMaxTime())) {
- return tvListQueryMap;
- }
+ TVList.RamInfo listRamInfo = null;
+
+ // calculateRamSize (synchronized method on TVList) was previously called
before
+ // lockQueryList to avoid deadlock concerns. For partial clone of
AlignedTVList, however
+ // calculateRamSize must now be called inside the lockQueryList section
because it depends on
+ // accessing columns on the AlignedTVList.
+ // This is safe because the lock ordering — queryListLock must always be
acquired before the
+ // TVList intrinsic lock (via synchronized methods like calculateRamSize,
clone). So no AB-BA
+ // deadlock is possible.
+ while (true) {
+ // The working TVList may be replaced by a concurrent query via
clone-and-swap
+ // (memChunk.setWorkingTVList(clone)). A queryListLock held on a
detached candidate does
+ // not protect the current working TVList, so after acquiring the lock,
re-verify it is
+ // still the current working list under the memChunk lock. If it was
replaced while
+ // waiting for candidate's queryListLock, retry with the current one.
+ final TVList candidate = memChunk.getWorkingTVList();
+ candidate.lockQueryList();
+ try {
+ synchronized (memChunk) {
+ if (memChunk.getWorkingTVList() != candidate) {
+ continue;
+ }
+ }
- if (!isWorkMemTable) {
- /*
- * 1. Q1 queries this TVList while it is still in the working memtable
and records a smaller
- * visible row count.
- * 2. Later writes append out-of-order rows to the same TVList, then
FLUSH moves the
- * memtable to the flushing list.
- * 3. Q2 queries the flushing memtable. If Q2 directly reuses the
original mutable TVList,
- * Q2's query-side sort may reorder the indices in place.
- * 4. Q1 continues to read with its old row count and the reordered
indices. The converted
- * value index can exceed Q1's bitmap range and cause out-of-bound
access.
- *
- * Therefore, this flushing branch can reuse the original list only
when it is already
- * sorted or no active query is using it. Otherwise, Q2 should read
from
- * workingListForFlush.
- */
- boolean canUseListDirectly = list.isSorted() ||
list.getQueryContextSet().isEmpty();
- LOGGER.debug(
- DataNodeSchemaMessages
-
.SCHEMA_LOG_FLUSHING_MEMTABLE_ADD_CURRENT_QUERY_CONTEXT_TO_MUTABLE_TVLIST_BEB0D766);
- if (canUseListDirectly) {
- list.getQueryContextSet().add(context);
- tvListQueryMap.put(list, list.rowCount());
- } else {
- TVList workingListForFlushSort =
memChunk.initWorkingListForFlushIfNecessary(list, true);
+ if (copyTimeFilter != null
+ && !copyTimeFilter.satisfyStartEndTime(
+ candidate.getMinTime(), candidate.getMaxTime())) {
+ return tvListQueryMap;
+ }
+
+ if (!isWorkMemTable) {
/*
- * The query will read from workingListForFlushSort, but
cloneForFlushSort() only clones
- * times and indices. The value arrays and bitmaps are still shared
with the original
- * list.
- *
- * Therefore, this query must also hold the original list until it
finishes. Adding
- * context to list.getQueryContextSet() lets flush/query cleanup see
that the original
- * list is still in use. Adding list to context.tvListSet makes
- * releaseTVListOwnedByQuery() remove this context from the original
list later.
+ * 1. Q1 queries this TVList while it is still in the working
memtable and records a smaller
+ * visible row count.
+ * 2. Later writes append out-of-order rows to the same TVList, then
FLUSH moves the
+ * memtable to the flushing list.
+ * 3. Q2 queries the flushing memtable. If Q2 directly reuses the
original mutable TVList,
+ * Q2's query-side sort may reorder the indices in place.
+ * 4. Q1 continues to read with its old row count and the reordered
indices. The converted
+ * value index can exceed Q1's bitmap range and cause
out-of-bound access.
*
- * Do not put the original list into tvListQueryMap here. The actual
read path must use
- * workingListForFlushSort to avoid sorting the original list in
place.
+ * Therefore, this flushing branch can reuse the original list only
when it is already
+ * sorted or no active query is using it. Otherwise, Q2 should read
from
+ * workingListForFlush.
*/
- list.getQueryContextSet().add(context);
- context.addTVListToSet(Collections.singleton(list));
- workingListForFlushSort.getQueryContextSet().add(context);
- tvListQueryMap.put(workingListForFlushSort,
workingListForFlushSort.rowCount());
- }
- } else {
- if (list.isSorted() || list.getQueryContextSet().isEmpty()) {
+ boolean canUseListDirectly =
+ candidate.isSorted() || candidate.getQueryContextSet().isEmpty();
LOGGER.debug(
DataNodeSchemaMessages
-
.SCHEMA_LOG_WORKING_MEMTABLE_ADD_CURRENT_QUERY_CONTEXT_TO_MUTABLE_TVLIST_8C937414);
- list.getQueryContextSet().add(context);
- tvListQueryMap.put(list, list.rowCount());
- } else {
- /*
- * +----------------------+
- * | MemTable |
- * | |
- * | +------------+ | +-----------------+
- * | | TVList |<---+--+ +---+ Previous Query |
- * | +-----^------+ | | | +-----------------+
- * | | | | |
- * +----------+-----------+ | | +----------------+
- * | Clone +---+---+ Current Query |
- * +-----+------+ | +----------------+
- * | TVList | <---------+
- * +------------+
- */
+
.SCHEMA_LOG_FLUSHING_MEMTABLE_ADD_CURRENT_QUERY_CONTEXT_TO_MUTABLE_TVLIST_BEB0D766);
+ if (canUseListDirectly) {
+ candidate.getQueryContextSet().add(context);
+ tvListQueryMap.put(candidate, candidate.rowCount());
+ } else {
+ TVList workingListForFlushSort =
+ memChunk.initWorkingListForFlushIfNecessary(candidate, true);
+ /*
+ * The query will read from workingListForFlushSort, but
cloneForFlushSort() only clones
+ * times and indices. The value arrays and bitmaps are still
shared with the original
+ * list.
+ *
+ * Therefore, this query must also hold the original list until it
finishes. Adding
+ * context to list.getQueryContextSet() lets flush/query cleanup
see that the original
+ * list is still in use. Adding list to context.tvListSet makes
+ * releaseTVListOwnedByQuery() remove this context from the
original list later.
+ *
+ * Do not put the original list into tvListQueryMap here. The
actual read path must use
+ * workingListForFlushSort to avoid sorting the original list in
place.
+ */
+ candidate.getQueryContextSet().add(context);
+ context.addTVListToSet(Collections.singleton(candidate));
+ // Query preparation is serialized by candidate's query-list lock,
but cleanup removes
+ // the context under workingListForFlushSort's own lock. Use the
same lock for this add
+ // to avoid concurrently mutating its HashSet. The lock order here
is candidate first,
+ // then workingListForFlushSort; cleanup never holds both locks at
the same time.
+ workingListForFlushSort.lockQueryList();
+ try {
+ workingListForFlushSort.getQueryContextSet().add(context);
+ } finally {
+ workingListForFlushSort.unlockQueryList();
+ }
+ tvListQueryMap.put(workingListForFlushSort,
workingListForFlushSort.rowCount());
+ }
+
+ // columnIndexList is to track column-level access for AlignedTVList.
+ // For TVList (primitive time series), it remains null and column
tracking is not needed.
+ if (columnIndexList != null && context instanceof
FragmentInstanceContext) {
+ ((FragmentInstanceContext) context).putAccessedColumns(candidate,
columnIndexList);
+ }
+ return tvListQueryMap;
+ }
+
+ if (candidate.isSorted() || candidate.getQueryContextSet().isEmpty()) {
LOGGER.debug(
DataNodeSchemaMessages
-
.SCHEMA_LOG_WORKING_MEMTABLE_CLONE_MUTABLE_TVLIST_AND_REPLACE_OLD_TVLIST_FD1EAE22);
- QueryContext firstQuery =
list.getQueryContextSet().iterator().next();
- // reserve query memory
- if (firstQuery instanceof FragmentInstanceContext) {
- MemoryReservationManager memoryReservationManager =
- ((FragmentInstanceContext)
firstQuery).getMemoryReservationContext();
-
memoryReservationManager.reserveMemoryCumulatively(listRamInfo.getRamSize());
- list.setReservedMemoryBytes(listRamInfo.getRamSize());
+
.SCHEMA_LOG_WORKING_MEMTABLE_ADD_CURRENT_QUERY_CONTEXT_TO_MUTABLE_TVLIST_8C937414);
+ candidate.getQueryContextSet().add(context);
+ tvListQueryMap.put(candidate, candidate.rowCount());
+
+ // columnIndexList is to track column-level access for AlignedTVList.
+ // For TVList (primitive time series), it remains null and column
tracking is not needed.
+ if (columnIndexList != null && context instanceof
FragmentInstanceContext) {
+ ((FragmentInstanceContext) context).putAccessedColumns(candidate,
columnIndexList);
+ }
+ return tvListQueryMap;
+ }
+
+ /*
+ * +----------------------+
+ * | MemTable |
+ * | |
+ * | +------------+ | +-----------------+
+ * | | TVList |<---+--+ +---+ Previous Query |
+ * | +-----^------+ | | | +-----------------+
+ * | | | | |
+ * +----------+-----------+ | | +----------------+
+ * | Clone +---+---+ Current Query |
+ * +-----+------+ | +----------------+
+ * | TVList | <---------+
+ * +------------+
+ */
+ LOGGER.debug(
+ DataNodeSchemaMessages
+
.SCHEMA_LOG_WORKING_MEMTABLE_CLONE_MUTABLE_TVLIST_AND_REPLACE_OLD_TVLIST_FD1EAE22);
+
+ synchronized (memChunk) {
+ // Re-check defensively before cloning and publishing the
replacement. The clone and the
+ // working-list swap must be done in the same memChunk critical
section, so a concurrent
+ // query can never observe a working TVList whose columns have
already been moved away.
+ if (memChunk.getWorkingTVList() != candidate) {
+ continue;
}
- list.setOwnerQuery(firstQuery);
- // clone TVList
- cloneList = list.clone();
- cloneList.getQueryContextSet().add(context);
- tvListQueryMap.put(cloneList, cloneList.rowCount());
+ // calculateRamSize (synchronized method on TVList) was previously
called before
+ // lockQueryList to avoid deadlock concerns. For partial clone of
AlignedTVList, however
+ // calculateRamSize must now be called inside the lockQueryList
section because it depends
+ // on accessing columns on the AlignedTVList.
+ // This is safe because the lock ordering - queryListLock must
always be acquired before
+ // the TVList intrinsic lock (via synchronized methods like
calculateRamSize, clone). So
+ // no AB-BA deadlock is possible.
+ Set<Integer> columnsToClone = candidate.getAccessedColumnsForQuery();
+ listRamInfo =
+ (columnsToClone == null)
+ ? candidate.calculateRamSize()
+ : ((AlignedTVList)
candidate).calculateRamSize(columnsToClone);
+
+ QueryContext firstQuery =
candidate.getQueryContextSet().iterator().next();
+ TVList cloneList = null;
+ AlignedTVList.PartialClonePlan partialClonePlan = null;
+ FragmentInstanceContext cloneContext =
+ columnIndexList != null && context instanceof
FragmentInstanceContext
+ ? (FragmentInstanceContext) context
+ : null;
+ MemoryReservationManager memoryReservationManager =
+ firstQuery instanceof FragmentInstanceContext
+ ? ((FragmentInstanceContext)
firstQuery).getMemoryReservationContext()
+ : null;
+ boolean reservationNeedsRollback = false;
+ boolean replacementPublished = false;
+ try {
+ // Reserve before allocating the clone, so this transient memory
increase is still
+ // protected by query-memory admission control. Ownership is not
published yet, and a
+ // later preparation failure rolls this exact reservation back
immediately.
+ if (memoryReservationManager != null) {
+
memoryReservationManager.reserveMemoryCumulatively(listRamInfo.getRamSize());
+ reservationNeedsRollback = true;
+ }
+
+ // Clone and validate without changing the source list.
PartialClonePlan.commit is the
+ // only destructive step and is allocation-free.
+ if (columnsToClone == null) {
+ cloneList = candidate.clone();
+ } else {
+ partialClonePlan = ((AlignedTVList)
candidate).preparePartialClone(columnsToClone);
+ cloneList = partialClonePlan.getCloneList();
+ }
+
+ cloneList.getQueryContextSet().add(context);
+ tvListQueryMap.put(cloneList, cloneList.rowCount());
+ if (cloneContext != null) {
+ cloneContext.putAccessedColumns(cloneList, columnIndexList);
+ }
+
+ if (partialClonePlan != null) {
+ partialClonePlan.commit();
+ }
+ memChunk.setWorkingTVList(cloneList);
+ replacementPublished = true;
+
+ // Publish query ownership only after the replacement is fully
committed. The
+ // candidate query-list lock prevents its owner from being
released concurrently.
+ if (memoryReservationManager != null) {
+ candidate.setReservedMemoryBytes(listRamInfo.getRamSize());
+ }
+ candidate.setOwnerQuery(firstQuery);
+ reservationNeedsRollback = false;
+ return tvListQueryMap;
+ } catch (RuntimeException | Error failure) {
+ if (reservationNeedsRollback) {
+ try {
+
memoryReservationManager.releaseMemoryImmediately(listRamInfo.getRamSize());
+ } catch (RuntimeException | Error rollbackFailure) {
+ failure.addSuppressed(rollbackFailure);
+ }
+ }
+
+ // Before commit, remove the only external reference installed for
the unpublished
+ // clone. Its arrays can then be reclaimed while candidate remains
the working list.
+ if (!replacementPublished && cloneList != null) {
+ cloneList.getQueryContextSet().remove(context);
+ tvListQueryMap.remove(cloneList);
+ if (cloneContext != null) {
+ cloneContext.removeAccessedColumns(cloneList);
+ }
+ }
+ throw failure;
+ }
}
+ } catch (MemoryNotEnoughException ex) {
+ if (listRamInfo != null) {
+ LOGGER.warn(
+ DataNodeSchemaMessages.FAILED_TO_RESERVE_MEMORY_TVLIST,
+ listRamInfo.getRamSize(),
+ listRamInfo.getTimestampsSize(),
+ listRamInfo.getArrayMemCost(),
+ listRamInfo.getRowCount(),
+ listRamInfo.getDataTypes());
+ }
+ throw ex;
+ } finally {
+ candidate.unlockQueryList();
}
- } catch (MemoryNotEnoughException ex) {
- LOGGER.warn(
- DataNodeSchemaMessages.FAILED_TO_RESERVE_MEMORY_TVLIST,
- listRamInfo.getRamSize(),
- listRamInfo.getTimestampsSize(),
- listRamInfo.getArrayMemCost(),
- listRamInfo.getRowCount(),
- listRamInfo.getDataTypes());
- throw ex;
- } finally {
- list.unlockQueryList();
}
- if (cloneList != null) {
- memChunk.setWorkingTVList(cloneList);
- }
- return tvListQueryMap;
}
}
@@ -451,11 +582,6 @@ class AlignedResourceByPathUtils extends
ResourceByPathUtils {
}
}
- // prepare AlignedTVList for query. It should clone TVList if necessary.
- Map<TVList, Integer> alignedTvListQueryMap =
- prepareTvListMapForQuery(
- context, alignedMemChunk, modsToMemtable == null,
globalTimeFilter);
-
// column index list for the query
// Columns with inconsistent types will be ignored and set -1
List<Integer> columnIndexList =
@@ -463,6 +589,11 @@ class AlignedResourceByPathUtils extends
ResourceByPathUtils {
List<TimeRange> timeColumnDeletion = null;
List<List<TimeRange>> valueColumnsDeletionList = null;
+ // prepare AlignedTVList for query. It should clone TVList if necessary.
+ Map<TVList, Integer> alignedTvListQueryMap =
+ prepareTvListMapForQuery(
+ context, alignedMemChunk, modsToMemtable == null,
globalTimeFilter, columnIndexList);
+
if (modsToMemtable != null) {
timeColumnDeletion =
ModificationUtils.constructDeletionList(
@@ -696,7 +827,7 @@ class MeasurementResourceByPathUtils extends
ResourceByPathUtils {
}
// prepare TVList for query. It should clone TVList if necessary.
Map<TVList, Integer> tvListQueryMap =
- prepareTvListMapForQuery(context, memChunk, modsToMemtable == null,
globalTimeFilter);
+ prepareTvListMapForQuery(context, memChunk, modsToMemtable == null,
globalTimeFilter, null);
List<TimeRange> deletionList = null;
if (modsToMemtable != null) {
deletionList =
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
index 9eb30ec5911..4f2bb0abd11 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.i18n.StorageEngineMessages;
import
org.apache.iotdb.db.queryengine.execution.fragment.FragmentInstanceContext;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
import
org.apache.iotdb.db.storageengine.dataregion.wal.buffer.IWALByteBufferView;
+import org.apache.iotdb.db.utils.datastructure.AlignedTVList;
import org.apache.iotdb.db.utils.datastructure.BatchEncodeInfo;
import org.apache.iotdb.db.utils.datastructure.TVList;
@@ -43,6 +44,7 @@ import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.Iterator;
import java.util.List;
+import java.util.Set;
import java.util.concurrent.BlockingQueue;
public abstract class AbstractWritableMemChunk implements IWritableMemChunk {
@@ -109,19 +111,41 @@ public abstract class AbstractWritableMemChunk implements
IWritableMemChunk {
}
}
+ /**
+ * Try to release the TVList. If there are active queries, transfer memory
ownership to the first
+ * query. For AlignedTVList, this will release non-query columns before
transferring to reduce
+ * memory footprint.
+ */
private void tryReleaseTvList(TVList tvList) {
- long tvListRamSize = tvList.calculateRamSize().getRamSize();
tvList.lockQueryList();
try {
if (tvList.getQueryContextSet().isEmpty()) {
tvList.clear();
} else {
QueryContext firstQuery =
tvList.getQueryContextSet().iterator().next();
+
+ // For AlignedTVList with active queries, release non-query columns
before
+ // transferring memory ownership to reduce memory footprint.
+ if (tvList instanceof AlignedTVList) {
+ AlignedTVList alignedTVList = (AlignedTVList) tvList;
+
+ // Get the union of all columns accessed by queries. An empty
(non-null) set means all
+ // queries are tracked but only access the time column, so all value
columns are
+ // released; null means some query is untracked and
releaseNonQueryColumns keeps
+ // everything.
+ Set<Integer> accessedColumns =
alignedTVList.getAccessedColumnsForQuery();
+ if (accessedColumns != null) {
+ // Release non-query columns to reduce memory before ownership
transfer
+ alignedTVList.releaseNonQueryColumns(accessedColumns);
+ }
+ }
+
// transfer memory from write process to read process. Here it
reserves read memory and
// releaseFlushedMemTable will release write memory.
if (firstQuery instanceof FragmentInstanceContext) {
MemoryReservationManager memoryReservationManager =
((FragmentInstanceContext)
firstQuery).getMemoryReservationContext();
+ long tvListRamSize = tvList.calculateRamSize().getRamSize();
memoryReservationManager.reserveMemoryCumulatively(tvListRamSize);
tvList.setReservedMemoryBytes(tvListRamSize);
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedReadOnlyMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedReadOnlyMemChunk.java
index 0ba04ad53e6..55e1899ceeb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedReadOnlyMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AlignedReadOnlyMemChunk.java
@@ -128,12 +128,12 @@ public class AlignedReadOnlyMemChunk extends
ReadOnlyMemChunk {
// We must update queryRowCount here, otherwise, it may be used later
to build
// BitMaps, causing bitmap array size mismatch and possible out of
bound.
entry.setValue(alignedTvList.sort());
- long alignedTvListRamSize =
alignedTvList.calculateRamSize().getRamSize();
alignedTvList.lockQueryList();
try {
FragmentInstanceContext ownerQuery =
(FragmentInstanceContext) alignedTvList.getOwnerQuery();
if (ownerQuery != null) {
+ long alignedTvListRamSize =
alignedTvList.calculateRamSize().getRamSize();
long deltaBytes = alignedTvListRamSize -
alignedTvList.getReservedMemoryBytes();
if (deltaBytes > 0) {
ownerQuery.getMemoryReservationContext().reserveMemoryCumulatively(deltaBytes);
@@ -387,12 +387,12 @@ public class AlignedReadOnlyMemChunk extends
ReadOnlyMemChunk {
int queryLength = entry.getValue();
if (!alignedTvList.isSorted() && queryLength >
alignedTvList.seqRowCount()) {
entry.setValue(alignedTvList.sort());
- long alignedTvListRamSize =
alignedTvList.calculateRamSize().getRamSize();
alignedTvList.lockQueryList();
try {
FragmentInstanceContext ownerQuery =
(FragmentInstanceContext) alignedTvList.getOwnerQuery();
if (ownerQuery != null) {
+ long alignedTvListRamSize =
alignedTvList.calculateRamSize().getRamSize();
long deltaBytes = alignedTvListRamSize -
alignedTvList.getReservedMemoryBytes();
if (deltaBytes > 0) {
ownerQuery.getMemoryReservationContext().reserveMemoryCumulatively(deltaBytes);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java
index 39629bbbcaa..bfefb9817bb 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/ReadOnlyMemChunk.java
@@ -139,11 +139,11 @@ public class ReadOnlyMemChunk {
int queryRowCount = entry.getValue();
if (!tvList.isSorted() && queryRowCount > tvList.seqRowCount()) {
entry.setValue(tvList.sort());
- long tvListRamSize = tvList.calculateRamSize().getRamSize();
tvList.lockQueryList();
try {
FragmentInstanceContext ownerQuery = (FragmentInstanceContext)
tvList.getOwnerQuery();
if (ownerQuery != null) {
+ long tvListRamSize = tvList.calculateRamSize().getRamSize();
long deltaBytes = tvListRamSize - tvList.getReservedMemoryBytes();
if (deltaBytes > 0) {
ownerQuery.getMemoryReservationContext().reserveMemoryCumulatively(deltaBytes);
@@ -298,11 +298,11 @@ public class ReadOnlyMemChunk {
int queryLength = entry.getValue();
if (!tvList.isSorted() && queryLength > tvList.seqRowCount()) {
entry.setValue(tvList.sort());
- long tvListRamSize = tvList.calculateRamSize().getRamSize();
tvList.lockQueryList();
try {
FragmentInstanceContext ownerQuery = (FragmentInstanceContext)
tvList.getOwnerQuery();
if (ownerQuery != null) {
+ long tvListRamSize = tvList.calculateRamSize().getRamSize();
long deltaBytes = tvListRamSize - tvList.getReservedMemoryBytes();
if (deltaBytes > 0) {
ownerQuery.getMemoryReservationContext().reserveMemoryCumulatively(deltaBytes);
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 035b9a89bb9..608c363552c 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
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.utils.datastructure;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
@@ -46,6 +47,7 @@ import org.apache.tsfile.read.filter.basic.Filter;
import org.apache.tsfile.utils.Binary;
import org.apache.tsfile.utils.BitMap;
import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.utils.RamUsageEstimator;
import org.apache.tsfile.utils.ReadWriteForEncodingUtils;
import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.apache.tsfile.utils.TsPrimitiveType;
@@ -58,8 +60,10 @@ import java.io.DataInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashSet;
import java.util.List;
import java.util.Objects;
+import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@@ -87,6 +91,49 @@ public abstract class AlignedTVList extends TVList {
private long materializedBitmapMemoryCost;
private long arrayMemCostWithoutPrimitiveArraysAndIndex;
+ /**
+ * A fully prepared partial clone. All allocations and validations are
completed before this plan
+ * is returned, so {@link #commit()} only moves already captured references
and updates primitive
+ * accounting fields.
+ */
+ public static final class PartialClonePlan {
+ private final AlignedTVList sourceList;
+ private final AlignedTVList cloneList;
+ // 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;
+
+ private boolean committed;
+
+ private PartialClonePlan(
+ AlignedTVList sourceList,
+ AlignedTVList cloneList,
+ Set<Integer> retainedColumns,
+ long sourceBitmapMemoryCost,
+ long cloneBitmapMemoryCost) {
+ this.sourceList = sourceList;
+ this.cloneList = cloneList;
+ this.retainedColumns = retainedColumns;
+ this.sourceBitmapMemoryCost = sourceBitmapMemoryCost;
+ this.cloneBitmapMemoryCost = cloneBitmapMemoryCost;
+ }
+
+ public AlignedTVList getCloneList() {
+ return cloneList;
+ }
+
+ /** Commit the prepared ownership transfer. This method is idempotent and
allocation-free. */
+ public synchronized void commit() {
+ if (committed) {
+ return;
+ }
+ sourceList.commitPartialClone(this);
+ committed = true;
+ }
+ }
+
// Data type list -> list of TVList, add 1 when expanded -> primitive array
of basic type.
// A null primitive array means all existing rows in that block are null for
the column.
// Index relation: columnIndex(dataTypeIndex) -> arrayIndex -> elementIndex
@@ -107,26 +154,37 @@ 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()];
- refreshArrayMemCostWithoutPrimitiveArrays();
values = new ArrayList<>(types.size());
for (int i = 0; i < types.size(); i++) {
- values.add(new ArrayList<>(getDefaultArrayNum()));
+ values.add(initializeValueColumns ? new
ArrayList<>(getDefaultArrayNum()) : null);
}
+ // arrayMemCostWithoutPrimitiveArrays depends on per-column value arrays,
so values must be
+ // initialized before computing it
+ refreshArrayMemCostWithoutPrimitiveArrays();
}
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);
+ return new TimAlignedTVList(dataTypes, initializeValueColumns);
}
}
@@ -140,6 +198,7 @@ public abstract class AlignedTVList extends TVList {
}
@Override
+ @TestOnly
public TVList getTvListByColumnIndex(
List<Integer> columnIndexList, List<TSDataType> dataTypeList, boolean
ignoreAllNullRows) {
List<List<Object>> values = new ArrayList<>();
@@ -178,17 +237,16 @@ public abstract class AlignedTVList extends TVList {
alignedTvList.allValueColDeletedMap = ignoreAllNullRows ?
getAllValueColDeletedMap() : null;
alignedTvList.timeColDeletedMap = this.timeColDeletedMap;
alignedTvList.timeDeletedCnt = this.timeDeletedCnt;
- alignedTvList.materializedBitmapMemoryCost =
calculateBitmapRamCost(bitMaps);
+ alignedTvList.materializedBitmapMemoryCost =
calculateBitmapRamCost(bitMaps, null);
for (int i = 0; i < columnIndexList.size(); i++) {
int columnIndex = columnIndexList.get(i);
if (columnIndex != -1 && values.get(i) != null) {
int materializedArrayCount = materializedValueArrayCounts[columnIndex];
alignedTvList.materializedValueArrayCounts[i] = materializedArrayCount;
alignedTvList.materializedValueArrayMemCost +=
- (long) materializedArrayCount *
valueListArrayMemCost(dataTypeList.get(i));
+ (long) materializedArrayCount *
primitiveArrayMemCost(dataTypeList.get(i));
}
}
-
return alignedTvList;
}
@@ -212,39 +270,190 @@ public abstract class AlignedTVList extends TVList {
public synchronized AlignedTVList clone() {
AlignedTVList cloneList = AlignedTVList.newAlignedList(new
ArrayList<>(dataTypes));
cloneAs(cloneList);
- cloneList.timeDeletedCnt = this.timeDeletedCnt;
- System.arraycopy(
- memoryBinaryChunkSize, 0, cloneList.memoryBinaryChunkSize, 0,
dataTypes.size());
+ cloneColumnDataTo(cloneList, null);
+ cloneList.materializedValueArrayCounts =
+ Arrays.copyOf(materializedValueArrayCounts,
materializedValueArrayCounts.length);
+ cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost;
+ return cloneList;
+ }
+
+ /**
+ * Prepare a partial clone without changing this TVList. The returned plan
must be committed only
+ * after the query-memory reservation succeeds.
+ */
+ public synchronized PartialClonePlan preparePartialClone(Set<Integer>
columnsToClone) {
+ Set<Integer> retainedColumns =
+ new HashSet<>(
+ Objects.requireNonNull(
+ columnsToClone,
+
DataNodeMiscMessages.EXCEPTION_COLUMNSTOCLONE_CANNOT_BE_NULL_458FDF37));
+ 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++) {
- // Clone value
- List<Object> columnValues = values.get(i);
- for (Object valueArray : columnValues) {
- cloneList.values.get(i).add(cloneValue(dataTypes.get(i), valueArray));
+ if (retainedColumns.contains(i)) {
+ cloneList.values.set(i, new ArrayList<>(values.get(i).size()));
}
- // Clone bitmap in columnIndex
+ }
+ cloneAs(cloneList);
+ cloneColumnDataTo(cloneList, retainedColumns);
+ return prepareMovePlan(cloneList, retainedColumns);
+ }
+
+ private PartialClonePlan prepareMovePlan(AlignedTVList cloneList,
Set<Integer> retainedColumns) {
+ Objects.requireNonNull(
+ cloneList,
DataNodeMiscMessages.EXCEPTION_CLONELIST_CANNOT_BE_NULL_47AEEA8F);
+ int columnCount = values.size();
+ if (cloneList.values.size() != columnCount
+ || cloneList.memoryBinaryChunkSize.length !=
memoryBinaryChunkSize.length) {
+ throw new IllegalStateException(
+ DataNodeMiscMessages
+
.EXCEPTION_TARGET_ALIGNEDTVLIST_HAS_INCOMPATIBLE_COLUMN_CONTAINERS_31FAC613);
+ }
+
+ // 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;
+ }
+
+ if (values.get(i) == null) {
+ throw new IllegalStateException(
+ String.format(
+ DataNodeMiscMessages
+
.EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_MOVE_08D46037,
+ i));
+ }
+ if (cloneList.values.get(i) != null) {
+ throw new IllegalStateException(
+ String.format(
+ DataNodeMiscMessages
+
.EXCEPTION_TARGET_VALUE_COLUMN_INDEX_ARG_IS_NOT_READY_FOR_MOVE_7889C74F,
+ i));
+ }
+
if (bitMaps != null && bitMaps.get(i) != null) {
- List<BitMap> columnBitMaps = bitMaps.get(i);
- if (cloneList.bitMaps == null) {
- cloneList.bitMaps = new ArrayList<>(dataTypes.size());
- for (int j = 0; j < dataTypes.size(); j++) {
- cloneList.bitMaps.add(null);
- }
+ if (cloneList.bitMaps == null
+ || cloneList.bitMaps.size() != bitMaps.size()
+ || cloneList.bitMaps.get(i) != null) {
+ throw new IllegalStateException(
+ String.format(
+ DataNodeMiscMessages
+
.EXCEPTION_TARGET_BITMAP_COLUMN_INDEX_ARG_IS_NOT_READY_FOR_MOVE_AE3B5F88,
+ i));
}
- if (cloneList.bitMaps.get(i) == null) {
- List<BitMap> cloneColumnBitMaps = new
ArrayList<>(columnBitMaps.size());
- for (BitMap bitMap : columnBitMaps) {
- cloneColumnBitMaps.add(bitMap == null ? null : bitMap.clone());
- }
- cloneList.bitMaps.set(i, cloneColumnBitMaps);
+ }
+ }
+
+ return new PartialClonePlan(
+ this,
+ cloneList,
+ retainedColumns,
+ calculateBitmapRamCost(bitMaps, retainedColumns),
+ calculateBitmapRamCost(bitMaps, null));
+ }
+
+ private synchronized void commitPartialClone(PartialClonePlan plan) {
+ // The clone keeps the deep-copied retained columns, so copy their
accounting too.
+ for (int i = 0; i < dataTypes.size(); i++) {
+ int materializedCount = materializedValueArrayCounts[i];
+ plan.cloneList.materializedValueArrayCounts[i] = materializedCount;
+ if (materializedCount > 0) {
+ plan.cloneList.materializedValueArrayMemCost +=
+ (long) materializedCount * primitiveArrayMemCost(dataTypes.get(i));
+ }
+ }
+ Set<Integer> retainedColumns = plan.retainedColumns;
+ for (int i = 0; i < dataTypes.size(); i++) {
+ if (retainedColumns.contains(i)) {
+ continue;
+ }
+
+ // Move value arrays and bitmaps from the source to the clone. The clone
was created with
+ // empty column containers, so the moved references are set into place.
+ 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 = bitMaps == null ? null : bitMaps.get(i);
+ if (columnBitMaps != null) {
+ plan.cloneList.bitMaps.set(i, columnBitMaps);
+ bitMaps.set(i, null);
+ }
+ memoryBinaryChunkSize[i] = 0;
+
+ // The source no longer owns the moved column's materialized arrays.
+ int materializedCount = materializedValueArrayCounts[i];
+ materializedValueArrayCounts[i] = 0;
+ if (materializedCount > 0) {
+ materializedValueArrayMemCost -=
+ (long) materializedCount * primitiveArrayMemCost(dataTypes.get(i));
+ }
+ }
+
+ materializedBitmapMemoryCost = plan.sourceBitmapMemoryCost;
+ plan.cloneList.materializedBitmapMemoryCost = plan.cloneBitmapMemoryCost;
+ // Refresh the cached per-block cost after the retained columns changed on
both lists.
+ refreshArrayMemCostWithoutPrimitiveArrays();
+ plan.cloneList.refreshArrayMemCostWithoutPrimitiveArrays();
+ }
+
+ /**
+ * Release memory for non-query columns in this TVList. This is used during
memory ownership
+ * transfer from write process to read process to reduce memory footprint.
Only columns that are
+ * accessed by active queries are retained; all other columns are released.
+ *
+ * @param columnsToKeep set of column indices that are accessed by queries
and should be kept. A
+ * null set means no access information is available and nothing is
released. An empty set
+ * means all queries are tracked but only access the time column, so all
value columns are
+ * released.
+ */
+ public synchronized void releaseNonQueryColumns(Set<Integer> columnsToKeep) {
+ if (columnsToKeep == null) {
+ return;
+ }
+
+ for (int i = 0; i < values.size(); i++) {
+ // Skip columns that should be kept or are already null
+ if (columnsToKeep.contains(i)) {
+ continue;
+ }
+
+ List<Object> columnValues = values.get(i);
+ if (columnValues == null) {
+ continue;
+ }
+
+ // Release memory for non-query columns
+ for (Object dataArray : columnValues) {
+ if (dataArray != null) {
+ PrimitiveArrayManager.release(dataArray);
}
}
+ values.set(i, null);
+ memoryBinaryChunkSize[i] = 0;
+
+ // Release bitmap memory for non-query columns
+ if (bitMaps != null && bitMaps.get(i) != null) {
+ bitMaps.set(i, null);
+ }
+
+ // Remove the released column from the materialized-array accounting
+ int materializedCount = materializedValueArrayCounts[i];
+ materializedValueArrayCounts[i] = 0;
+ if (materializedCount > 0) {
+ materializedValueArrayMemCost -=
+ (long) materializedCount * primitiveArrayMemCost(dataTypes.get(i));
+ }
}
- cloneList.timeColDeletedMap = timeColDeletedMap == null ? null :
timeColDeletedMap.clone();
- cloneList.materializedValueArrayCounts =
- Arrays.copyOf(materializedValueArrayCounts,
materializedValueArrayCounts.length);
- cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost;
- cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
- return cloneList;
+
+ materializedBitmapMemoryCost = calculateBitmapRamCost(bitMaps,
columnsToKeep);
+ // Refresh the cached per-block cost after releasing columns
+ refreshArrayMemCostWithoutPrimitiveArrays();
}
@SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity
warning
@@ -618,6 +827,27 @@ public abstract class AlignedTVList extends TVList {
return dataTypes;
}
+ /**
+ * Get the union of all columns accessed by queries on this AlignedTVList.
This method should be
+ * called with queryListLock held for thread safety.
+ *
+ * @return set of accessed column indices, or null if any query on this
TVList is untracked or
+ * does not record accessed columns. An empty (non-null) set means all
queries are tracked but
+ * only access the time column (e.g. a time-only scan), so all value
columns can be released.
+ */
+ @Override
+ public Set<Integer> getAccessedColumnsForQuery() {
+ Set<Integer> accessedColumns = new HashSet<>();
+ for (QueryContext queryContext : getQueryContextSet()) {
+ Set<Integer> columns = queryContext.getAccessedAlignedColumns(this);
+ if (columns == null) {
+ return null;
+ }
+ accessedColumns.addAll(columns);
+ }
+ return accessedColumns;
+ }
+
@Override
/*
* Must be synchronized with sort() on the same TVList instance: a query may
sort
@@ -778,6 +1008,76 @@ public abstract class AlignedTVList extends TVList {
}
}
+ /*
+ * There are two clone modes:
+ * 1. Full clone: columnsToClone is null, meaning no column filter is
applied. All columns are
+ * deep-cloned.
+ * 2. Partial clone: columnsToClone is non-null. Columns in columnsToClone
are deep-cloned for the
+ * query that keeps using the source TVList; columns not in
columnsToClone are not copied here.
+ * They are moved from the source TVList to cloneList later, and
cloneList becomes the new
+ * working list in the memtable.
+ *
+ * This method only performs the allocation phase: copy row-level time
deletion state, clone
+ * requested value/bitmap arrays, and prepare bitmap containers that will be
needed by moved
+ * columns. It must not clear or move columns from
+ * the source TVList here. The destructive move is performed only by
PartialClonePlan.commit()
+ * after cloneList and the ownership-transfer plan are fully prepared for
publication.
+ */
+ private void cloneColumnDataTo(AlignedTVList cloneList, Set<Integer>
columnsToClone) {
+ cloneList.timeDeletedCnt = timeDeletedCnt;
+ cloneList.timeColDeletedMap = timeColDeletedMap == null ? null :
timeColDeletedMap.clone();
+
+ boolean cloneAllColumns = columnsToClone == null;
+ System.arraycopy(
+ memoryBinaryChunkSize, 0, cloneList.memoryBinaryChunkSize, 0,
dataTypes.size());
+ boolean hasBitMapsToMove = false;
+ for (int i = 0; i < values.size(); i++) {
+ // Clone value
+ List<Object> columnValues = values.get(i);
+ if (columnValues == null) {
+ throw new IllegalStateException(
+ String.format(
+ DataNodeMiscMessages
+
.EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_CLONE_795EB1C5,
+ i));
+ }
+ boolean shouldCloneColumn = cloneAllColumns ||
columnsToClone.contains(i);
+ if (!shouldCloneColumn) {
+ hasBitMapsToMove |= bitMaps != null && bitMaps.get(i) != null;
+ continue;
+ }
+
+ for (Object valueArray : columnValues) {
+ cloneList.values.get(i).add(cloneValue(dataTypes.get(i), valueArray));
+ }
+ // Clone bitmap in columnIndex
+ if (bitMaps != null && bitMaps.get(i) != null) {
+ List<BitMap> columnBitMaps = bitMaps.get(i);
+ if (cloneList.bitMaps == null) {
+ cloneList.bitMaps = new ArrayList<>(dataTypes.size());
+ for (int j = 0; j < dataTypes.size(); j++) {
+ cloneList.bitMaps.add(null);
+ }
+ }
+ if (cloneList.bitMaps.get(i) == null) {
+ List<BitMap> cloneColumnBitMaps = new ArrayList<>();
+ for (BitMap bitMap : columnBitMaps) {
+ cloneColumnBitMaps.add(bitMap == null ? null : bitMap.clone());
+ }
+ cloneList.bitMaps.set(i, cloneColumnBitMaps);
+ }
+ }
+ }
+ cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost;
+
+ if (hasBitMapsToMove && cloneList.bitMaps == null) {
+ cloneList.bitMaps = new ArrayList<>(dataTypes.size());
+ for (int i = 0; i < dataTypes.size(); i++) {
+ cloneList.bitMaps.add(null);
+ }
+ }
+ }
+
@Override
protected void clearValue() {
for (int i = 0; i < dataTypes.size(); i++) {
@@ -815,7 +1115,15 @@ public abstract class AlignedTVList extends TVList {
indices.add((int[]) getPrimitiveArraysByType(TSDataType.INT32));
}
for (int i = 0; i < dataTypes.size(); i++) {
- values.get(i).add(null);
+ List<Object> columnValues = values.get(i);
+ if (columnValues == null) {
+ throw new IllegalStateException(
+ String.format(
+ DataNodeMiscMessages
+
.EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_EXPAND_68E0C8B6,
+ i));
+ }
+ columnValues.add(null);
if (bitMaps != null && bitMaps.get(i) != null) {
bitMaps.get(i).add(null);
materializedBitmapMemoryCost += bitmapReferenceRamCost();
@@ -1070,7 +1378,7 @@ public abstract class AlignedTVList extends TVList {
valueArray = getPrimitiveArraysByType(dataTypes.get(columnIndex));
columnValues.set(arrayIndex, valueArray);
materializedValueArrayCounts[columnIndex]++;
- materializedValueArrayMemCost +=
valueListArrayMemCost(dataTypes.get(columnIndex));
+ materializedValueArrayMemCost +=
primitiveArrayMemCost(dataTypes.get(columnIndex));
if (elementIndex > 0) {
getBitMap(columnIndex, arrayIndex).markRange(0, elementIndex);
}
@@ -1079,6 +1387,15 @@ public abstract class AlignedTVList extends TVList {
}
private BitMap getBitMap(int columnIndex, int arrayIndex) {
+ List<Object> columnValues = values.get(columnIndex);
+ if (columnValues == null) {
+ throw new IllegalStateException(
+ String.format(
+ DataNodeMiscMessages
+
.EXCEPTION_MISSING_VALUE_ARRAYS_FOR_ALIGNED_COLUMN_INDEX_ARG_DURING_MARK_NULL_VALUE_2893628E,
+ columnIndex));
+ }
+
// init BitMaps if doesn't have
if (bitMaps == null) {
List<List<BitMap>> localBitMaps = new ArrayList<>(dataTypes.size());
@@ -1090,8 +1407,8 @@ public abstract class AlignedTVList extends TVList {
// if the bitmap in columnIndex is null, init the bitmap of this column
from the beginning
if (bitMaps.get(columnIndex) == null) {
- List<BitMap> columnBitMaps = new
ArrayList<>(values.get(columnIndex).size());
- for (int i = 0; i < values.get(columnIndex).size(); i++) {
+ List<BitMap> columnBitMaps = new ArrayList<>(columnValues.size());
+ for (int i = 0; i < columnValues.size(); i++) {
columnBitMaps.add(null);
}
bitMaps.set(columnIndex, columnBitMaps);
@@ -1137,22 +1454,115 @@ public abstract class AlignedTVList extends TVList {
new ArrayList<>(dataTypes));
}
+ public synchronized RamInfo calculateRamSize(Set<Integer> columnsToClone) {
+ return new RamInfo(
+ timestamps.size(),
+ alignedTvListArrayMemCost(columnsToClone),
+ getRamSize(columnsToClone),
+ rowCount,
+ new ArrayList<>(dataTypes));
+ }
+
public synchronized long getRamSize() {
return (long) timestamps.size() *
alignedTvListArrayMemCostWithoutPrimitiveArrays()
+ materializedValueArrayMemCost
- + materializedBitmapMemoryCost;
+ + materializedBitmapMemoryCost
+ + calculateContainerRamCost(null);
+ }
+
+ public synchronized long getRamSize(Set<Integer> columnsToClone) {
+ long size =
+ (long) timestamps.size() *
alignedTvListArrayMemCostWithoutPrimitiveArrays(columnsToClone);
+ for (int i = 0; i < dataTypes.size(); i++) {
+ if (columnsToClone != null && !columnsToClone.contains(i)) {
+ continue;
+ }
+ TSDataType dataType = dataTypes.get(i);
+ if (dataType != null) {
+ size += (long) materializedValueArrayCounts[i] *
primitiveArrayMemCost(dataType);
+ }
+ }
+ return size
+ + calculateBitmapRamCost(bitMaps, columnsToClone)
+ + calculateContainerRamCost(columnsToClone);
+ }
+
+ /**
+ * Calculate the one-time container memory retained by this list.
+ *
+ * <p>The outer N-wide containers (the {@code values}/{@code bitMaps}
ArrayLists, the {@code
+ * dataTypes} list and the accounting arrays kept after a partial clone) are
not charged anywhere
+ * else and are counted in full, including all their slot references.
+ *
+ * <p>For a retained column, the inner value list is charged with the
references of all its slots
+ * (including null placeholders for blocks that were never materialized), so
each slot reference
+ * is counted exactly once; the materialized primitive array payload and
header are charged once
+ * per materialized block by {@code materializedValueArrayMemCost}. The
timestamps/indices inner
+ * lists only contribute their list object and backing-array header here,
since their primitive
+ * arrays are already charged per block by {@code
+ * alignedTvListArrayMemCostWithoutPrimitiveArrays}.
+ */
+ long calculateContainerRamCost(Set<Integer> retainedColumns) {
+ long size = 0;
+
+ size += listRamCostWithReferences(dataTypes);
+ size += RamUsageEstimator.sizeOfLongArray(memoryBinaryChunkSize.length);
+ size +=
RamUsageEstimator.sizeOfIntArray(materializedValueArrayCounts.length);
+ size += listRamCostWithoutReferences(timestamps);
+ if (indices != null) {
+ size += listRamCostWithoutReferences(indices);
+ }
+
+ size += listRamCostWithReferences(values);
+ for (int i = 0; i < values.size(); i++) {
+ if (retainedColumns != null && !retainedColumns.contains(i)) {
+ continue;
+ }
+ List<Object> columnValues = values.get(i);
+ if (columnValues != null) {
+ size += listRamCostWithReferences(columnValues);
+ }
+ }
+
+ if (bitMaps != null) {
+ size += listRamCostWithReferences(bitMaps);
+ for (int i = 0; i < bitMaps.size(); i++) {
+ if (retainedColumns != null && !retainedColumns.contains(i)) {
+ continue;
+ }
+ List<BitMap> columnBitMaps = bitMaps.get(i);
+ if (columnBitMaps != null) {
+ size += listRamCostWithoutReferences(columnBitMaps);
+ }
+ }
+ }
+ return size;
+ }
+
+ static long listRamCostWithReferences(List<?> list) {
+ return RamUsageEstimator.shallowSizeOf(list) +
RamUsageEstimator.sizeOfObjectArray(list.size());
+ }
+
+ static long listRamCostWithoutReferences(List<?> list) {
+ return RamUsageEstimator.shallowSizeOf(list)
+ + (list.isEmpty() ? 0 : RamUsageEstimator.sizeOfObjectArray(0));
}
- private static long calculateBitmapRamCost(List<List<BitMap>> bitMaps) {
+ private static long calculateBitmapRamCost(
+ List<List<BitMap>> bitMaps, Set<Integer> columnsToClone) {
if (bitMaps == null) {
return 0;
}
long size = 0;
- for (List<BitMap> columnBitMaps : bitMaps) {
+ for (int i = 0, length = bitMaps.size(); i < length; i++) {
+ if (columnsToClone != null && !columnsToClone.contains(i)) {
+ continue;
+ }
+ List<BitMap> columnBitMaps = bitMaps.get(i);
if (columnBitMaps == null) {
continue;
}
- size += (long) columnBitMaps.size() * bitmapReferenceRamCost();
+ size += columnBitMaps.size() * bitmapReferenceRamCost();
for (BitMap bitMap : columnBitMaps) {
if (bitMap != null) {
size += bitMap.ramBytesUsed();
@@ -1200,27 +1610,29 @@ public abstract class AlignedTVList extends TVList {
*
* @return AlignedTvListArrayMemSize
*/
- public long alignedTvListArrayMemCost() {
+ public long alignedTvListArrayMemCost(Set<Integer> columnsToClone) {
long size = 0;
- // value & bitmap array mem size
+ int retainedColumnNum = 0;
+ // value array mem size
for (int column = 0; column < dataTypes.size(); column++) {
+ if (columnsToClone != null && !columnsToClone.contains(column)) {
+ continue;
+ }
TSDataType type = dataTypes.get(column);
- if (type != null) {
+ if (type != null && values.get(column) != null) {
+ retainedColumnNum++;
size += (long) PrimitiveArrayManager.ARRAY_SIZE * (long)
type.getDataTypeSize();
}
}
- // size is 0 when all types are null
- if (size == 0) {
- return size;
- }
+
// time array mem size
size += PrimitiveArrayManager.ARRAY_SIZE * 8L;
// index array mem size
size += (indices != null) ? PrimitiveArrayManager.ARRAY_SIZE * 4L : 0;
// array headers mem size
- size += (long) NUM_BYTES_ARRAY_HEADER * (2 + dataTypes.size());
+ size += (long) NUM_BYTES_ARRAY_HEADER * (2 + retainedColumnNum);
// Object references size in ArrayList
- size += (long) NUM_BYTES_OBJECT_REF * (2 + dataTypes.size());
+ size += (long) NUM_BYTES_OBJECT_REF * (2 + retainedColumnNum);
return size;
}
@@ -1229,13 +1641,27 @@ public abstract class AlignedTVList extends TVList {
+ (indices != null ? (long) PrimitiveArrayManager.ARRAY_SIZE *
Integer.BYTES : 0);
}
+ private long alignedTvListArrayMemCostWithoutPrimitiveArrays(Set<Integer>
retainedColumns) {
+ long size = alignedTvListArrayMemCost(retainedColumns);
+ for (int i = 0; i < dataTypes.size(); i++) {
+ TSDataType dataType = dataTypes.get(i);
+ if (dataType != null
+ && values.get(i) != null
+ && (retainedColumns == null || retainedColumns.contains(i))) {
+ size -= valueListArrayMemCost(dataType);
+ }
+ }
+ return size;
+ }
+
private void refreshArrayMemCostWithoutPrimitiveArrays() {
long size = alignedTvListArrayMemCost();
if (indices != null) {
size -= (long) PrimitiveArrayManager.ARRAY_SIZE * Integer.BYTES;
}
- for (TSDataType dataType : dataTypes) {
- if (dataType != null) {
+ for (int i = 0; i < dataTypes.size(); i++) {
+ TSDataType dataType = dataTypes.get(i);
+ if (dataType != null && values.get(i) != null) {
size -= valueListArrayMemCost(dataType);
}
}
@@ -1255,6 +1681,10 @@ public abstract class AlignedTVList extends TVList {
return size;
}
+ public long alignedTvListArrayMemCost() {
+ return alignedTvListArrayMemCost((Set<Integer>) null);
+ }
+
/**
* Get the single column array mem cost by give type.
*
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 9dbd11cf626..a7aa2ff16e1 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
@@ -551,6 +551,7 @@ public abstract class TVList implements WALEntryValue {
throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
}
+ @TestOnly
public TVList getTvListByColumnIndex(
List<Integer> columnIndexList, List<TSDataType> dataTypeList, boolean
ignoreAllNullRows) {
throw new UnsupportedOperationException(ERR_DATATYPE_NOT_CONSISTENT);
@@ -824,6 +825,16 @@ public abstract class TVList implements WALEntryValue {
return queryContextSet;
}
+ /**
+ * Get the union of all columns accessed by queries on this TVList. For
non-AlignedTVList, returns
+ * empty set. This method should be called with queryListLock held for
thread safety.
+ *
+ * @return set of accessed column indices, or empty set if no columns are
tracked
+ */
+ public Set<Integer> getAccessedColumnsForQuery() {
+ return null;
+ }
+
public List<BitMap> getBitMap() {
return bitMap;
}
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 e8c7994fb20..79e66f92d9e 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
@@ -24,8 +24,8 @@ import org.apache.iotdb.calc.exception.QueryProcessException;
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.AlignedFullPath;
import org.apache.iotdb.commons.path.NonAlignedFullPath;
-import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.common.FragmentInstanceId;
import org.apache.iotdb.db.queryengine.common.PlanFragmentId;
@@ -35,7 +35,7 @@ import
org.apache.iotdb.db.queryengine.execution.exchange.MPPDataExchangeManager
import org.apache.iotdb.db.queryengine.execution.exchange.sink.ISink;
import org.apache.iotdb.db.queryengine.execution.schedule.IDriverScheduler;
import org.apache.iotdb.db.storageengine.dataregion.DataRegion;
-import org.apache.iotdb.db.storageengine.dataregion.memtable.DeviceIDFactory;
+import
org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedWritableMemChunk;
import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
import org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunk;
import
org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunkGroup;
@@ -49,7 +49,10 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IDeviceID;
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;
@@ -59,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;
@@ -69,6 +74,10 @@ 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;
@@ -188,6 +197,55 @@ public class FragmentInstanceExecutionTest {
}
}
+ @Test
+ public void testTVListOwnerTransferTimeOnlyAlignedReleasesAllValueColumns()
+ throws InterruptedException {
+ ExecutorService instanceNotificationExecutor =
+ IoTDBThreadPoolFactory.newFixedThreadPool(1,
"test-instance-notification");
+ try {
+ List<IMeasurementSchema> schemas =
+ new ArrayList<>(
+ Arrays.asList(
+ new MeasurementSchema("s0", TSDataType.INT64,
TSEncoding.PLAIN),
+ new MeasurementSchema("s1", TSDataType.INT64,
TSEncoding.PLAIN),
+ new MeasurementSchema("s2", TSDataType.INT64,
TSEncoding.PLAIN)));
+ AlignedWritableMemChunk memChunk = new AlignedWritableMemChunk(schemas,
true);
+ for (int i = 0; i < 100; i++) {
+ memChunk.putAlignedRow(i, new Object[] {(long) i, (long) i * 2, (long)
i * 3});
+ }
+ AlignedTVList tvList = memChunk.getWorkingTVList();
+
+ // A table-model time-only query: tracked on the TVList with an empty
column list. The empty
+ // set must be distinguished from "untracked" so the flush-time
ownership transfer releases
+ // all value columns.
+ FragmentInstanceId id = new FragmentInstanceId(new
PlanFragmentId(MOCK_QUERY_ID, 1), "1");
+ FragmentInstanceStateMachine stateMachine =
+ new FragmentInstanceStateMachine(id, instanceNotificationExecutor);
+ FragmentInstanceContext queryContext = createFragmentInstanceContext(id,
stateMachine);
+ queryContext.addTVListToSet(ImmutableSet.of(tvList));
+ tvList.lockQueryList();
+ try {
+ tvList.getQueryContextSet().add(queryContext);
+ queryContext.putAccessedColumns(tvList, Collections.emptyList());
+ } finally {
+ tvList.unlockQueryList();
+ }
+
+ // Flush-time ownership transfer: the TVList is handed over to the only
(time-only) query.
+ memChunk.release();
+
+ // All value columns must have been released, while timestamps remain
readable.
+ assertNull(tvList.getValues().get(0));
+ assertNull(tvList.getValues().get(1));
+ assertNull(tvList.getValues().get(2));
+ assertEquals(100, tvList.rowCount());
+ assertEquals(99L, tvList.getTime(99));
+ assertSame(queryContext, tvList.getOwnerQuery());
+ } finally {
+ shutdownAndAwaitTermination(instanceNotificationExecutor);
+ }
+ }
+
private static void shutdownAndAwaitTermination(ExecutorService executor)
throws InterruptedException {
executor.shutdown();
@@ -256,6 +314,137 @@ public class FragmentInstanceExecutionTest {
}
}
+ @Test
+ public void testAlignedTVListPartialColumnCloneEndToEnd() throws
InterruptedException {
+ 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));
+ AlignedFullPath fullPath1 =
+ new AlignedFullPath(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(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));
+ AlignedFullPath fullPath2 =
+ new AlignedFullPath(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(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 {
+ shutdownAndAwaitTermination(instanceNotificationExecutor);
+ }
+ }
+
private FragmentInstanceExecution createFragmentInstanceExecution(int id,
Executor executor)
throws CpuNotEnoughException {
IDriverScheduler scheduler = Mockito.mock(IDriverScheduler.class);
@@ -308,7 +497,7 @@ public class FragmentInstanceExecutionTest {
int rows = 100;
for (int i = 0; i < 100; i++) {
memTable.write(
- DeviceIDFactory.getInstance().getDeviceID(new PartialPath(deviceId)),
+ IDeviceID.Factory.DEFAULT_FACTORY.create(deviceId),
Collections.singletonList(
new MeasurementSchema(measurementId, TSDataType.INT32,
TSEncoding.PLAIN)),
rows - i - 1,
@@ -316,4 +505,21 @@ public class FragmentInstanceExecutionTest {
}
return memTable;
}
+
+ private IMemTable createMemTable(String deviceId, List<IMeasurementSchema>
schemaList)
+ throws IllegalPathException {
+ PrimitiveMemTable memTable = new PrimitiveMemTable("root.test", "1");
+
+ // Insert data in reverse order to make it unsorted
+ int rows = 100;
+ for (int i = rows - 1; i >= 0; i--) {
+ Object[] values = new Object[5];
+ for (int j = 0; j < 5; j++) {
+ values[j] = (long) i * 100 + j;
+ }
+ memTable.writeAlignedRow(
+ IDeviceID.Factory.DEFAULT_FACTORY.create(deviceId), schemaList, i,
values);
+ }
+ return memTable;
+ }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryModificationLoaderTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryModificationLoaderTest.java
index 65cbc0f5b9f..14673ffda64 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryModificationLoaderTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/fragment/QueryModificationLoaderTest.java
@@ -323,6 +323,11 @@ public class QueryModificationLoaderTest {
reservedBytes -= size;
}
+ @Override
+ public void releaseMemoryImmediately(long size) {
+ reservedBytes -= size;
+ }
+
@Override
public void releaseAllReservedMemory() {
reservedBytes = 0;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlannerOperatorsMemoryTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlannerOperatorsMemoryTest.java
index 6d0cabb0443..f7362501833 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlannerOperatorsMemoryTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/planner/LocalExecutionPlannerOperatorsMemoryTest.java
@@ -19,6 +19,7 @@
package org.apache.iotdb.db.queryengine.plan.planner;
+import org.apache.iotdb.calc.exception.MemoryNotEnoughException;
import org.apache.iotdb.db.queryengine.common.QueryId;
import
org.apache.iotdb.db.queryengine.plan.planner.memory.NotThreadSafeMemoryReservationManager;
@@ -161,4 +162,49 @@ public class LocalExecutionPlannerOperatorsMemoryTest {
Assert.assertEquals(0L, manager.getReservedBytesInTotalForTest());
Assert.assertEquals(freeBefore, PLANNER.getFreeMemoryForOperators());
}
+
+ @Test
+ public void testImmediateReservationRollback() {
+ long request = Math.min(1024L, PLANNER.getFreeMemoryForOperators());
+ if (request <= 0) {
+ return;
+ }
+
+ NotThreadSafeMemoryReservationManager manager =
+ new NotThreadSafeMemoryReservationManager(new QueryId("normal_query"),
"test");
+ long freeBefore = PLANNER.getFreeMemoryForOperators();
+
+ manager.reserveMemoryCumulatively(request);
+ manager.releaseMemoryImmediately(request);
+ manager.reserveMemoryImmediately();
+
+ Assert.assertEquals(0L, manager.getReservedBytesInTotalForTest());
+ Assert.assertEquals(freeBefore, PLANNER.getFreeMemoryForOperators());
+
+ manager.reserveMemoryImmediately(request);
+ manager.releaseMemoryImmediately(request);
+
+ Assert.assertEquals(0L, manager.getReservedBytesInTotalForTest());
+ Assert.assertEquals(freeBefore, PLANNER.getFreeMemoryForOperators());
+ }
+
+ @Test
+ public void testFailedCumulativeReservationDoesNotRemainPending() {
+ long freeBefore = PLANNER.getFreeMemoryForOperators();
+ long request = freeBefore + MEMORY_BATCH_THRESHOLD;
+ NotThreadSafeMemoryReservationManager manager =
+ new NotThreadSafeMemoryReservationManager(new QueryId("normal_query"),
"test");
+
+ try {
+ manager.reserveMemoryCumulatively(request);
+ Assert.fail("Expected insufficient query memory");
+ } catch (MemoryNotEnoughException expected) {
+ // expected
+ }
+
+ // A stale pending reservation would make this retry fail again.
+ manager.reserveMemoryImmediately();
+ Assert.assertEquals(0L, manager.getReservedBytesInTotalForTest());
+ Assert.assertEquals(freeBefore, PLANNER.getFreeMemoryForOperators());
+ }
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtilsTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtilsTest.java
new file mode 100644
index 00000000000..334e6836e79
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtilsTest.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iotdb.db.schemaengine.schemaregion.utils;
+
+import org.apache.iotdb.commons.path.NonAlignedFullPath;
+import org.apache.iotdb.db.queryengine.execution.fragment.QueryContext;
+import org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunk;
+import org.apache.iotdb.db.utils.datastructure.TVList;
+
+import org.apache.tsfile.enums.TSDataType;
+import org.apache.tsfile.file.metadata.IDeviceID;
+import org.apache.tsfile.write.schema.MeasurementSchema;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ResourceByPathUtilsTest {
+
+ @Test
+ public void testFlushingQueryLocksTemporaryTVListBeforeRegistration() throws
Exception {
+ TVList candidate = TVList.newList(TSDataType.INT64);
+ candidate.putLong(2, 2);
+ candidate.putLong(1, 1);
+ Assert.assertFalse(candidate.isSorted());
+
+ QueryContext previousQuery = new QueryContext(1, false);
+ candidate.lockQueryList();
+ try {
+ candidate.getQueryContextSet().add(previousQuery);
+ } finally {
+ candidate.unlockQueryList();
+ }
+
+ TVList temporaryList = candidate.cloneForFlushSort();
+ IWritableMemChunk memChunk = mock(IWritableMemChunk.class);
+ when(memChunk.getSortedList()).thenReturn(Collections.emptyList());
+ when(memChunk.getWorkingTVList()).thenReturn(candidate);
+ CountDownLatch temporaryListInitialized = new CountDownLatch(1);
+ when(memChunk.initWorkingListForFlushIfNecessary(candidate, true))
+ .thenAnswer(
+ ignored -> {
+ temporaryListInitialized.countDown();
+ return temporaryList;
+ });
+
+ ResourceByPathUtils resourceByPathUtils =
+ ResourceByPathUtils.getResourceInstance(
+ new NonAlignedFullPath(
+ IDeviceID.Factory.DEFAULT_FACTORY.create("root.test.d"),
+ new MeasurementSchema("s", TSDataType.INT64)));
+ QueryContext currentQuery = new QueryContext(2, false);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ Future<Map<TVList, Integer>> result = null;
+ try {
+ temporaryList.lockQueryList();
+ try {
+ result =
+ executor.submit(
+ () ->
+ resourceByPathUtils.prepareTvListMapForQuery(
+ currentQuery, memChunk, false, null, null));
+ Assert.assertTrue(temporaryListInitialized.await(3, TimeUnit.SECONDS));
+ Future<Map<TVList, Integer>> blockedResult = result;
+ Assert.assertThrows(
+ TimeoutException.class, () -> blockedResult.get(200,
TimeUnit.MILLISECONDS));
+ } finally {
+ temporaryList.unlockQueryList();
+ }
+
+ Map<TVList, Integer> tvListQueryMap = result.get(3, TimeUnit.SECONDS);
+ Assert.assertTrue(tvListQueryMap.containsKey(temporaryList));
+ temporaryList.lockQueryList();
+ try {
+
Assert.assertTrue(temporaryList.getQueryContextSet().contains(currentQuery));
+ } finally {
+ temporaryList.unlockQueryList();
+ }
+ } finally {
+ if (result != null) {
+ result.cancel(true);
+ }
+ executor.shutdownNow();
+ executor.awaitTermination(3, TimeUnit.SECONDS);
+ }
+ }
+}
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 43832097c0e..e1b402a64dd 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
@@ -541,21 +541,21 @@ public class TsFileProcessorTest {
true,
new long[5]);
IMemTable memTable = processor.getWorkMemTable();
- Assert.assertEquals(1596552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1788704, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNode(100, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(1596552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1788704, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNode(200, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(1596552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1788704, memTable.getTVListsRamCost());
Assert.assertEquals(90000, memTable.getTotalPointsNum());
Assert.assertEquals(720360, memTable.memSize());
// Test records
@@ -564,7 +564,7 @@ public class TsFileProcessorTest {
record.addTuple(DataPoint.getDataPoint(dataType, measurementId,
String.valueOf(i)));
processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]);
}
- Assert.assertEquals(1598168, memTable.getTVListsRamCost());
+ Assert.assertEquals(1790320, memTable.getTVListsRamCost());
Assert.assertEquals(90100, memTable.getTotalPointsNum());
Assert.assertEquals(721560, memTable.memSize());
}
@@ -589,9 +589,12 @@ public class TsFileProcessorTest {
actualProcessor.insertTablet(
genSingleMeasurementTablet(rowCount, true), rangeList, actualResults,
false, new long[5]);
+ // The failed-only block never materializes its value array, so the only
difference from the
+ // expected processor is the primitive array payload+header; the slot
reference is charged once
+ // by the container model and is identical on both sides.
Assert.assertEquals(
expectedProcessor.getWorkMemTable().getTVListsRamCost()
- - AlignedTVList.valueListArrayMemCost(dataType),
+ - AlignedTVList.primitiveArrayMemCost(dataType),
actualProcessor.getWorkMemTable().getTVListsRamCost());
Assert.assertEquals(
TSStatusCode.OUT_OF_TTL.getStatusCode(),
actualResults[failedIndex].getCode());
@@ -616,11 +619,13 @@ public class TsFileProcessorTest {
true,
new long[5]);
- long denseBlockCost =
- AlignedTVList.alignedTvListArrayMemCost(
- new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null);
+ // A new block charges the per-block cost (timestamps + headers/refs) plus
the primitive array
+ // of the newly materialized column; the column's slot reference is
charged once by the
+ // container model and is unchanged when the inner list grows by one slot
(8-byte aligned).
Assert.assertEquals(
- denseBlockCost - AlignedTVList.valueListArrayMemCost(TSDataType.INT32),
+ AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays(
+ new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null)
+ + AlignedTVList.primitiveArrayMemCost(TSDataType.INT32),
processor.getWorkMemTable().getTVListsRamCost() -
ramCostBeforeNewBlock);
}
@@ -642,11 +647,13 @@ public class TsFileProcessorTest {
rowNode.setAligned(true);
processor.insert(rowNode, new long[5]);
- long denseBlockCost =
- AlignedTVList.alignedTvListArrayMemCost(
- new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null);
+ // A new block charges the per-block cost (timestamps + headers/refs) plus
the primitive array
+ // of the newly materialized column; the column's slot reference is
charged once by the
+ // container model and is unchanged when the inner list grows by one slot
(8-byte aligned).
Assert.assertEquals(
- denseBlockCost - AlignedTVList.valueListArrayMemCost(TSDataType.INT32),
+ AlignedTVList.alignedTvListArrayMemCostWithoutPrimitiveArrays(
+ new TSDataType[] {TSDataType.INT32, TSDataType.INT32}, null)
+ + AlignedTVList.primitiveArrayMemCost(TSDataType.INT32),
processor.getWorkMemTable().getTVListsRamCost() -
ramCostBeforeNewBlock);
}
@@ -778,57 +785,56 @@ public class TsFileProcessorTest {
true,
new long[5]);
IMemTable memTable = processor.getWorkMemTable();
- Assert.assertEquals(1596552, memTable.getTVListsRamCost());
+ Assert.assertEquals(1788704, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNodeFors3000ToS6000(0, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(3504552, memTable.getTVListsRamCost());
+ Assert.assertEquals(4152744, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNode(100, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(3504552, memTable.getTVListsRamCost());
+ Assert.assertEquals(4152744, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNodeFors3000ToS6000(100, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(3504552, memTable.getTVListsRamCost());
+ Assert.assertEquals(4152744, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNode(200, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(3504552, memTable.getTVListsRamCost());
+ Assert.assertEquals(4152744, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNodeFors3000ToS6000(200, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(3504552, memTable.getTVListsRamCost());
+ Assert.assertEquals(4152744, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNode(300, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(
- 5269104 - 3000L * AlignedTVList.bitmapRamCost(),
memTable.getTVListsRamCost());
+ Assert.assertEquals(5761296, memTable.getTVListsRamCost());
processor.insertTablet(
genInsertTableNodeFors3000ToS6000(300, true),
Collections.singletonList(new int[] {0, 10}),
new TSStatus[10],
true,
new long[5]);
- Assert.assertEquals(7009104, memTable.getTVListsRamCost());
+ Assert.assertEquals(7633296, memTable.getTVListsRamCost());
Assert.assertEquals(240000, memTable.getTotalPointsNum());
Assert.assertEquals(1920960, memTable.memSize());
@@ -838,14 +844,14 @@ public class TsFileProcessorTest {
record.addTuple(DataPoint.getDataPoint(dataType, measurementId,
String.valueOf(i)));
processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]);
}
- Assert.assertEquals(7010720, memTable.getTVListsRamCost());
+ Assert.assertEquals(7634912, memTable.getTVListsRamCost());
// Test records
for (int i = 1; i <= 100; i++) {
TSRecord record = new TSRecord(deviceId, i);
record.addTuple(DataPoint.getDataPoint(dataType, "s1",
String.valueOf(i)));
processor.insert(buildInsertRowNodeByTSRecord(record), new long[5]);
}
- Assert.assertEquals(7012336, memTable.getTVListsRamCost());
+ Assert.assertEquals(7636528, memTable.getTVListsRamCost());
Assert.assertEquals(240200, memTable.getTotalPointsNum());
Assert.assertEquals(1923360, memTable.memSize());
}
@@ -1118,8 +1124,10 @@ public class TsFileProcessorTest {
insertAlignedRow(processor, denseDevice, PrimitiveArrayManager.ARRAY_SIZE,
true);
long denseRowRamIncrement = memTable.getTVListsRamCost() -
ramCostBeforeDenseRow;
+ // Both rows advance the same block (identical container growth), so the
difference is only the
+ // dense row's materialized value array payload+header, not its slot
reference.
Assert.assertEquals(
- AlignedTVList.valueListArrayMemCost(dataType),
+ AlignedTVList.primitiveArrayMemCost(dataType),
denseRowRamIncrement - sparseRowRamIncrement);
}
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 fbbd35cff43..aceb2e23420 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
@@ -36,7 +36,10 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import static
org.apache.iotdb.db.storageengine.rescon.memory.PrimitiveArrayManager.ARRAY_SIZE;
import static org.apache.tsfile.utils.RamUsageEstimator.NUM_BYTES_ARRAY_HEADER;
@@ -175,6 +178,7 @@ public class AlignedTVListTest {
}
Assert.assertNull(tvList.getBitMaps());
+ long ramSizeBeforeBitmapAllocation =
tvList.calculateRamSize().getRamSize();
tvList.putAlignedValue(ARRAY_SIZE * 2 + 1L, new Object[] {null, 1L});
List<BitMap> firstColumnBitMaps = tvList.getBitMaps().get(0);
@@ -188,9 +192,14 @@ public class AlignedTVListTest {
firstColumnBitMaps.get(2).ramBytesUsed() < new
BitMap(ARRAY_SIZE).ramBytesUsed());
Assert.assertTrue(tvList.isNullValue(ARRAY_SIZE * 2 + 1, 0));
Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE * 2, 0));
+ // Only the lazily allocated bitmap, its slot references and the container
of the newly
+ // created bitmap structure are charged; no additional value array is
materialized.
Assert.assertEquals(
- 3L * AlignedTVList.bitmapReferenceRamCost() +
AlignedTVList.bitmapRamCost(),
- tvList.calculateRamSize().getRamSize() - 3L *
tvList.alignedTvListArrayMemCost());
+ 3L * AlignedTVList.bitmapReferenceRamCost()
+ + AlignedTVList.bitmapRamCost()
+ + AlignedTVList.listRamCostWithReferences(tvList.getBitMaps())
+ +
AlignedTVList.listRamCostWithoutReferences(tvList.getBitMaps().get(0)),
+ tvList.calculateRamSize().getRamSize() -
ramSizeBeforeBitmapAllocation);
}
@Test
@@ -266,12 +275,17 @@ public class AlignedTVListTest {
Assert.assertEquals(1, tvList.getLongByValueIndex(ARRAY_SIZE + 1, 1));
long ramSizeBeforeExtension = tvList.calculateRamSize().getRamSize();
+ long containerBeforeExtension = tvList.calculateContainerRamCost(null);
tvList.extendColumn(TSDataType.INT32);
Assert.assertNull(tvList.getValues().get(2).get(0));
Assert.assertNull(tvList.getValues().get(2).get(1));
Assert.assertNull(tvList.getBitMaps().get(2));
- Assert.assertEquals(ramSizeBeforeExtension,
tvList.calculateRamSize().getRamSize());
+ // extendColumn only adds the N-wide container overhead of the new column;
neither a value
+ // array nor a bitmap is materialized.
+ Assert.assertEquals(
+ ramSizeBeforeExtension + tvList.calculateContainerRamCost(null) -
containerBeforeExtension,
+ tvList.calculateRamSize().getRamSize());
long ramSizeBeforeExtendedColumnMaterialization =
tvList.calculateRamSize().getRamSize();
tvList.putAlignedValue(ARRAY_SIZE + 2L, new Object[] {null, null, 2});
@@ -286,10 +300,13 @@ public class AlignedTVListTest {
Assert.assertTrue(tvList.isNullValue(0, 2));
Assert.assertFalse(tvList.isNullValue(ARRAY_SIZE + 2, 2));
Assert.assertEquals(2, tvList.getIntByValueIndex(ARRAY_SIZE + 2, 2));
+ // Materializing the extended column charges its primitive array, the
lazily created bitmap
+ // (slot references + bitmap) and the container of the new bitmap
structure.
Assert.assertEquals(
- AlignedTVList.valueListArrayMemCost(TSDataType.INT32)
+ AlignedTVList.primitiveArrayMemCost(TSDataType.INT32)
+ 2L * AlignedTVList.bitmapReferenceRamCost()
- + AlignedTVList.bitmapRamCost(),
+ + AlignedTVList.bitmapRamCost()
+ +
AlignedTVList.listRamCostWithoutReferences(tvList.getBitMaps().get(2)),
tvList.calculateRamSize().getRamSize() -
ramSizeBeforeExtendedColumnMaterialization);
}
@@ -304,10 +321,14 @@ public class AlignedTVListTest {
long ramSizeBeforeMaterialization = tvList.calculateRamSize().getRamSize();
tvList.putAlignedValue(ARRAY_SIZE + 1L, new Object[] {1L, 1L});
+ // Materializing the second column charges its primitive array, the lazily
created bitmap
+ // (slot references + bitmap) and the container of the new bitmap
structure.
Assert.assertEquals(
- AlignedTVList.valueListArrayMemCost(TSDataType.INT64)
+ AlignedTVList.primitiveArrayMemCost(TSDataType.INT64)
+ 2L * AlignedTVList.bitmapReferenceRamCost()
- + AlignedTVList.bitmapRamCost(),
+ + AlignedTVList.bitmapRamCost()
+ + AlignedTVList.listRamCostWithReferences(tvList.getBitMaps())
+ +
AlignedTVList.listRamCostWithoutReferences(tvList.getBitMaps().get(1)),
tvList.calculateRamSize().getRamSize() - ramSizeBeforeMaterialization);
Assert.assertEquals(
@@ -321,14 +342,18 @@ public class AlignedTVListTest {
Assert.assertEquals(
(long) projectedTvList.getValues().get(0).size()
*
projectedTvList.alignedTvListArrayMemCostWithoutPrimitiveArrays()
- + AlignedTVList.valueListArrayMemCost(TSDataType.INT64)
+ + AlignedTVList.primitiveArrayMemCost(TSDataType.INT64)
+ (long) projectedTvList.getBitMaps().get(0).size()
* AlignedTVList.bitmapReferenceRamCost()
- + AlignedTVList.bitmapRamCost(),
+ + AlignedTVList.bitmapRamCost()
+ + projectedTvList.calculateContainerRamCost(null),
projectedTvList.calculateRamSize().getRamSize());
tvList.clear();
- Assert.assertEquals(0, tvList.calculateRamSize().getRamSize());
+ // clear() keeps the N-wide containers for reuse, so only the retained
(empty) container
+ // baseline remains charged; no per-block, materialized-array or bitmap
payload is left.
+ Assert.assertEquals(
+ tvList.calculateContainerRamCost(null),
tvList.calculateRamSize().getRamSize());
}
@Test
@@ -340,9 +365,12 @@ public class AlignedTVListTest {
}
int blockCount = tvList.getValues().get(0).size();
- long denseRamSize = blockCount * tvList.alignedTvListArrayMemCost();
+ // The second column is never materialized (only nulls were written), so
only the first
+ // column's primitive arrays are charged; the N-wide container baseline is
added once.
long expectedRamSize =
- denseRamSize - blockCount *
AlignedTVList.valueListArrayMemCost(TSDataType.INT64);
+ (long) blockCount *
tvList.alignedTvListArrayMemCostWithoutPrimitiveArrays()
+ + (long) blockCount *
AlignedTVList.primitiveArrayMemCost(TSDataType.INT64)
+ + tvList.calculateContainerRamCost(null);
Assert.assertEquals(expectedRamSize,
tvList.calculateRamSize().getRamSize());
Assert.assertNull(tvList.getBitMaps());
@@ -586,4 +614,169 @@ public class AlignedTVListTest {
Assert.assertEquals(tvList.memoryBinaryChunkSize[0], 0);
Assert.assertEquals(tvList.memoryBinaryChunkSize[1], 0);
}
+
+ @Test
+ public void testMovesUnclonedColumns() {
+ List<TSDataType> dataTypes = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ dataTypes.add(TSDataType.INT64);
+ }
+ AlignedTVList tvList = AlignedTVList.newAlignedList(dataTypes);
+ tvList.putAlignedValue(0, new Object[] {1L, 2L, null});
+
+ Set<Integer> columnsToClone = Collections.singleton(1);
+ long retainedRamSize =
tvList.calculateRamSize(columnsToClone).getRamSize();
+ AlignedTVList.PartialClonePlan partialClonePlan =
tvList.preparePartialClone(columnsToClone);
+ AlignedTVList clonedTvList = partialClonePlan.getCloneList();
+
+ Assert.assertNotNull(tvList.getValues().get(0));
+ Assert.assertNotNull(tvList.getValues().get(2));
+ Assert.assertEquals(1L, tvList.getLongByValueIndex(0, 0));
+ Assert.assertTrue(tvList.isNullValue(0, 2));
+ Assert.assertEquals(2L, clonedTvList.getLongByValueIndex(0, 1));
+
+ partialClonePlan.commit();
+
+ Assert.assertNull(tvList.getValues().get(0));
+ Assert.assertNull(tvList.getValues().get(2));
+ Assert.assertTrue(tvList.isNullValue(0, 0));
+ Assert.assertTrue(tvList.isNullValue(0, 2));
+ Assert.assertEquals(1L, clonedTvList.getLongByValueIndex(0, 0));
+ Assert.assertEquals(2L, clonedTvList.getLongByValueIndex(0, 1));
+ Assert.assertTrue(clonedTvList.isNullValue(0, 2));
+ Assert.assertEquals(retainedRamSize,
tvList.calculateRamSize().getRamSize());
+ }
+
+ @Test
+ public void testPartialRamSizeScalesWithRetainedColumns() {
+ int columnCount = 256;
+ List<TSDataType> dataTypes = new ArrayList<>(columnCount);
+ Object[] values = new Object[columnCount];
+ for (int i = 0; i < columnCount; i++) {
+ dataTypes.add(TSDataType.INT64);
+ values[i] = (long) i;
+ }
+
+ AlignedTVList tvList = AlignedTVList.newAlignedList(dataTypes);
+ tvList.putAlignedValue(1, values);
+ Set<Integer> retainedColumns = Collections.singleton(0);
+ long retainedRamSize =
tvList.calculateRamSize(retainedColumns).getRamSize();
+ long fullRamSize = tvList.calculateRamSize().getRamSize();
+
+ // The N-wide container baseline is retained regardless of how many
columns are cloned, so
+ // exclude it to verify that the per-column payload scales with the
retained column count:
+ // keeping 1 of 256 columns must cost far less than the full list.
+ long retainedPayload = retainedRamSize -
tvList.calculateContainerRamCost(retainedColumns);
+ long fullPayload = fullRamSize - tvList.calculateContainerRamCost(null);
+ Assert.assertTrue(retainedPayload < fullPayload / 64);
+
+ AlignedTVList.PartialClonePlan plan =
tvList.preparePartialClone(retainedColumns);
+ plan.commit();
+ Assert.assertEquals(retainedRamSize,
tvList.calculateRamSize().getRamSize());
+ }
+
+ @Test
+ public void testPartialReservationMatchesCleanupCalculation() {
+ for (boolean createIndices : new boolean[] {false, true}) {
+ for (boolean retainValueColumn : new boolean[] {false, true}) {
+ AlignedTVList tvList =
+ AlignedTVList.newAlignedList(
+ new ArrayList<>(
+ Arrays.asList(TSDataType.INT64, TSDataType.INT64,
TSDataType.INT64)));
+ for (int i = 0; i <= ARRAY_SIZE; i++) {
+ long time = createIndices ? ARRAY_SIZE - i : i;
+ tvList.putAlignedValue(
+ time, new Object[] {(long) i, i % 2 == 0 ? null : (long) i,
(long) i});
+ }
+ if (createIndices) {
+ Assert.assertFalse(tvList.isSorted());
+ tvList.sort();
+ Assert.assertNotNull(tvList.getIndices());
+ } else {
+ Assert.assertNull(tvList.getIndices());
+ }
+
+ Set<Integer> retainedColumns =
+ retainValueColumn ? Collections.singleton(1) :
Collections.emptySet();
+ long reservedMemoryBytes =
tvList.calculateRamSize(retainedColumns).getRamSize();
+ tvList.setReservedMemoryBytes(reservedMemoryBytes);
+
+ AlignedTVList.PartialClonePlan plan =
tvList.preparePartialClone(retainedColumns);
+ plan.commit();
+
+ long cleanupMemoryBytes = tvList.calculateRamSize().getRamSize();
+ String scenario =
+ String.format(
+ "createIndices=%s, retainValueColumn=%s", createIndices,
retainValueColumn);
+ Assert.assertEquals(scenario, reservedMemoryBytes, cleanupMemoryBytes);
+ Assert.assertEquals(scenario, tvList.getReservedMemoryBytes(),
cleanupMemoryBytes);
+ }
+ }
+ }
+
+ @Test
+ public void testPartialCloneFailureLeavesSourceUntouched() {
+ AlignedTVList tvList =
+ AlignedTVList.newAlignedList(
+ Arrays.asList(TSDataType.INT64, TSDataType.INT64,
TSDataType.INT64));
+ // Materialize the first column before writing a null so master creates
its lazy bitmap.
+ tvList.putAlignedValue(0, new Object[] {1L, 2L, 3L});
+ tvList.putAlignedValue(1, new Object[] {null, 4L, 5L});
+
+ List<Object> firstColumnValues = tvList.getValues().get(0);
+ List<Object> secondColumnValues = tvList.getValues().get(1);
+ List<Object> thirdColumnValues = tvList.getValues().get(2);
+ List<BitMap> firstColumnBitMaps = tvList.getBitMaps().get(0);
+ Object invalidThirdColumnArray = new int[ARRAY_SIZE];
+ thirdColumnValues.set(0, invalidThirdColumnArray);
+
+ Set<Integer> columnsToClone = new HashSet<>(Arrays.asList(0, 1, 2));
+ Assert.assertThrows(ClassCastException.class, () ->
tvList.preparePartialClone(columnsToClone));
+
+ Assert.assertSame(firstColumnValues, tvList.getValues().get(0));
+ Assert.assertSame(secondColumnValues, tvList.getValues().get(1));
+ Assert.assertSame(thirdColumnValues, tvList.getValues().get(2));
+ Assert.assertSame(invalidThirdColumnArray,
tvList.getValues().get(2).get(0));
+ Assert.assertSame(firstColumnBitMaps, tvList.getBitMaps().get(0));
+ Assert.assertTrue(tvList.isNullValue(1, 0));
+ Assert.assertEquals(2L, tvList.getLongByValueIndex(0, 1));
+ }
+
+ @Test
+ public void testReleaseNonQueryColumnsWithBitmaps() {
+ List<TSDataType> dataTypes = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ dataTypes.add(TSDataType.INT64);
+ }
+ AlignedTVList tvList = AlignedTVList.newAlignedList(dataTypes);
+ for (int i = 0; i < 100; i++) {
+ Object[] values = new Object[3];
+ values[0] = (long) i;
+ // Alternate non-null and null values so the lazily allocated value
array and bitmap both
+ // exist in every block.
+ values[1] = i % 2 == 0 ? (long) i : null;
+ values[2] = (long) (i * 100);
+ tvList.putAlignedValue(i, values);
+ }
+
+ // Verify bitmaps were created for column 1
+ Assert.assertNotNull(tvList.getBitMaps());
+ Assert.assertNotNull(tvList.getBitMaps().get(1));
+
+ // Keep only column 0 and 2, release column 1
+ Set<Integer> columnsToKeep = new HashSet<>(Arrays.asList(0, 2));
+ tvList.releaseNonQueryColumns(columnsToKeep);
+
+ // Verify column 1 is released
+ Assert.assertNull(tvList.getValues().get(1));
+ Assert.assertNull(tvList.getBitMaps().get(1));
+
+ // Verify columns 0 and 2 are intact
+ Assert.assertFalse(tvList.getValues().get(0).isEmpty());
+ Assert.assertFalse(tvList.getValues().get(2).isEmpty());
+ for (int i = 0; i < 100; i++) {
+ Assert.assertEquals((long) i, tvList.getLongByValueIndex(i, 0));
+ Assert.assertEquals((long) (i * 100), tvList.getLongByValueIndex(i, 2));
+ }
+ }
}
