jt2594838 commented on code in PR #17591:
URL: https://github.com/apache/iotdb/pull/17591#discussion_r3709270118
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/utils/ResourceByPathUtils.java:
##########
@@ -221,39 +240,68 @@ protected Map<TVList, Integer> prepareTvListMapForQuery(
*/
LOGGER.debug(
"Working MemTable - clone mutable TVList and replace old TVList
in working MemTable");
- 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);
- // clone TVList
- cloneList = list.clone();
- cloneList.getQueryContextSet().add(context);
- tvListQueryMap.put(cloneList, cloneList.rowCount());
+ // Synchronize on memChunk to prevent concurrent clone-and-replace
races.
+ // Another query entering this same branch could call
memChunk.setWorkingTVList(cloneList)
+ // between our check above and the clone below. By locking on
memChunk we ensure only one
+ // query performs the clone-and-swap at a time.
+ synchronized (memChunk) {
+ // We re-fetch via getWorkingTVList() inside the lock because the
working TVList may
+ // have already been replaced by a concurrent query that acquired
the lock first —
+ // operating on the stale outer 'list' would clone an
already-detached copy.
+ TVList workingTVList = memChunk.getWorkingTVList();
+ Set<Integer> columnsToClone =
getAccessedColumnsForQuery(workingTVList);
+ listRamInfo =
+ (columnsToClone == null)
+ ? workingTVList.calculateRamSize()
+ : ((AlignedTVList)
workingTVList).calculateRamSize(columnsToClone);
+
+ // reserve query memory
+ QueryContext firstQuery =
workingTVList.getQueryContextSet().iterator().next();
+ if (firstQuery instanceof FragmentInstanceContext) {
+ MemoryReservationManager memoryReservationManager =
+ ((FragmentInstanceContext)
firstQuery).getMemoryReservationContext();
+
memoryReservationManager.reserveMemoryCumulatively(listRamInfo.getRamSize());
+ workingTVList.setReservedMemoryBytes(listRamInfo.getRamSize());
+ }
+ workingTVList.setOwnerQuery(firstQuery);
+
+ // clone TVList
+ TVList cloneList =
+ (columnsToClone == null)
+ ? workingTVList.clone()
+ : ((AlignedTVList) workingTVList).clone(columnsToClone);
+
+ cloneList.getQueryContextSet().add(context);
+ tvListQueryMap.put(cloneList, cloneList.rowCount());
+ if (columnIndexList != null && context instanceof
FragmentInstanceContext) {
+ ((FragmentInstanceContext)
context).putAccessedColumns(cloneList, columnIndexList);
+ }
+
+ memChunk.setWorkingTVList(cloneList);
Review Comment:
The memChunk after the query seems to only contain the queried columns.
Will other columns be missing during flush?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/utils/datastructure/AlignedTVList.java:
##########
@@ -635,6 +613,63 @@ protected Object cloneValue(TSDataType type, Object value)
{
}
}
+ private void cloneColumnDataTo(AlignedTVList cloneList, Set<Integer>
columnsToClone) {
+ boolean cloneAllColumns = columnsToClone == null;
+ System.arraycopy(
+ memoryBinaryChunkSize, 0, cloneList.memoryBinaryChunkSize, 0,
dataTypes.size());
+ for (int i = 0; i < values.size(); i++) {
+ List<Object> columnValues = values.get(i);
+ if (columnValues == null) {
+ continue;
+ }
+ boolean shouldCloneColumn = cloneAllColumns ||
columnsToClone.contains(i);
+ if (!shouldCloneColumn) {
+ cloneList.values.set(i, columnValues);
+ values.set(i, null);
+ if (bitMaps != null && bitMaps.get(i) != null) {
+ cloneList.ensureBitMapsInitialized();
+ cloneList.bitMaps.set(i, bitMaps.get(i));
+ bitMaps.set(i, null);
+ }
+ memoryBinaryChunkSize[i] = 0;
+ continue;
+ }
Review Comment:
Better to add some comments (in this method and the caller) to explain that
columns not in columnsToClone will be moved from the current list to the new
list.
The semantics cannot be read from the method signature.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
