You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by sh...@apache.org on 2026/06/10 10:00:31 UTC
(iotdb) 03/11: FileReaderManager
This is an automated email from the ASF dual-hosted git repository.
shuwenwei pushed a commit to branch read_tsfile_table_function
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 194e609befce9dbc28f1b73e6223f9f2f80ca69a
Author: shuwenwei <s1...@gmail.com>
AuthorDate: Thu Jun 4 11:11:34 2026 +0800
FileReaderManager
---
.../fragment/FragmentInstanceContext.java | 16 +++-
.../execution/fragment/QueryContext.java | 4 +
.../execution/operator/source/FileLoaderUtils.java | 10 ++-
.../OrderedExternalTsFileTableScanOperator.java | 6 +-
.../ConvertSchemaPredicateToFilterVisitor.java | 8 +-
.../db/storageengine/buffer/BloomFilterCache.java | 24 +++++-
.../iotdb/db/storageengine/buffer/ChunkCache.java | 31 ++++++--
.../buffer/TimeSeriesMetadataCache.java | 27 ++++++-
.../dataregion/read/control/FileReaderManager.java | 88 ++++++++++++++++++++++
.../read/reader/chunk/DiskAlignedChunkLoader.java | 6 +-
.../read/reader/chunk/DiskChunkLoader.java | 6 +-
11 files changed, 196 insertions(+), 30 deletions(-)
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 56214b971f5..654df2301d8 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
@@ -122,6 +122,7 @@ public class FragmentInstanceContext extends QueryContext {
private Map<IDeviceID, DeviceContext> devicePathsToContext;
private List<String> externalTsFilePaths;
+ private List<TsFileResource> externalTsFileResources;
// Shared by all scan operators in this fragment instance to avoid memory problem
protected IQueryDataSource sharedQueryDataSource;
@@ -236,6 +237,11 @@ public class FragmentInstanceContext extends QueryContext {
this.queryDataSourceType = queryDataSourceType;
}
+ @Override
+ public boolean isExternalTsFileScan() {
+ return queryDataSourceType == QueryDataSourceType.EXTERNAL_TSFILE_SCAN;
+ }
+
@TestOnly
public static FragmentInstanceContext createFragmentInstanceContext(
FragmentInstanceId id, FragmentInstanceStateMachine stateMachine) {
@@ -802,7 +808,7 @@ public class FragmentInstanceContext extends QueryContext {
return true;
}
- List<TsFileResource> externalTsFileResources = new ArrayList<>(externalTsFilePaths.size());
+ externalTsFileResources = new ArrayList<>(externalTsFilePaths.size());
for (String externalTsFilePath : externalTsFilePaths) {
TsFileResource resource =
new TsFileResource(new File(externalTsFilePath), TsFileResourceStatus.NORMAL);
@@ -820,6 +826,7 @@ public class FragmentInstanceContext extends QueryContext {
resource.setTimeIndex(new FileTimeIndex(Long.MIN_VALUE, Long.MAX_VALUE));
}
externalTsFileResources.add(resource);
+ FileReaderManager.getInstance().increaseExternalFileReaderReference(externalTsFilePath);
}
this.sharedQueryDataSource =
@@ -1075,6 +1082,13 @@ public class FragmentInstanceContext extends QueryContext {
unClosedFilePaths = null;
}
+ if (externalTsFileResources != null) {
+ for (TsFileResource tsFile : externalTsFileResources) {
+ FileReaderManager.getInstance().decreaseExternalFileReaderReference(tsFile.getTsFilePath());
+ }
+ externalTsFileResources = null;
+ }
+
// release TVList/AlignedTVList owned by current query
releaseTVListOwnedByQuery();
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 d3edbe2e0d0..f89d8ddefe5 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
@@ -97,6 +97,10 @@ public class QueryContext {
this.timeout = timeout;
}
+ public boolean isExternalTsFileScan() {
+ return false;
+ }
+
// Only used for query with table data(Tree view is not included)
public boolean collectTable(String table) {
// In the current version (2025.08.14), there is only one table under one FI
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java
index 254c061187a..dcc38e49a61 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/FileLoaderUtils.java
@@ -110,7 +110,8 @@ public class FileLoaderUtils {
context.ignoreNotExistsDevice()
|| resource.getTimeIndexType() == ITimeIndex.FILE_TIME_INDEX_TYPE,
context.isDebug(),
- context);
+ context,
+ context.isExternalTsFileScan());
if (timeSeriesMetadata != null) {
long t2 = System.nanoTime();
List<ModEntry> pathModifications =
@@ -297,6 +298,7 @@ public class FileLoaderUtils {
boolean isDebug = context.isDebug();
String filePath = resource.getTsFilePath();
IDeviceID deviceId = alignedPath.getDeviceId();
+ boolean isExternalTsFile = context.isExternalTsFileScan();
// when resource.getTimeIndexType() == 1, TsFileResource.timeIndexType is deviceTimeIndex
// we should not ignore the non-exist of device in TsFileMetadata
@@ -308,7 +310,8 @@ public class FileLoaderUtils {
context.ignoreNotExistsDevice()
|| resource.getTimeIndexType() == ITimeIndex.FILE_TIME_INDEX_TYPE,
isDebug,
- context);
+ context,
+ isExternalTsFile);
if (timeColumn != null) {
// only need time column, like count_time aggregation
if (valueMeasurementList.isEmpty()) {
@@ -337,7 +340,8 @@ public class FileLoaderUtils {
context.ignoreNotExistsDevice()
|| resource.getTimeIndexType() == ITimeIndex.FILE_TIME_INDEX_TYPE,
isDebug,
- context);
+ context,
+ isExternalTsFile);
exist = (exist || (valueColumn != null));
valueTimeSeriesMetadataList.add(valueColumn);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/OrderedExternalTsFileTableScanOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/OrderedExternalTsFileTableScanOperator.java
index 3b84e1a1908..2b1fdf561df 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/OrderedExternalTsFileTableScanOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/OrderedExternalTsFileTableScanOperator.java
@@ -140,9 +140,9 @@ public class OrderedExternalTsFileTableScanOperator extends AbstractTableScanOpe
reader,
tableName,
((OperatorContext) operatorContext)
- .getInstanceContext()
- .getQueryStatistics()
- .getLoadTimeSeriesMetadataActualIOSize()
+ .getInstanceContext()
+ .getQueryStatistics()
+ .getLoadTimeSeriesMetadataActualIOSize()
::addAndGet);
while (deviceIterator.hasNext()) {
IDeviceID deviceID = deviceIterator.next();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/predicate/schema/ConvertSchemaPredicateToFilterVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/predicate/schema/ConvertSchemaPredicateToFilterVisitor.java
index 7bcf09fd73a..a7781bc1e64 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/predicate/schema/ConvertSchemaPredicateToFilterVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/predicate/schema/ConvertSchemaPredicateToFilterVisitor.java
@@ -239,15 +239,11 @@ public class ConvertSchemaPredicateToFilterVisitor
final BetweenPredicate node, final Context context) {
final SchemaFilter lowerBoundFilter =
new ComparisonExpression(
- ComparisonExpression.Operator.LESS_THAN_OR_EQUAL,
- node.getMin(),
- node.getValue())
+ ComparisonExpression.Operator.LESS_THAN_OR_EQUAL, node.getMin(), node.getValue())
.accept(this, context);
final SchemaFilter upperBoundFilter =
new ComparisonExpression(
- ComparisonExpression.Operator.LESS_THAN_OR_EQUAL,
- node.getValue(),
- node.getMax())
+ ComparisonExpression.Operator.LESS_THAN_OR_EQUAL, node.getValue(), node.getMax())
.accept(this, context);
if (Objects.isNull(lowerBoundFilter) || Objects.isNull(upperBoundFilter)) {
return null;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/BloomFilterCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/BloomFilterCache.java
index 380583efbdb..eb0ae5554a5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/BloomFilterCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/BloomFilterCache.java
@@ -101,9 +101,20 @@ public class BloomFilterCache {
LongConsumer cacheHitAdder,
LongConsumer cacheMissAdder)
throws IOException {
- BloomFilterLoader loader = new BloomFilterLoader(ioSizeRecorder);
+ return get(key, debug, ioSizeRecorder, cacheHitAdder, cacheMissAdder, false);
+ }
+
+ public BloomFilter get(
+ BloomFilterCacheKey key,
+ boolean debug,
+ LongConsumer ioSizeRecorder,
+ LongConsumer cacheHitAdder,
+ LongConsumer cacheMissAdder,
+ boolean externalTsFile)
+ throws IOException {
+ BloomFilterLoader loader = new BloomFilterLoader(ioSizeRecorder, externalTsFile);
try {
- if (!CACHE_ENABLE) {
+ if (!CACHE_ENABLE || externalTsFile) {
return loader.apply(key);
}
@@ -203,9 +214,15 @@ public class BloomFilterCache {
private boolean cacheMiss = false;
private final LongConsumer ioSizeRecorder;
+ private final boolean externalTsFile;
private BloomFilterLoader(LongConsumer ioSizeRecorder) {
+ this(ioSizeRecorder, false);
+ }
+
+ private BloomFilterLoader(LongConsumer ioSizeRecorder, boolean externalTsFile) {
this.ioSizeRecorder = ioSizeRecorder;
+ this.externalTsFile = externalTsFile;
}
@Override
@@ -218,7 +235,8 @@ public class BloomFilterCache {
bloomFilterCacheKey.filePath,
bloomFilterCacheKey.tsFileID,
true,
- ioSizeRecorder);
+ ioSizeRecorder,
+ externalTsFile);
return reader.readBloomFilter(ioSizeRecorder);
} catch (IOException e) {
throw new IoTDBIORuntimeException(e);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/ChunkCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/ChunkCache.java
index d58c488efe9..08d750328d5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/ChunkCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/ChunkCache.java
@@ -121,7 +121,8 @@ public class ChunkCache {
false,
emptyConsumer,
emptyConsumer,
- emptyConsumer);
+ emptyConsumer,
+ false);
}
public Chunk get(
@@ -130,6 +131,16 @@ public class ChunkCache {
Statistics chunkStatistic,
QueryContext queryContext)
throws IOException {
+ return get(chunkCacheKey, timeRangeList, chunkStatistic, queryContext, false);
+ }
+
+ public Chunk get(
+ ChunkCacheKey chunkCacheKey,
+ List<TimeRange> timeRangeList,
+ Statistics chunkStatistic,
+ QueryContext queryContext,
+ boolean externalTsFile)
+ throws IOException {
LongConsumer ioSizeRecorder =
queryContext.getQueryStatistics().getLoadChunkActualIOSize()::addAndGet;
LongConsumer cacheHitAdder =
@@ -143,7 +154,8 @@ public class ChunkCache {
queryContext.isDebug(),
ioSizeRecorder,
cacheHitAdder,
- cacheMissAdder);
+ cacheMissAdder,
+ externalTsFile);
}
private Chunk get(
@@ -153,12 +165,13 @@ public class ChunkCache {
boolean debug,
LongConsumer ioSizeRecorder,
LongConsumer cacheHitAdder,
- LongConsumer cacheMissAdder)
+ LongConsumer cacheMissAdder,
+ boolean externalTsFile)
throws IOException {
long startTime = System.nanoTime();
- ChunkLoader chunkLoader = new ChunkLoader(ioSizeRecorder);
+ ChunkLoader chunkLoader = new ChunkLoader(ioSizeRecorder, externalTsFile);
try {
- if (!CACHE_ENABLE) {
+ if (!CACHE_ENABLE || externalTsFile) {
Chunk chunk = chunkLoader.apply(chunkCacheKey);
return constructChunk(chunk, timeRangeList, chunkStatistic);
}
@@ -297,9 +310,15 @@ public class ChunkCache {
private boolean cacheMiss = false;
private final LongConsumer ioSizeRecorder;
+ private final boolean externalTsFile;
private ChunkLoader(LongConsumer ioSizeRecorder) {
+ this(ioSizeRecorder, false);
+ }
+
+ private ChunkLoader(LongConsumer ioSizeRecorder, boolean externalTsFile) {
this.ioSizeRecorder = ioSizeRecorder;
+ this.externalTsFile = externalTsFile;
}
@Override
@@ -310,7 +329,7 @@ public class ChunkCache {
cacheMiss = true;
TsFileSequenceReader reader =
FileReaderManager.getInstance()
- .get(key.getFilePath(), key.tsFileID, key.closed, ioSizeRecorder);
+ .get(key.getFilePath(), key.tsFileID, key.closed, ioSizeRecorder, externalTsFile);
Chunk chunk = reader.readMemChunk(key.offsetOfChunkHeader, ioSizeRecorder);
// to save memory footprint, we don't save measurementId in ChunkHeader of Chunk
chunk.getHeader().setMeasurementID(null);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCache.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCache.java
index cc3b2ee5aff..ba26db0f1d7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCache.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/buffer/TimeSeriesMetadataCache.java
@@ -119,6 +119,19 @@ public class TimeSeriesMetadataCache {
boolean debug,
QueryContext queryContext)
throws IOException {
+ return get(filePath, key, allSensors, ignoreNotExists, debug, queryContext, false);
+ }
+
+ @SuppressWarnings({"squid:S1860", "squid:S6541", "squid:S3776"}) // Suppress synchronize warning
+ public TimeseriesMetadata get(
+ String filePath,
+ TimeSeriesMetadataCacheKey key,
+ Set<String> allSensors,
+ boolean ignoreNotExists,
+ boolean debug,
+ QueryContext queryContext,
+ boolean externalTsFile)
+ throws IOException {
long startTime = System.nanoTime();
long loadBloomFilterTime = 0;
LongConsumer timeSeriesMetadataIoSizeRecorder =
@@ -127,14 +140,14 @@ public class TimeSeriesMetadataCache {
queryContext.getQueryStatistics().getLoadBloomFilterActualIOSize()::addAndGet;
boolean cacheHit = true;
try {
- if (!CACHE_ENABLE) {
+ if (!CACHE_ENABLE || externalTsFile) {
String deviceStringFormat = key.device.toString();
cacheHit = false;
// bloom filter part
TsFileSequenceReader reader =
FileReaderManager.getInstance()
- .get(filePath, key.tsFileID, true, bloomFilterIoSizeRecorder);
+ .get(filePath, key.tsFileID, true, bloomFilterIoSizeRecorder, externalTsFile);
BloomFilter bloomFilter = reader.readBloomFilter(bloomFilterIoSizeRecorder);
queryContext.getQueryStatistics().getLoadBloomFilterFromDiskCount().incrementAndGet();
if (bloomFilter != null
@@ -182,7 +195,8 @@ public class TimeSeriesMetadataCache {
queryContext.getQueryStatistics().getLoadBloomFilterFromCacheCount()
::addAndGet,
queryContext.getQueryStatistics().getLoadBloomFilterFromDiskCount()
- ::addAndGet);
+ ::addAndGet,
+ externalTsFile);
if (bloomFilter != null
&& !bloomFilter.contains(
deviceStringFormat + TsFileConstant.PATH_SEPARATOR + key.measurement)) {
@@ -196,7 +210,12 @@ public class TimeSeriesMetadataCache {
loadBloomFilterTime = System.nanoTime() - loadBloomFilterStartTime;
TsFileSequenceReader reader =
FileReaderManager.getInstance()
- .get(filePath, key.tsFileID, true, timeSeriesMetadataIoSizeRecorder);
+ .get(
+ filePath,
+ key.tsFileID,
+ true,
+ timeSeriesMetadataIoSizeRecorder,
+ externalTsFile);
List<TimeseriesMetadata> timeSeriesMetadataList =
reader.readTimeseriesMetadata(
key.device,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManager.java
index 3952c7480a5..d2dac0b775b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManager.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/control/FileReaderManager.java
@@ -80,11 +80,19 @@ public class FileReaderManager {
*/
private Map<TsFileID, AtomicInteger> unclosedReferenceMap;
+ /** External TsFile readers. The key is the file path. */
+ private Map<String, TsFileSequenceReader> externalFileReaderMap;
+
+ /** Reference count of external TsFile readers. */
+ private Map<String, AtomicInteger> externalReferenceMap;
+
private FileReaderManager() {
closedFileReaderMap = new ConcurrentHashMap<>();
unclosedFileReaderMap = new ConcurrentHashMap<>();
closedReferenceMap = new ConcurrentHashMap<>();
unclosedReferenceMap = new ConcurrentHashMap<>();
+ externalFileReaderMap = new ConcurrentHashMap<>();
+ externalReferenceMap = new ConcurrentHashMap<>();
}
public static FileReaderManager getInstance() {
@@ -137,6 +145,20 @@ public class FileReaderManager {
public synchronized TsFileSequenceReader get(
String filePath, TsFileID tsFileID, boolean isClosed, LongConsumer ioSizeRecorder)
throws IOException {
+ return get(filePath, tsFileID, isClosed, ioSizeRecorder, false);
+ }
+
+ @SuppressWarnings("squid:S2095")
+ public synchronized TsFileSequenceReader get(
+ String filePath,
+ TsFileID tsFileID,
+ boolean isClosed,
+ LongConsumer ioSizeRecorder,
+ boolean isExternalTsFile)
+ throws IOException {
+ if (isExternalTsFile) {
+ return getExternalTsFileReader(filePath, ioSizeRecorder);
+ }
Map<TsFileID, TsFileSequenceReader> readerMap =
!isClosed ? unclosedFileReaderMap : closedFileReaderMap;
@@ -170,6 +192,25 @@ public class FileReaderManager {
return readerMap.get(tsFileID);
}
+ private TsFileSequenceReader getExternalTsFileReader(String filePath, LongConsumer ioSizeRecorder)
+ throws IOException {
+ TsFileSequenceReader reader = externalFileReaderMap.get(filePath);
+ if (reader == null) {
+ int currentOpenedReaderCount = externalFileReaderMap.size();
+ if (currentOpenedReaderCount >= MAX_CACHED_FILE_SIZE
+ && (currentOpenedReaderCount % PRINT_INTERVAL == 0)) {
+ logger.warn(StorageEngineMessages.QUERY_OPENED_FILES, externalFileReaderMap.size());
+ }
+ reader =
+ new TsFileSequenceReader(
+ filePath,
+ ioSizeRecorder,
+ EncryptDBUtils.getFirstEncryptParamFromTSFilePath(filePath));
+ externalFileReaderMap.put(filePath, reader);
+ }
+ return reader;
+ }
+
/**
* Increase the reference count of the reader specified by filePath. Only when the reference count
* of a reader equals zero, the reader can be closed and removed.
@@ -189,6 +230,10 @@ public class FileReaderManager {
}
}
+ public synchronized void increaseExternalFileReaderReference(String filePath) {
+ externalReferenceMap.computeIfAbsent(filePath, k -> new AtomicInteger()).getAndIncrement();
+ }
+
/**
* Decrease the reference count of the reader specified by filePath. This method is latch-free.
* Only when the reference count of a reader equals zero, the reader can be closed and removed.
@@ -207,6 +252,38 @@ public class FileReaderManager {
tsFile.readUnlock();
}
+ public synchronized void decreaseExternalFileReaderReference(String filePath) {
+ AtomicInteger reference = externalReferenceMap.get(filePath);
+ if (reference != null && reference.decrementAndGet() == 0) {
+ closeUnUsedExternalReaderAndRemoveRef(filePath);
+ }
+ }
+
+ private void closeUnUsedExternalReaderAndRemoveRef(String readerKey) {
+ synchronized (this) {
+ AtomicInteger reference = externalReferenceMap.get(readerKey);
+ if (reference != null && reference.get() != 0) {
+ return;
+ }
+
+ TsFileSequenceReader reader = externalFileReaderMap.get(readerKey);
+ if (reader != null) {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ logger.error(
+ StorageEngineMessages.CANNOT_CLOSE_TSFILE_SEQUENCE_READER, reader.getFileName(), e);
+ }
+ }
+ externalFileReaderMap.remove(readerKey);
+ externalReferenceMap.remove(readerKey);
+ if (resourceLogger.isDebugEnabled()) {
+ resourceLogger.debug(
+ "{} externalTsFileReader is closed because of no reference.", readerKey);
+ }
+ }
+ }
+
private void closeUnUsedReaderAndRemoveRef(
String tsFilePath, TsFileID tsFileID, boolean isClosed) {
Map<TsFileID, TsFileSequenceReader> readerMap =
@@ -263,6 +340,17 @@ public class FileReaderManager {
unclosedReferenceMap.remove(entry.getKey());
iterator.remove();
}
+ Iterator<Map.Entry<String, TsFileSequenceReader>> externalIterator =
+ externalFileReaderMap.entrySet().iterator();
+ while (externalIterator.hasNext()) {
+ Map.Entry<String, TsFileSequenceReader> entry = externalIterator.next();
+ entry.getValue().close();
+ if (resourceLogger.isDebugEnabled()) {
+ resourceLogger.debug("{} externalTsFileReader is closed.", entry.getKey());
+ }
+ externalReferenceMap.remove(entry.getKey());
+ externalIterator.remove();
+ }
}
/** This method is only for unit tests. */
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java
index 7c1e9d12635..2ea8ec0c6ba 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskAlignedChunkLoader.java
@@ -92,7 +92,8 @@ public class DiskAlignedChunkLoader implements IChunkLoader {
resource.isClosed()),
timeChunkMetadata.getDeleteIntervalList(),
timeChunkMetadata.getStatistics(),
- context);
+ context,
+ context.isExternalTsFileScan());
List<Chunk> valueChunkList = new ArrayList<>();
for (IChunkMetadata valueChunkMetadata : alignedChunkMetadata.getValueChunkMetadataList()) {
Chunk chunk =
@@ -107,7 +108,8 @@ public class DiskAlignedChunkLoader implements IChunkLoader {
resource.isClosed()),
valueChunkMetadata.getDeleteIntervalList(),
valueChunkMetadata.getStatistics(),
- context);
+ context,
+ context.isExternalTsFileScan());
final TsFileID tsFileID = getTsFileID();
if (chunk != null
&& tsFileID.regionId > 0
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskChunkLoader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskChunkLoader.java
index 7d259a9e505..73355bed0c8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskChunkLoader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/read/reader/chunk/DiskChunkLoader.java
@@ -64,7 +64,8 @@ public class DiskChunkLoader implements IChunkLoader {
resource.isClosed()),
chunkMetaData.getDeleteIntervalList(),
chunkMetaData.getStatistics(),
- context);
+ context,
+ context.isExternalTsFileScan());
}
@Override
@@ -87,7 +88,8 @@ public class DiskChunkLoader implements IChunkLoader {
resource.isClosed()),
chunkMetaData.getDeleteIntervalList(),
chunkMetaData.getStatistics(),
- context);
+ context,
+ context.isExternalTsFileScan());
byte chunkType = chunk.getHeader().getChunkType();
if (chunkType != MetaMarker.CHUNK_HEADER
&& chunkType != MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER) {