clone partial columns of aligned tvlist for query (#18409)
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 f042033..18ad895 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
@@ -49,6 +49,12 @@ 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 7aa15eb..54c0dc4 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
@@ -75,6 +75,11 @@ } @Override + public void releaseMemoryImmediately(long size) { + cumulativeRelease += size; + } + + @Override public void releaseAllReservedMemory() {} @Override
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 364e443..e2556a3 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
@@ -162,6 +162,9 @@ public void releaseMemoryCumulatively(long size) {} @Override + public void releaseMemoryImmediately(long size) {} + + @Override public void releaseAllReservedMemory() {} @Override
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 ea535e0..aed2a17 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
@@ -129,6 +129,11 @@ } @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 080fabd..e762116 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 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 346dc8d..c9a3f49 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 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 e29e2a2..b62e6a5 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.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 @@ 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 @@ 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 @@ */ 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 @@ // 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 83b567b..a9cea56e 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 @@ 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 d1c34e3..8b45582 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
@@ -38,6 +38,9 @@ public void releaseMemoryCumulatively(long size) {} @Override + public void releaseMemoryImmediately(long size) {} + + @Override public void releaseAllReservedMemory() {} @Override
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 7192489..39e987a 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 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 @@ } } + @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 0a1c6ee..71676e5b 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
@@ -52,6 +52,11 @@ } @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 17d5dfd..f95fcfd 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.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.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 @@ 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 @@ .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; - 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); - /* - * 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. - */ - list.getQueryContextSet().add(context); - context.addTVListToSet(Collections.singleton(list)); - workingListForFlushSort.getQueryContextSet().add(context); - tvListQueryMap.put(workingListForFlushSort, workingListForFlushSort.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. + 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; + } } - } else { - if (list.isSorted() || list.getQueryContextSet().isEmpty()) { + + if (copyTimeFilter != null + && !copyTimeFilter.satisfyStartEndTime( + candidate.getMinTime(), candidate.getMaxTime())) { + return tvListQueryMap; + } + + 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 = + candidate.isSorted() || candidate.getQueryContextSet().isEmpty(); + LOGGER.debug( + DataNodeSchemaMessages + .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_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 | <---------+ - * +------------+ - */ - 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()); - } - list.setOwnerQuery(firstQuery); + candidate.getQueryContextSet().add(context); + tvListQueryMap.put(candidate, candidate.rowCount()); - // clone TVList - cloneList = list.clone(); - cloneList.getQueryContextSet().add(context); - tvListQueryMap.put(cloneList, cloneList.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; + } + + // 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 @@ } } - // 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 @@ 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 @@ } // 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 9eb30ec..4f2bb0a 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.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.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 @@ } } + /** + * 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 0ba04ad..55e1899 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 @@ // 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 @@ 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 39629bb..bfefb98 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 @@ 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 @@ 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 035b9a8..608c363 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 @@ 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.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.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 @@ 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 @@ 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 @@ } @Override + @TestOnly public TVList getTvListByColumnIndex( List<Integer> columnIndexList, List<TSDataType> dataTypeList, boolean ignoreAllNullRows) { List<List<Object>> values = new ArrayList<>(); @@ -178,17 +237,16 @@ 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,41 +270,192 @@ 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()); - 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)); - } - // 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<>(columnBitMaps.size()); - for (BitMap bitMap : columnBitMaps) { - cloneColumnBitMaps.add(bitMap == null ? null : bitMap.clone()); - } - cloneList.bitMaps.set(i, cloneColumnBitMaps); - } - } - } - cloneList.timeColDeletedMap = timeColDeletedMap == null ? null : timeColDeletedMap.clone(); + cloneColumnDataTo(cloneList, null); cloneList.materializedValueArrayCounts = Arrays.copyOf(materializedValueArrayCounts, materializedValueArrayCounts.length); cloneList.materializedValueArrayMemCost = materializedValueArrayMemCost; - cloneList.materializedBitmapMemoryCost = materializedBitmapMemoryCost; 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++) { + if (retainedColumns.contains(i)) { + cloneList.values.set(i, new ArrayList<>(values.get(i).size())); + } + } + 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) { + 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)); + } + } + } + + 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)); + } + } + + materializedBitmapMemoryCost = calculateBitmapRamCost(bitMaps, columnsToKeep); + // Refresh the cached per-block cost after releasing columns + refreshArrayMemCostWithoutPrimitiveArrays(); + } + @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning @Override public synchronized void putAlignedValue(long timestamp, Object[] value) { @@ -618,6 +827,27 @@ 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 @@ } } + /* + * 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 @@ 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 @@ 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 @@ } 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 @@ // 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 @@ 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); } - private static long calculateBitmapRamCost(List<List<BitMap>> bitMaps) { + 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, 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 @@ * * @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 @@ + (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 @@ 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 2603790..c4f5174 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 @@ 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 c5bd555..9ea50a0 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 @@ 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 9dbd11c..a7aa2ff 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 @@ 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 @@ 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 61d5c75..c39297d 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 @@ 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 e8c7994..79e66f9 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.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.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.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.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.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 @@ } } + @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 @@ } } + @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 @@ 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 @@ } 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 65cbc0f..14673ff 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
@@ -324,6 +324,11 @@ } @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 6d0cabb..f736250 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 @@ 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 0000000..334e683 --- /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 4383209..e1b402a 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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 fbbd35c..aceb2e2 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.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 @@ } 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ } 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 @@ 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)); + } + } }