Copilot commented on code in PR #17193:
URL: https://github.com/apache/iotdb/pull/17193#discussion_r2791879095
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/flush/MemTableFlushTask.java:
##########
@@ -272,6 +272,7 @@ public void run() {
times = new long[MAX_NUMBER_OF_POINTS_IN_PAGE];
}
writableMemChunk.encode(ioTaskQueue, encodeInfo, times);
+ writableMemChunk.releaseTemporaryTvListForFlush();
Review Comment:
`releaseTemporaryTvListForFlush()` is only called on the normal path. If
`encode(...)` throws (e.g., unchecked exception during iterator/encoding),
`workingListForFlush` will remain referenced on the mem chunk and may retain
large cloned arrays longer than necessary. Wrap the `encode(...)` call in a
`try/finally` so `releaseTemporaryTvListForFlush()` is always executed.
```suggestion
try {
writableMemChunk.encode(ioTaskQueue, encodeInfo, times);
} finally {
writableMemChunk.releaseTemporaryTvListForFlush();
}
```
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/AbstractWritableMemChunk.java:
##########
@@ -198,7 +200,45 @@ public abstract void writeAlignedTablet(
public abstract IMeasurementSchema getSchema();
@Override
- public abstract void sortTvListForFlush();
+ public void sortTvListForFlush() {
+ TVList workingList = getWorkingTVList();
+ if (workingList.isSorted()) {
+ workingListForFlush = workingList;
+ return;
+ }
+
+ /*
+ * Concurrency background:
+ *
+ * A query may start earlier and record the current row count (rows) of
the TVList as its visible range.
+ * After that, new unseq writes may arrive and immediately trigger a
flush, which will sort the TVList.
+ *
+ * During sorting, the underlying indices array of the TVList may be
reordered.
+ * If the query continues to use the previously recorded rows as its upper
bound,
+ * it may convert a logical index to a physical index via the updated
indices array.
+ *
+ * In this case, the converted physical index may exceed the previously
visible
+ * rows range, leading to invalid access or unexpected behavior.
+ *
+ * To avoid this issue, when there are active queries on the working
TVList, we must
+ * clone the times and indices before sorting, so that the flush sort does
not mutate
+ * the data structures that concurrent queries rely on.
+ */
+ boolean needCloneTimesAndIndicesInWorkingTVList;
+ workingList.lockQueryList();
+ try {
+ needCloneTimesAndIndicesInWorkingTVList =
!workingList.getQueryContextSet().isEmpty();
+ } finally {
+ workingList.unlockQueryList();
+ }
+ workingListForFlush =
+ needCloneTimesAndIndicesInWorkingTVList ?
workingList.cloneForFlushSort() : workingList;
+ workingListForFlush.sort();
+ }
+
Review Comment:
This method overrides [IWritableMemChunk.releaseTemporaryTvListForFlush](1);
it is advisable to add an Override annotation.
```suggestion
@Override
```
--
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]
