You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ja...@apache.org on 2021/12/18 01:14:07 UTC
[iotdb] branch rel/0.12 updated: [To rel/0.12] [IOTDB-2101] Reduce the memory footprint of QueryDataSource (#4572)
This is an automated email from the ASF dual-hosted git repository.
jackietien pushed a commit to branch rel/0.12
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/0.12 by this push:
new dfe6918 [To rel/0.12] [IOTDB-2101] Reduce the memory footprint of QueryDataSource (#4572)
dfe6918 is described below
commit dfe691854fe025bfe9ae6d30013cec967de9411c
Author: liuminghui233 <36...@users.noreply.github.com>
AuthorDate: Sat Dec 18 09:13:27 2021 +0800
[To rel/0.12] [IOTDB-2101] Reduce the memory footprint of QueryDataSource (#4572)
---
.../org/apache/iotdb/cluster/common/TestUtils.java | 3 +-
.../org/apache/iotdb/db/engine/StorageEngine.java | 48 ++--
.../db/engine/querycontext/QueryDataSource.java | 55 +---
.../engine/storagegroup/StorageGroupProcessor.java | 99 ++-----
.../db/engine/storagegroup/TsFileProcessor.java | 235 ++++++-----------
.../db/engine/storagegroup/TsFileResource.java | 293 +++++++++++----------
.../org/apache/iotdb/db/metadata/MManager.java | 8 +-
.../java/org/apache/iotdb/db/metadata/MTree.java | 2 +-
.../iotdb/db/query/control/QueryFileManager.java | 33 +--
.../db/query/control/QueryResourceManager.java | 131 ++++-----
.../query/dataset/groupby/GroupByFillDataSet.java | 69 +++--
.../groupby/GroupByWithValueFilterDataSet.java | 13 +-
.../groupby/GroupByWithoutValueFilterDataSet.java | 5 +-
.../db/query/executor/AggregationExecutor.java | 7 +-
.../iotdb/db/query/executor/FillQueryExecutor.java | 53 +++-
.../iotdb/db/query/executor/LastQueryExecutor.java | 6 +-
.../db/query/executor/RawDataQueryExecutor.java | 6 +-
.../iotdb/db/query/executor/fill/LinearFill.java | 6 +-
.../iotdb/db/query/executor/fill/PreviousFill.java | 3 +-
.../chunk/metadata/MemChunkMetadataLoader.java | 4 +-
.../query/timegenerator/ServerTimeGenerator.java | 8 +-
.../org/apache/iotdb/db/utils/FileLoaderUtils.java | 2 +-
.../java/org/apache/iotdb/db/utils/QueryUtils.java | 17 --
.../iotdb/db/engine/merge/ConcurrentMergeTest.java | 8 +-
.../engine/modification/DeletionFileNodeTest.java | 22 +-
.../storagegroup/StorageGroupProcessorTest.java | 68 +++--
.../iotdb/db/engine/storagegroup/TTLTest.java | 15 +-
.../engine/storagegroup/TsFileProcessorTest.java | 118 ++++++---
.../iotdb/db/integration/IoTDBAlignByDeviceIT.java | 2 +-
.../iotdb/db/qp/physical/InsertTabletPlanTest.java | 4 -
30 files changed, 643 insertions(+), 700 deletions(-)
diff --git a/cluster/src/test/java/org/apache/iotdb/cluster/common/TestUtils.java b/cluster/src/test/java/org/apache/iotdb/cluster/common/TestUtils.java
index 07861b7..57ae69a 100644
--- a/cluster/src/test/java/org/apache/iotdb/cluster/common/TestUtils.java
+++ b/cluster/src/test/java/org/apache/iotdb/cluster/common/TestUtils.java
@@ -358,7 +358,7 @@ public class TestUtils {
/**
* The TsFileResource's path should be consist with the {@link
- * org.apache.iotdb.db.utils.FilePathUtils#splitTsFilePath(TsFileResource)}
+ * org.apache.iotdb.tsfile.utils.FilePathUtils#splitTsFilePath(String)}
*/
public static List<TsFileResource> prepareTsFileResources(
int sgNum, int fileNum, int seriesNum, int ptNum, boolean asHardLink)
@@ -410,6 +410,7 @@ public class TestUtils {
resource.updateEndTime(TestUtils.getTestSg(sgNum), (i + 1) * ptNum - 1);
resource.setMaxPlanIndex(i);
resource.setMinPlanIndex(i);
+ resource.setClosed(true);
resource.serialize();
ret.add(resource);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
index a9b294a..33d48ca 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/StorageEngine.java
@@ -27,7 +27,6 @@ import org.apache.iotdb.db.engine.flush.CloseFileListener;
import org.apache.iotdb.db.engine.flush.FlushListener;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy;
import org.apache.iotdb.db.engine.flush.TsFileFlushPolicy.DirectFlushPolicy;
-import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor.TimePartitionFilter;
import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
@@ -47,7 +46,7 @@ import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.qp.physical.crud.InsertRowsOfOneDevicePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.query.context.QueryContext;
-import org.apache.iotdb.db.query.control.QueryFileManager;
+import org.apache.iotdb.db.query.control.QueryResourceManager;
import org.apache.iotdb.db.rescon.SystemInfo;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.IoTDB;
@@ -57,7 +56,7 @@ import org.apache.iotdb.db.utils.UpgradeUtils;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
-import org.apache.iotdb.tsfile.read.expression.impl.SingleSeriesExpression;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.utils.FilePathUtils;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -809,27 +808,6 @@ public class StorageEngine implements IService {
}
}
- /** query data. */
- public QueryDataSource query(
- SingleSeriesExpression seriesExpression,
- QueryContext context,
- QueryFileManager filePathsManager)
- throws StorageEngineException, QueryProcessException {
- PartialPath fullPath = (PartialPath) seriesExpression.getSeriesPath();
- PartialPath deviceId = fullPath.getDevicePath();
- StorageGroupProcessor storageGroupProcessor = getProcessor(deviceId);
- return storageGroupProcessor.query(
- fullPath, context, filePathsManager, seriesExpression.getFilter());
- }
-
- public QueryDataSource getAllQueryDataSource(SingleSeriesExpression seriesExpression)
- throws StorageEngineException, QueryProcessException {
- PartialPath fullPath = (PartialPath) seriesExpression.getSeriesPath();
- PartialPath deviceId = fullPath.getDevicePath();
- StorageGroupProcessor storageGroupProcessor = getProcessor(deviceId);
- return storageGroupProcessor.getAllQueryDataSource(seriesExpression.getFilter());
- }
-
public String getStorageGroupPath(PartialPath path) throws StorageEngineException {
PartialPath deviceId = path.getDevicePath();
StorageGroupProcessor storageGroupProcessor = getProcessor(deviceId);
@@ -1077,6 +1055,28 @@ public class StorageEngine implements IService {
return list;
}
+ /**
+ * get all merge lock of the storage group processor related to the query and init QueryDataSource
+ */
+ public List<StorageGroupProcessor> mergeLockAndInitQueryDataSource(
+ List<PartialPath> pathList, QueryContext context, Filter timeFilter)
+ throws StorageEngineException, QueryProcessException {
+ Map<StorageGroupProcessor, List<PartialPath>> map = new HashMap<>();
+ for (PartialPath path : pathList) {
+ map.computeIfAbsent(getProcessor(path.getDevicePath()), key -> new ArrayList<>()).add(path);
+ }
+ List<StorageGroupProcessor> list =
+ map.keySet().stream()
+ .sorted(Comparator.comparing(StorageGroupProcessor::getVirtualStorageGroupId))
+ .collect(Collectors.toList());
+ list.forEach(StorageGroupProcessor::readLock);
+
+ // init QueryDataSource
+ QueryResourceManager.getInstance().initQueryDataSource(map, context, timeFilter);
+
+ return list;
+ }
+
/** unlock all merge lock of the storage group processor related to the query */
public void mergeUnLock(List<StorageGroupProcessor> list) {
list.forEach(StorageGroupProcessor::readUnlock);
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/querycontext/QueryDataSource.java b/server/src/main/java/org/apache/iotdb/db/engine/querycontext/QueryDataSource.java
index b481edd..ab65286 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/querycontext/QueryDataSource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/querycontext/QueryDataSource.java
@@ -27,17 +27,23 @@ import org.apache.iotdb.tsfile.read.filter.operator.AndFilter;
import java.util.List;
public class QueryDataSource {
+
+ /**
+ * TsFileResources used by query job.
+ *
+ * <p>Note: Sequences under the same virtual storage group share two lists of TsFileResources (seq
+ * and unseq).
+ */
private final List<TsFileResource> seqResources;
+
private final List<TsFileResource> unseqResources;
- private TsFileResource unclosedSeqResource;
- private TsFileResource unclosedUnseqResource;
+ /* The traversal order of unseqResources (different for each device) */
+ private int[] unSeqFileOrderIndex;
/** data older than currentTime - dataTTL should be ignored. */
private long dataTTL = Long.MAX_VALUE;
- private int[] unSeqFileOrderIndex;
-
public QueryDataSource(List<TsFileResource> seqResources, List<TsFileResource> unseqResources) {
this.seqResources = seqResources;
this.unseqResources = unseqResources;
@@ -51,30 +57,10 @@ public class QueryDataSource {
return unseqResources;
}
- public TsFileResource getUnclosedSeqResource() {
- return unclosedSeqResource;
- }
-
- public TsFileResource getUnclosedUnseqResource() {
- return unclosedUnseqResource;
- }
-
public long getDataTTL() {
return dataTTL;
}
- public int[] getUnSeqFileOrderIndex() {
- return unSeqFileOrderIndex;
- }
-
- public void setUnclosedSeqResource(TsFileResource unclosedSeqResource) {
- this.unclosedSeqResource = unclosedSeqResource;
- }
-
- public void setUnclosedUnseqResource(TsFileResource unclosedUnseqResource) {
- this.unclosedUnseqResource = unclosedUnseqResource;
- }
-
public void setDataTTL(long dataTTL) {
this.dataTTL = dataTTL;
}
@@ -98,8 +84,6 @@ public class QueryDataSource {
public TsFileResource getSeqResourceByIndex(int curIndex) {
if (curIndex < seqResources.size()) {
return seqResources.get(curIndex);
- } else if (curIndex == seqResources.size()) {
- return unclosedSeqResource;
}
return null;
}
@@ -108,32 +92,19 @@ public class QueryDataSource {
int actualIndex = unSeqFileOrderIndex[curIndex];
if (actualIndex < unseqResources.size()) {
return unseqResources.get(actualIndex);
- } else if (actualIndex == unseqResources.size()) {
- return unclosedUnseqResource;
}
return null;
}
public boolean hasNextSeqResource(int curIndex, boolean ascending) {
- if (ascending) {
- return unclosedSeqResource == null
- ? curIndex < seqResources.size()
- : curIndex <= seqResources.size();
- }
- return curIndex >= 0;
+ return ascending ? curIndex < seqResources.size() : curIndex >= 0;
}
public boolean hasNextUnseqResource(int curIndex) {
- return unclosedUnseqResource == null
- ? curIndex < unseqResources.size()
- : curIndex <= unseqResources.size();
+ return curIndex < unseqResources.size();
}
public int getSeqResourcesSize() {
- return seqResources.size() + (unclosedSeqResource == null ? 0 : 1);
- }
-
- public int getUnseqResourcesSize() {
- return unseqResources.size() + (unclosedUnseqResource == null ? 0 : 1);
+ return seqResources.size();
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
index 8da66de..d0b21fd 100755
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessor.java
@@ -77,7 +77,6 @@ import org.apache.iotdb.tsfile.fileSystem.fsFactory.FSFactory;
import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.apache.iotdb.tsfile.utils.Pair;
-import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.writer.RestorableTsFileIOWriter;
import org.apache.commons.io.FileUtils;
@@ -1573,9 +1572,8 @@ public class StorageGroupProcessor {
}
}
- // TODO need a read lock, please consider the concurrency with flush manager threads.
public QueryDataSource query(
- PartialPath fullPath,
+ List<PartialPath> pathList,
QueryContext context,
QueryFileManager filePathsManager,
Filter timeFilter)
@@ -1586,7 +1584,7 @@ public class StorageGroupProcessor {
getFileResourceListForQuery(
tsFileManagement.getTsFileList(true),
upgradeSeqFileList,
- fullPath,
+ pathList,
context,
timeFilter,
true);
@@ -1594,7 +1592,7 @@ public class StorageGroupProcessor {
getFileResourceListForQuery(
tsFileManagement.getTsFileList(false),
upgradeUnseqFileList,
- fullPath,
+ pathList,
context,
timeFilter,
false);
@@ -1614,25 +1612,6 @@ public class StorageGroupProcessor {
}
}
- public QueryDataSource getAllQueryDataSource(Filter timeFilter) throws QueryProcessException {
- readLock();
- try {
- Pair<List<TsFileResource>, TsFileResource> seqResources =
- getFileResourceListForQuery(tsFileManagement.getTsFileList(true), timeFilter, true);
- Pair<List<TsFileResource>, TsFileResource> unseqResources =
- getFileResourceListForQuery(tsFileManagement.getTsFileList(false), timeFilter, false);
- QueryDataSource dataSource = new QueryDataSource(seqResources.left, unseqResources.left);
- dataSource.setUnclosedSeqResource(seqResources.right);
- dataSource.setUnclosedUnseqResource(unseqResources.right);
- dataSource.setDataTTL(dataTTL);
- return dataSource;
- } catch (MetadataException e) {
- throw new QueryProcessException(e);
- } finally {
- readUnlock();
- }
- }
-
public void readLock() {
// apply read lock for SG insert lock to prevent inconsistent with concurrently writing memtable
insertLock.readLock().lock();
@@ -1655,14 +1634,6 @@ public class StorageGroupProcessor {
insertLock.writeLock().unlock();
}
- public void closeQueryLock() {
- closeQueryLock.readLock().lock();
- }
-
- public void closeQueryUnLock() {
- closeQueryLock.readLock().unlock();
- }
-
/**
* @param tsFileResources includes sealed and unsealed tsfile resources
* @return fill unsealed tsfile resources with memory data and ChunkMetadataList of data in disk
@@ -1670,35 +1641,30 @@ public class StorageGroupProcessor {
private List<TsFileResource> getFileResourceListForQuery(
Collection<TsFileResource> tsFileResources,
List<TsFileResource> upgradeTsFileResources,
- PartialPath fullPath,
+ List<PartialPath> pathList,
QueryContext context,
Filter timeFilter,
boolean isSeq)
throws MetadataException {
- String deviceId = fullPath.getDevice();
-
if (context.isDebug()) {
DEBUG_LOGGER.info(
- "Path: {}.{}, get tsfile list: {} isSeq: {} timefilter: {}",
- deviceId,
- fullPath.getMeasurement(),
+ "Path: {}, get tsfile list: {} isSeq: {} timefilter: {}",
+ pathList,
tsFileResources,
isSeq,
(timeFilter == null ? "null" : timeFilter));
}
- MeasurementSchema schema = IoTDB.metaManager.getSeriesSchema(fullPath);
-
List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>();
+
long timeLowerBound =
dataTTL != Long.MAX_VALUE ? System.currentTimeMillis() - dataTTL : Long.MIN_VALUE;
context.setQueryTimeLowerBound(timeLowerBound);
// for upgrade files and old files must be closed
for (TsFileResource tsFileResource : upgradeTsFileResources) {
- if (!tsFileResource.isSatisfied(
- fullPath.getDevice(), timeFilter, isSeq, dataTTL, context.isDebug())) {
+ if (!tsFileResource.isSatisfied(timeFilter, isSeq, dataTTL, context.isDebug())) {
continue;
}
closeQueryLock.readLock().lock();
@@ -1710,8 +1676,7 @@ public class StorageGroupProcessor {
}
for (TsFileResource tsFileResource : tsFileResources) {
- if (!tsFileResource.isSatisfied(
- fullPath.getDevice(), timeFilter, isSeq, dataTTL, context.isDebug())) {
+ if (!tsFileResource.isSatisfied(timeFilter, isSeq, dataTTL, context.isDebug())) {
continue;
}
closeQueryLock.readLock().lock();
@@ -1719,16 +1684,7 @@ public class StorageGroupProcessor {
if (tsFileResource.isClosed()) {
tsfileResourcesForQuery.add(tsFileResource);
} else {
- tsFileResource
- .getUnsealedFileProcessor()
- .query(
- deviceId,
- fullPath.getMeasurement(),
- schema.getType(),
- schema.getEncodingType(),
- schema.getProps(),
- context,
- tsfileResourcesForQuery);
+ tsFileResource.getProcessor().query(pathList, context, tsfileResourcesForQuery);
}
} catch (IOException e) {
throw new MetadataException(e);
@@ -1740,34 +1696,6 @@ public class StorageGroupProcessor {
}
/**
- * @param tsFileResources includes sealed and unsealed tsfile resources
- * @return fill unsealed tsfile resources with memory data and ChunkMetadataList of data in disk
- */
- private Pair<List<TsFileResource>, TsFileResource> getFileResourceListForQuery(
- Collection<TsFileResource> tsFileResources, Filter timeFilter, boolean isSeq)
- throws MetadataException {
- List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>();
- TsFileResource unclosedTsfileResourceForQuery = null;
- for (TsFileResource tsFileResource : tsFileResources) {
- if (!tsFileResource.isSatisfied(timeFilter, isSeq, dataTTL)) {
- continue;
- }
- closeQueryLock.readLock().lock();
- try {
- if (tsFileResource.isClosed()) {
- tsfileResourcesForQuery.add(tsFileResource);
- } else {
- // There is at most one unclosed tsFile
- unclosedTsfileResourceForQuery = tsFileResource;
- }
- } finally {
- closeQueryLock.readLock().unlock();
- }
- }
- return new Pair<>(tsfileResourcesForQuery, unclosedTsfileResourceForQuery);
- }
-
- /**
* Delete data whose timestamp <= 'timestamp' and belongs to the time series
* deviceId.measurementId.
*
@@ -1892,7 +1820,7 @@ public class StorageGroupProcessor {
// delete data in memory of unsealed file
if (!tsFileResource.isClosed()) {
- TsFileProcessor tsfileProcessor = tsFileResource.getUnsealedFileProcessor();
+ TsFileProcessor tsfileProcessor = tsFileResource.getProcessor();
tsfileProcessor.deleteDataInMemory(deletion, devicePaths);
}
@@ -2885,6 +2813,11 @@ public class StorageGroupProcessor {
return virtualStorageGroupId;
}
+ /** @return virtual storage group name, like root.sg1/0 */
+ public String getStorageGroupPath() {
+ return logicalStorageGroupName + File.separator + virtualStorageGroupId;
+ }
+
public StorageGroupInfo getStorageGroupInfo() {
return storageGroupInfo;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
index 51aa068..859ece6 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessor.java
@@ -59,7 +59,6 @@ import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.service.rpc.thrift.TSStatus;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.common.TimeRange;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -73,6 +72,7 @@ import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@@ -1089,172 +1089,95 @@ public class TsFileProcessor {
* memtables and then compact them into one TimeValuePairSorter). Then get the related
* ChunkMetadata of data on disk.
*
- * @param deviceId device id
- * @param measurementId measurements id
- * @param dataType data type
- * @param encoding encoding
+ * @param seriesPaths selected paths
*/
- @SuppressWarnings("squid:S3776") // Suppress high Cognitive Complexity warning
public void query(
- String deviceId,
- String measurementId,
- TSDataType dataType,
- TSEncoding encoding,
- Map<String, String> props,
+ List<PartialPath> seriesPaths,
QueryContext context,
List<TsFileResource> tsfileResourcesForQuery)
- throws IOException, MetadataException {
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: {} get flushQueryLock and hotCompactionMergeLock read lock",
- storageGroupName,
- tsFileResource.getTsFile().getName());
- }
- flushQueryLock.readLock().lock();
- try {
- List<ReadOnlyMemChunk> readOnlyMemChunks = new ArrayList<>();
- for (IMemTable flushingMemTable : flushingMemTables) {
- if (flushingMemTable.isSignalMemTable()) {
- continue;
- }
- List<TimeRange> deletionList =
- constructDeletionList(
- flushingMemTable, deviceId, measurementId, context.getQueryTimeLowerBound());
- ReadOnlyMemChunk memChunk =
- flushingMemTable.query(
- deviceId,
- measurementId,
- dataType,
- encoding,
- props,
- context.getQueryTimeLowerBound(),
- deletionList);
- if (memChunk != null) {
- readOnlyMemChunks.add(memChunk);
- }
- }
- if (workMemTable != null) {
- ReadOnlyMemChunk memChunk =
- workMemTable.query(
- deviceId,
- measurementId,
- dataType,
- encoding,
- props,
- context.getQueryTimeLowerBound(),
- null);
- if (memChunk != null) {
- readOnlyMemChunks.add(memChunk);
- }
- }
-
- ModificationFile modificationFile = tsFileResource.getModFile();
- List<Modification> modifications =
- context.getPathModifications(
- modificationFile,
- new PartialPath(deviceId + IoTDBConstant.PATH_SEPARATOR + measurementId));
-
- List<ChunkMetadata> chunkMetadataList =
- writer.getVisibleMetadataList(deviceId, measurementId, dataType);
- QueryUtils.modifyChunkMetaData(chunkMetadataList, modifications);
- chunkMetadataList.removeIf(context::chunkNotSatisfy);
-
- // get in memory data
- if (!readOnlyMemChunks.isEmpty() || !chunkMetadataList.isEmpty()) {
- tsfileResourcesForQuery.add(
- new TsFileResource(readOnlyMemChunks, chunkMetadataList, tsFileResource));
- }
- } catch (QueryProcessException e) {
- logger.error(
- "{}: {} get ReadOnlyMemChunk has error",
- storageGroupName,
- tsFileResource.getTsFile().getName(),
- e);
- } finally {
- flushQueryLock.readLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: {} release flushQueryLock",
- storageGroupName,
- tsFileResource.getTsFile().getName());
- }
- }
- }
-
- public TsFileResource query(PartialPath seriesPath, QueryContext context) throws IOException {
- String deviceId = seriesPath.getDevice();
- String measurementId = seriesPath.getMeasurement();
+ throws IOException {
+ Map<PartialPath, List<ChunkMetadata>> pathToChunkMetadataListMap = new HashMap<>();
+ Map<PartialPath, List<ReadOnlyMemChunk>> pathToReadOnlyMemChunkMap = new HashMap<>();
+ for (PartialPath seriesPath : seriesPaths) {
+ String deviceId = seriesPath.getDevice();
+ String measurementId = seriesPath.getMeasurement();
- flushQueryLock.readLock().lock();
- try {
- MeasurementSchema schema = IoTDB.metaManager.getSeriesSchema(seriesPath);
- List<ReadOnlyMemChunk> readOnlyMemChunks = new ArrayList<>();
- for (IMemTable flushingMemTable : flushingMemTables) {
- if (flushingMemTable.isSignalMemTable()) {
- continue;
- }
- List<TimeRange> deletionList =
- constructDeletionList(
- flushingMemTable, deviceId, measurementId, context.getQueryTimeLowerBound());
- ReadOnlyMemChunk memChunk =
- flushingMemTable.query(
- deviceId,
- measurementId,
- schema.getType(),
- schema.getEncodingType(),
- schema.getProps(),
- context.getQueryTimeLowerBound(),
- deletionList);
- if (memChunk != null) {
- readOnlyMemChunks.add(memChunk);
+ flushQueryLock.readLock().lock();
+ try {
+ MeasurementSchema schema = IoTDB.metaManager.getSeriesSchema(seriesPath);
+ List<ReadOnlyMemChunk> readOnlyMemChunks = new ArrayList<>();
+ for (IMemTable flushingMemTable : flushingMemTables) {
+ if (flushingMemTable.isSignalMemTable()) {
+ continue;
+ }
+ List<TimeRange> deletionList =
+ constructDeletionList(
+ flushingMemTable, deviceId, measurementId, context.getQueryTimeLowerBound());
+ ReadOnlyMemChunk memChunk =
+ flushingMemTable.query(
+ deviceId,
+ measurementId,
+ schema.getType(),
+ schema.getEncodingType(),
+ schema.getProps(),
+ context.getQueryTimeLowerBound(),
+ deletionList);
+ if (memChunk != null) {
+ readOnlyMemChunks.add(memChunk);
+ }
}
- }
- if (workMemTable != null) {
- ReadOnlyMemChunk memChunk =
- workMemTable.query(
- deviceId,
- measurementId,
- schema.getType(),
- schema.getEncodingType(),
- schema.getProps(),
- context.getQueryTimeLowerBound(),
- null);
- if (memChunk != null) {
- readOnlyMemChunks.add(memChunk);
+ if (workMemTable != null) {
+ ReadOnlyMemChunk memChunk =
+ workMemTable.query(
+ deviceId,
+ measurementId,
+ schema.getType(),
+ schema.getEncodingType(),
+ schema.getProps(),
+ context.getQueryTimeLowerBound(),
+ null);
+ if (memChunk != null) {
+ readOnlyMemChunks.add(memChunk);
+ }
}
- }
-
- ModificationFile modificationFile = tsFileResource.getModFile();
- List<Modification> modifications =
- context.getPathModifications(
- modificationFile,
- new PartialPath(deviceId + IoTDBConstant.PATH_SEPARATOR + measurementId));
- List<ChunkMetadata> chunkMetadataList =
- writer.getVisibleMetadataList(deviceId, measurementId, schema.getType());
- QueryUtils.modifyChunkMetaData(chunkMetadataList, modifications);
- chunkMetadataList.removeIf(context::chunkNotSatisfy);
-
- // get in memory data
- if (!readOnlyMemChunks.isEmpty() || !chunkMetadataList.isEmpty()) {
- return new TsFileResource(readOnlyMemChunks, chunkMetadataList, tsFileResource);
- }
- } catch (QueryProcessException | MetadataException e) {
- logger.error(
- "{}: {} get ReadOnlyMemChunk has error",
- storageGroupName,
- tsFileResource.getTsFile().getName(),
- e);
- } finally {
- flushQueryLock.readLock().unlock();
- if (logger.isDebugEnabled()) {
- logger.debug(
- "{}: {} release flushQueryLock",
+ ModificationFile modificationFile = tsFileResource.getModFile();
+ List<Modification> modifications =
+ context.getPathModifications(
+ modificationFile,
+ new PartialPath(deviceId + IoTDBConstant.PATH_SEPARATOR + measurementId));
+
+ List<ChunkMetadata> chunkMetadataList =
+ writer.getVisibleMetadataList(deviceId, measurementId, schema.getType());
+ QueryUtils.modifyChunkMetaData(chunkMetadataList, modifications);
+ chunkMetadataList.removeIf(context::chunkNotSatisfy);
+
+ // get in memory data
+ if (!readOnlyMemChunks.isEmpty() || !chunkMetadataList.isEmpty()) {
+ pathToReadOnlyMemChunkMap.put(seriesPath, readOnlyMemChunks);
+ pathToChunkMetadataListMap.put(seriesPath, chunkMetadataList);
+ }
+ } catch (QueryProcessException | MetadataException e) {
+ logger.error(
+ "{}: {} get ReadOnlyMemChunk has error",
storageGroupName,
- tsFileResource.getTsFile().getName());
+ tsFileResource.getTsFile().getName(),
+ e);
+ } finally {
+ flushQueryLock.readLock().unlock();
+ if (logger.isDebugEnabled()) {
+ logger.debug(
+ "{}: {} release flushQueryLock",
+ storageGroupName,
+ tsFileResource.getTsFile().getName());
+ }
}
}
- return null;
+ if (!pathToReadOnlyMemChunkMap.isEmpty() || !pathToChunkMetadataListMap.isEmpty()) {
+ tsfileResourcesForQuery.add(
+ new TsFileResource(
+ pathToReadOnlyMemChunkMap, pathToChunkMetadataListMap, tsFileResource));
+ }
}
public long getTimeRangeId() {
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
index ee9cdc3..5ba3272 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/storagegroup/TsFileResource.java
@@ -28,6 +28,7 @@ import org.apache.iotdb.db.engine.storagegroup.timeindex.ITimeIndex;
import org.apache.iotdb.db.engine.storagegroup.timeindex.TimeIndexLevel;
import org.apache.iotdb.db.engine.upgrade.UpgradeTask;
import org.apache.iotdb.db.exception.PartitionViolationException;
+import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.query.filter.TsFileFilter;
import org.apache.iotdb.db.service.UpgradeSevice;
import org.apache.iotdb.db.utils.TestOnly;
@@ -69,14 +70,13 @@ import static org.apache.iotdb.db.conf.IoTDBConstant.FILE_NAME_SUFFIX_UNSEQMERGE
import static org.apache.iotdb.db.conf.IoTDBConstant.FILE_NAME_SUFFIX_VERSION_INDEX;
import static org.apache.iotdb.tsfile.common.constant.TsFileConstant.TSFILE_SUFFIX;
-@SuppressWarnings("java:S1135") // ignore todos
public class TsFileResource {
- private static final Logger logger = LoggerFactory.getLogger(TsFileResource.class);
+ private static final Logger LOGGER = LoggerFactory.getLogger(TsFileResource.class);
private static final Logger DEBUG_LOGGER = LoggerFactory.getLogger("QUERY_DEBUG");
- private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
+ private static final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
// tsfile
private File file;
@@ -87,38 +87,20 @@ public class TsFileResource {
/** version number */
public static final byte VERSION_NUMBER = 1;
- public TsFileProcessor getProcessor() {
- return processor;
- }
-
- private TsFileProcessor processor;
-
- protected ITimeIndex timeIndex;
-
/** time index type, fileTimeIndex = 0, deviceTimeIndex = 1 */
private byte timeIndexType;
+ protected ITimeIndex timeIndex;
+
private ModificationFile modFile;
private volatile boolean closed = false;
private volatile boolean deleted = false;
private volatile boolean isMerging = false;
- private TsFileLock tsFileLock = new TsFileLock();
+ protected TsFileLock tsFileLock = new TsFileLock();
- private Random random = new Random();
-
- /**
- * Chunk metadata list of unsealed tsfile. Only be set in a temporal TsFileResource in a query
- * process.
- */
- private List<ChunkMetadata> chunkMetadataList;
-
- /** Mem chunk data. Only be set in a temporal TsFileResource in a query process. */
- private List<ReadOnlyMemChunk> readOnlyMemChunk;
-
- /** used for unsealed file to get TimeseriesMetadata */
- private TimeseriesMetadata timeSeriesMetadata;
+ private final Random random = new Random();
private FSFactory fsFactory = FSFactoryProducer.getFSFactory();
@@ -137,13 +119,6 @@ public class TsFileResource {
*/
private boolean isSeq;
- /**
- * If it is not null, it indicates that the current tsfile resource is a snapshot of the
- * originTsFileResource, and if so, when we want to used the lock, we should try to acquire the
- * lock of originTsFileResource
- */
- private TsFileResource originTsFileResource;
-
/** Maximum index of plans executed within this TsFile. */
protected long maxPlanIndex = Long.MIN_VALUE;
@@ -152,6 +127,27 @@ public class TsFileResource {
private long version = 0;
+ /**
+ * Chunk metadata list of unsealed tsfile. Only be set in a temporal TsFileResource in a query
+ * process.
+ */
+ private Map<PartialPath, List<ChunkMetadata>> pathToChunkMetadataListMap;
+
+ /** Mem chunk data. Only be set in a temporal TsFileResource in a query process. */
+ private Map<PartialPath, List<ReadOnlyMemChunk>> pathToReadOnlyMemChunkMap;
+
+ /** used for unsealed file to get TimeseriesMetadata */
+ private Map<PartialPath, TimeseriesMetadata> pathToTimeSeriesMetadataMap;
+
+ /**
+ * If it is not null, it indicates that the current tsfile resource is a snapshot of the
+ * originTsFileResource, and if so, when we want to used the lock, we should try to acquire the
+ * lock of originTsFileResource
+ */
+ private TsFileResource originTsFileResource;
+
+ private TsFileProcessor processor;
+
public TsFileResource() {}
public TsFileResource(TsFileResource other) throws IOException {
@@ -163,9 +159,9 @@ public class TsFileResource {
this.closed = other.closed;
this.deleted = other.deleted;
this.isMerging = other.isMerging;
- this.chunkMetadataList = other.chunkMetadataList;
- this.readOnlyMemChunk = other.readOnlyMemChunk;
- generateTimeSeriesMetadata();
+ this.pathToChunkMetadataListMap = other.pathToChunkMetadataListMap;
+ this.pathToReadOnlyMemChunkMap = other.pathToReadOnlyMemChunkMap;
+ generatePathToTimeSeriesMetadataMap();
this.tsFileLock = other.tsFileLock;
this.fsFactory = other.fsFactory;
this.maxPlanIndex = other.maxPlanIndex;
@@ -177,21 +173,22 @@ public class TsFileResource {
public TsFileResource(File file) {
this.file = file;
this.version = FilePathUtils.splitAndGetTsFileVersion(this.file.getName());
- this.timeIndex = config.getTimeIndexLevel().getTimeIndex();
- this.timeIndexType = (byte) config.getTimeIndexLevel().ordinal();
+ this.timeIndex = CONFIG.getTimeIndexLevel().getTimeIndex();
+ this.timeIndexType = (byte) CONFIG.getTimeIndexLevel().ordinal();
}
- /** unsealed TsFile */
+ /** unsealed TsFile, for writter */
public TsFileResource(File file, TsFileProcessor processor, int deviceNumInLastClosedTsFile) {
this.file = file;
this.version = FilePathUtils.splitAndGetTsFileVersion(this.file.getName());
- this.timeIndex = config.getTimeIndexLevel().getTimeIndex(deviceNumInLastClosedTsFile);
- this.timeIndexType = (byte) config.getTimeIndexLevel().ordinal();
+ this.timeIndex = CONFIG.getTimeIndexLevel().getTimeIndex(deviceNumInLastClosedTsFile);
+ this.timeIndexType = (byte) CONFIG.getTimeIndexLevel().ordinal();
this.processor = processor;
}
- /** unsealed TsFile */
+ /** unsealed TsFile, for query */
public TsFileResource(
+ PartialPath path,
List<ReadOnlyMemChunk> readOnlyMemChunk,
List<ChunkMetadata> chunkMetadataList,
TsFileResource originTsFileResource)
@@ -199,11 +196,29 @@ public class TsFileResource {
this.file = originTsFileResource.file;
this.timeIndex = originTsFileResource.timeIndex;
this.timeIndexType = originTsFileResource.timeIndexType;
- this.chunkMetadataList = chunkMetadataList;
- this.readOnlyMemChunk = readOnlyMemChunk;
+ this.pathToReadOnlyMemChunkMap = new HashMap<>();
+ pathToReadOnlyMemChunkMap.put(path, readOnlyMemChunk);
+ this.pathToChunkMetadataListMap = new HashMap<>();
+ pathToChunkMetadataListMap.put(path, chunkMetadataList);
+ this.originTsFileResource = originTsFileResource;
+ this.version = originTsFileResource.version;
+ generatePathToTimeSeriesMetadataMap();
+ }
+
+ /** unsealed TsFile, for query */
+ public TsFileResource(
+ Map<PartialPath, List<ReadOnlyMemChunk>> pathToReadOnlyMemChunkMap,
+ Map<PartialPath, List<ChunkMetadata>> pathToChunkMetadataListMap,
+ TsFileResource originTsFileResource)
+ throws IOException {
+ this.file = originTsFileResource.file;
+ this.timeIndex = originTsFileResource.timeIndex;
+ this.timeIndexType = originTsFileResource.timeIndexType;
+ this.pathToReadOnlyMemChunkMap = pathToReadOnlyMemChunkMap;
+ this.pathToChunkMetadataListMap = pathToChunkMetadataListMap;
this.originTsFileResource = originTsFileResource;
this.version = originTsFileResource.version;
- generateTimeSeriesMetadata();
+ generatePathToTimeSeriesMetadataMap();
}
@TestOnly
@@ -214,39 +229,6 @@ public class TsFileResource {
this.timeIndexType = 1;
}
- private void generateTimeSeriesMetadata() throws IOException {
- timeSeriesMetadata = new TimeseriesMetadata();
- timeSeriesMetadata.setOffsetOfChunkMetaDataList(-1);
- timeSeriesMetadata.setDataSizeOfChunkMetaDataList(-1);
-
- if (!(chunkMetadataList == null || chunkMetadataList.isEmpty())) {
- timeSeriesMetadata.setMeasurementId(chunkMetadataList.get(0).getMeasurementUid());
- TSDataType dataType = chunkMetadataList.get(0).getDataType();
- timeSeriesMetadata.setTSDataType(dataType);
- } else if (!(readOnlyMemChunk == null || readOnlyMemChunk.isEmpty())) {
- timeSeriesMetadata.setMeasurementId(readOnlyMemChunk.get(0).getMeasurementUid());
- TSDataType dataType = readOnlyMemChunk.get(0).getDataType();
- timeSeriesMetadata.setTSDataType(dataType);
- }
- if (timeSeriesMetadata.getTSDataType() != null) {
- Statistics<?> seriesStatistics =
- Statistics.getStatsByType(timeSeriesMetadata.getTSDataType());
- // flush chunkMetadataList one by one
- for (ChunkMetadata chunkMetadata : chunkMetadataList) {
- seriesStatistics.mergeStatistics(chunkMetadata.getStatistics());
- }
-
- for (ReadOnlyMemChunk memChunk : readOnlyMemChunk) {
- if (!memChunk.isEmpty()) {
- seriesStatistics.mergeStatistics(memChunk.getChunkMetaData().getStatistics());
- }
- }
- timeSeriesMetadata.setStatistics(seriesStatistics);
- } else {
- timeSeriesMetadata = null;
- }
- }
-
public synchronized void serialize() throws IOException {
try (OutputStream outputStream =
fsFactory.getBufferedOutputStream(file + RESOURCE_SUFFIX + TEMP_SUFFIX)) {
@@ -350,14 +332,6 @@ public class TsFileResource {
return fsFactory.getFile(file + RESOURCE_SUFFIX).exists();
}
- public List<ChunkMetadata> getChunkMetadataList() {
- return new ArrayList<>(chunkMetadataList);
- }
-
- public List<ReadOnlyMemChunk> getReadOnlyMemChunk() {
- return readOnlyMemChunk;
- }
-
public synchronized ModificationFile getModFile() {
if (modFile == null) {
modFile = new ModificationFile(file.getPath() + ModificationFile.FILE_SUFFIX);
@@ -423,6 +397,71 @@ public class TsFileResource {
return closed;
}
+ private void generatePathToTimeSeriesMetadataMap() throws IOException {
+ pathToTimeSeriesMetadataMap = new HashMap<>();
+ if (pathToChunkMetadataListMap == null || pathToReadOnlyMemChunkMap == null) return;
+ for (PartialPath path : pathToChunkMetadataListMap.keySet()) {
+ TimeseriesMetadata timeSeriesMetadata = new TimeseriesMetadata();
+ timeSeriesMetadata.setOffsetOfChunkMetaDataList(-1);
+ timeSeriesMetadata.setDataSizeOfChunkMetaDataList(-1);
+
+ if (pathToChunkMetadataListMap.containsKey(path)
+ && !pathToChunkMetadataListMap.get(path).isEmpty()) {
+ timeSeriesMetadata.setMeasurementId(
+ pathToChunkMetadataListMap.get(path).get(0).getMeasurementUid());
+ TSDataType dataType = pathToChunkMetadataListMap.get(path).get(0).getDataType();
+ timeSeriesMetadata.setTSDataType(dataType);
+ } else if (pathToReadOnlyMemChunkMap.containsKey(path)
+ && !pathToReadOnlyMemChunkMap.get(path).isEmpty()) {
+ timeSeriesMetadata.setMeasurementId(
+ pathToReadOnlyMemChunkMap.get(path).get(0).getMeasurementUid());
+ TSDataType dataType = pathToReadOnlyMemChunkMap.get(path).get(0).getDataType();
+ timeSeriesMetadata.setTSDataType(dataType);
+ }
+ if (timeSeriesMetadata.getTSDataType() != null) {
+ Statistics<?> seriesStatistics =
+ Statistics.getStatsByType(timeSeriesMetadata.getTSDataType());
+ // flush chunkMetadataList one by one
+ for (ChunkMetadata chunkMetadata : pathToChunkMetadataListMap.get(path)) {
+ seriesStatistics.mergeStatistics(chunkMetadata.getStatistics());
+ }
+
+ for (ReadOnlyMemChunk memChunk : pathToReadOnlyMemChunkMap.get(path)) {
+ if (!memChunk.isEmpty()) {
+ seriesStatistics.mergeStatistics(memChunk.getChunkMetaData().getStatistics());
+ }
+ }
+ timeSeriesMetadata.setStatistics(seriesStatistics);
+ } else {
+ timeSeriesMetadata = null;
+ }
+ pathToTimeSeriesMetadataMap.put(path, timeSeriesMetadata);
+ }
+ }
+
+ public List<ChunkMetadata> getChunkMetadataList(PartialPath seriesPath) {
+ return new ArrayList<>(pathToChunkMetadataListMap.get(seriesPath));
+ }
+
+ public List<ReadOnlyMemChunk> getReadOnlyMemChunk(PartialPath seriesPath) {
+ return pathToReadOnlyMemChunkMap.get(seriesPath);
+ }
+
+ public TimeseriesMetadata getTimeSeriesMetadata(PartialPath seriesPath) {
+ if (pathToTimeSeriesMetadataMap.containsKey(seriesPath)) {
+ return pathToTimeSeriesMetadataMap.get(seriesPath);
+ }
+ return null;
+ }
+
+ public TsFileProcessor getProcessor() {
+ return processor;
+ }
+
+ public void setProcessor(TsFileProcessor processor) {
+ this.processor = processor;
+ }
+
public void close() throws IOException {
closed = true;
if (modFile != null) {
@@ -430,14 +469,16 @@ public class TsFileResource {
modFile = null;
}
processor = null;
- chunkMetadataList = null;
+ pathToChunkMetadataListMap = null;
+ pathToReadOnlyMemChunkMap = null;
+ pathToTimeSeriesMetadataMap = null;
timeIndex.close();
}
- public TsFileProcessor getUnsealedFileProcessor() {
- return processor;
- }
-
+ /**
+ * If originTsFileResource is not null, we should acquire the read lock of originTsFileResource
+ * before construct the current TsFileResource
+ */
public void writeLock() {
if (originTsFileResource == null) {
tsFileLock.writeLock();
@@ -454,10 +495,6 @@ public class TsFileResource {
}
}
- /**
- * If originTsFileResource is not null, we should acquire the read lock of originTsFileResource
- * before construct the current TsFileResource
- */
public void readLock() {
if (originTsFileResource == null) {
tsFileLock.readLock();
@@ -492,13 +529,13 @@ public class TsFileResource {
try {
fsFactory.deleteIfExists(file);
} catch (IOException e) {
- logger.error("TsFile {} cannot be deleted: {}", file, e.getMessage());
+ LOGGER.error("TsFile {} cannot be deleted: {}", file, e.getMessage());
}
removeResourceFile();
try {
fsFactory.deleteIfExists(fsFactory.getFile(file.getPath() + ModificationFile.FILE_SUFFIX));
} catch (IOException e) {
- logger.error("ModificationFile {} cannot be deleted: {}", file, e.getMessage());
+ LOGGER.error("ModificationFile {} cannot be deleted: {}", file, e.getMessage());
}
}
@@ -507,7 +544,7 @@ public class TsFileResource {
fsFactory.deleteIfExists(fsFactory.getFile(file.getPath() + RESOURCE_SUFFIX + TEMP_SUFFIX));
fsFactory.deleteIfExists(fsFactory.getFile(file.getPath() + RESOURCE_SUFFIX));
} catch (IOException e) {
- logger.error("TsFileResource {} cannot be deleted: {}", file, e.getMessage());
+ LOGGER.error("TsFileResource {} cannot be deleted: {}", file, e.getMessage());
}
}
@@ -571,23 +608,14 @@ public class TsFileResource {
return timeIndex.stillLives(timeLowerBound);
}
- /** @return true if the device is contained in the TsFile and it lives beyond TTL */
- public boolean isSatisfied(
- String deviceId, Filter timeFilter, boolean isSeq, long ttl, boolean debug) {
- if (!getDevices().contains(deviceId)) {
- if (debug) {
- DEBUG_LOGGER.info(
- "Path: {} file {} is not satisfied because of no device!", deviceId, file);
- }
- return false;
- }
-
- long startTime = getStartTime(deviceId);
- long endTime = closed || !isSeq ? getEndTime(deviceId) : Long.MAX_VALUE;
+ /** @return true if the TsFile lives beyond TTL */
+ public boolean isSatisfied(Filter timeFilter, boolean isSeq, long ttl, boolean debug) {
+ long startTime = getFileStartTime();
+ long endTime = closed || !isSeq ? getFileEndTime() : Long.MAX_VALUE;
if (!isAlive(endTime, ttl)) {
if (debug) {
- DEBUG_LOGGER.info("Path: {} file {} is not satisfied because of ttl!", deviceId, file);
+ DEBUG_LOGGER.info("file {} is not satisfied because of ttl!", file);
}
return false;
}
@@ -595,29 +623,13 @@ public class TsFileResource {
if (timeFilter != null) {
boolean res = timeFilter.satisfyStartEndTime(startTime, endTime);
if (debug && !res) {
- DEBUG_LOGGER.info(
- "Path: {} file {} is not satisfied because of time filter!", deviceId, fsFactory);
+ DEBUG_LOGGER.info("Path: file {} is not satisfied because of time filter!", fsFactory);
}
return res;
}
return true;
}
- /** @return true if the TsFile lives beyond TTL */
- public boolean isSatisfied(Filter timeFilter, boolean isSeq, long ttl) {
- long startTime = getFileStartTime();
- long endTime = closed || !isSeq ? getFileEndTime() : Long.MAX_VALUE;
-
- if (!isAlive(endTime, ttl)) {
- return false;
- }
-
- if (timeFilter != null) {
- return timeFilter.satisfyStartEndTime(startTime, endTime);
- }
- return true;
- }
-
/** @return true if the device is contained in the TsFile and it lives beyond TTL */
public boolean isSatisfied(
String deviceId,
@@ -627,6 +639,10 @@ public class TsFileResource {
long ttl,
boolean debug) {
if (fileFilter != null && fileFilter.fileNotSatisfy(this)) {
+ if (debug) {
+ DEBUG_LOGGER.info(
+ "Path: {} file {} is not satisfied because of fileFilter!", deviceId, file);
+ }
return false;
}
@@ -641,13 +657,6 @@ public class TsFileResource {
long startTime = getStartTime(deviceId);
long endTime = closed || !isSeq ? getEndTime(deviceId) : Long.MAX_VALUE;
- if (!isAlive(endTime, ttl)) {
- if (debug) {
- DEBUG_LOGGER.info("Path: {} file {} is not satisfied because of ttl!", deviceId, file);
- }
- return false;
- }
-
if (timeFilter != null) {
boolean res = timeFilter.satisfyStartEndTime(startTime, endTime);
if (debug && !res) {
@@ -664,14 +673,6 @@ public class TsFileResource {
return dataTTL == Long.MAX_VALUE || (System.currentTimeMillis() - time) <= dataTTL;
}
- public void setProcessor(TsFileProcessor processor) {
- this.processor = processor;
- }
-
- public TimeseriesMetadata getTimeSeriesMetadata() {
- return timeSeriesMetadata;
- }
-
public void setUpgradedResources(List<TsFileResource> upgradedResources) {
this.upgradedResources = upgradedResources;
}
@@ -732,7 +733,7 @@ public class TsFileResource {
try {
newResource = new TsFileResource(this);
} catch (IOException e) {
- logger.error("Cannot create hardlink for {}", file, e);
+ LOGGER.error("Cannot create hardlink for {}", file, e);
return null;
}
@@ -751,7 +752,7 @@ public class TsFileResource {
} catch (FileAlreadyExistsException e) {
// retry a different name if the file is already created
} catch (IOException e) {
- logger.error("Cannot create hardlink for {}", file, e);
+ LOGGER.error("Cannot create hardlink for {}", file, e);
return null;
}
}
@@ -795,7 +796,7 @@ public class TsFileResource {
try {
serialize();
} catch (IOException e) {
- logger.error(
+ LOGGER.error(
"Cannot serialize TsFileResource {} when updating plan index {}-{}",
this,
maxPlanIndex,
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
index b03dbd6..051ac26 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.*;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.logfile.MLogReader;
import org.apache.iotdb.db.metadata.logfile.MLogWriter;
import org.apache.iotdb.db.metadata.logfile.TagLogFile;
@@ -822,7 +823,10 @@ public class MManager {
try {
list =
StorageEngine.getInstance()
- .mergeLock(allMatchedNodes.stream().map(MNode::getPartialPath).collect(toList()));
+ .mergeLockAndInitQueryDataSource(
+ allMatchedNodes.stream().map(MNode::getPartialPath).collect(toList()),
+ context,
+ null);
try {
allMatchedNodes =
allMatchedNodes.stream()
@@ -835,7 +839,7 @@ public class MManager {
} finally {
StorageEngine.getInstance().mergeUnLock(list);
}
- } catch (StorageEngineException e) {
+ } catch (StorageEngineException | QueryProcessException e) {
throw new MetadataException(e);
}
} else {
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
index 89e03c0..1058277 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MTree.java
@@ -123,7 +123,7 @@ public class MTree implements Serializable {
try {
QueryDataSource dataSource =
QueryResourceManager.getInstance()
- .getQueryDataSourceByPath(node.getPartialPath(), queryContext, null);
+ .getQueryDataSource(node.getPartialPath(), queryContext, null);
Set<String> measurementSet = new HashSet<>();
measurementSet.add(node.getPartialPath().getFullPath());
LastPointReader lastReader =
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/QueryFileManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/QueryFileManager.java
index 9e52a37..87d2c79 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/QueryFileManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/QueryFileManager.java
@@ -60,16 +60,6 @@ public class QueryFileManager {
// closed unsequence TsFileResource
addUsedFilesForQuery(queryId, dataSource.getUnseqResources());
-
- // unclosed sequence TsFileResource
- if (addUsedFileForQuery(queryId, dataSource.getUnclosedSeqResource())) {
- dataSource.setUnclosedSeqResource(null);
- }
-
- // unclosed unsequence TsFileResource
- if (addUsedFileForQuery(queryId, dataSource.getUnclosedUnseqResource())) {
- dataSource.setUnclosedUnseqResource(null);
- }
}
private void addUsedFilesForQuery(long queryId, List<TsFileResource> resources) {
@@ -81,8 +71,10 @@ public class QueryFileManager {
// this file may be deleted just before we lock it
if (tsFileResource.isDeleted()) {
+ Map<Long, Map<TsFileResource, TsFileResource>> pathMap =
+ !isClosed ? unsealedFilePathsMap : sealedFilePathsMap;
// This resource may be removed by other threads of this query.
- if (sealedFilePathsMap.get(queryId).remove(tsFileResource) != null) {
+ if (pathMap.get(queryId).remove(tsFileResource) != null) {
FileReaderManager.getInstance().decreaseFileReaderReference(tsFileResource, isClosed);
}
iterator.remove();
@@ -90,25 +82,6 @@ public class QueryFileManager {
}
}
- private boolean addUsedFileForQuery(long queryId, TsFileResource tsFileResource) {
- if (tsFileResource == null) {
- return false;
- }
-
- boolean isClosed = tsFileResource.isClosed();
- addFilePathToMap(queryId, tsFileResource, isClosed);
-
- // this file may be deleted just before we lock it
- if (tsFileResource.isDeleted()) {
- // This resource may be removed by other threads of this query.
- if (unsealedFilePathsMap.get(queryId).remove(tsFileResource) != null) {
- FileReaderManager.getInstance().decreaseFileReaderReference(tsFileResource, isClosed);
- }
- return true;
- }
- return false;
- }
-
/**
* Whenever the jdbc request is closed normally or abnormally, this method must be invoked. All
* file paths used by this jdbc request must be cleared and thus the usage reference must be
diff --git a/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java b/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
index 3ee86f4..54598cf 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/control/QueryResourceManager.java
@@ -23,8 +23,7 @@ import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
-import org.apache.iotdb.db.engine.storagegroup.TsFileProcessor;
-import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
+import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
@@ -32,7 +31,6 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.externalsort.serialize.IExternalSortFileDeserializer;
import org.apache.iotdb.db.query.udf.service.TemporaryQueryDataFileService;
import org.apache.iotdb.db.utils.QueryUtils;
-import org.apache.iotdb.tsfile.read.expression.impl.SingleSeriesExpression;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
import org.slf4j.Logger;
@@ -47,15 +45,19 @@ import java.util.concurrent.atomic.AtomicLong;
/**
* QueryResourceManager manages resource (file streams) used by each query job, and assign Ids to
* the jobs. During the life cycle of a query, the following methods must be called in strict order:
- * 1. assignQueryId - get an Id for the new query. 2. getQueryDataSource - open files for the job or
- * reuse existing readers. 3. endQueryForGivenJob - release the resource used by this job.
+ *
+ * <p>1. assignQueryId - get an Id for the new query.
+ *
+ * <p>2. getQueryDataSource - open files for the job or reuse existing readers.
+ *
+ * <p>3. endQueryForGivenJob - release the resource used by this job.
*/
public class QueryResourceManager {
private final AtomicLong queryIdAtom = new AtomicLong();
private final QueryFileManager filePathsManager;
private static final Logger logger = LoggerFactory.getLogger(QueryResourceManager.class);
- private IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
+ private final IoTDBConfig CONFIG = IoTDBDescriptor.getInstance().getConfig();
/**
* Record temporary files used for external sorting.
@@ -64,12 +66,17 @@ public class QueryResourceManager {
*/
private final Map<Long, List<IExternalSortFileDeserializer>> externalSortFileMap;
+ /**
+ * Record QueryDataSource used in queries
+ *
+ * <p>Key: query job id. Value: QueryDataSource corresponding to each virtual storage group.
+ */
private final Map<Long, Map<String, QueryDataSource>> cachedQueryDataSourcesMap;
private QueryResourceManager() {
filePathsManager = new QueryFileManager();
externalSortFileMap = new ConcurrentHashMap<>();
- cachedQueryDataSourcesMap = new HashMap<>();
+ cachedQueryDataSourcesMap = new ConcurrentHashMap<>();
}
public static QueryResourceManager getInstance() {
@@ -96,25 +103,36 @@ public class QueryResourceManager {
externalSortFileMap.computeIfAbsent(queryId, x -> new ArrayList<>()).add(deserializer);
}
- public QueryDataSource getQueryDataSourceByPath(
- PartialPath selectedPath, QueryContext context, Filter filter)
- throws StorageEngineException, QueryProcessException {
-
- SingleSeriesExpression singleSeriesExpression =
- new SingleSeriesExpression(selectedPath, filter);
- QueryDataSource queryDataSource =
- StorageEngine.getInstance().query(singleSeriesExpression, context, filePathsManager);
- // calculate the distinct number of seq and unseq tsfiles
- if (CONFIG.isEnablePerformanceTracing()) {
- TracingManager.getInstance()
- .getTracingInfo(context.getQueryId())
- .addTsFileSet(queryDataSource.getSeqResources(), queryDataSource.getUnseqResources());
+ /**
+ * The method is called in mergeLock() when executing query. This method will get all the
+ * QueryDataSource needed for this query and put them in the cachedQueryDataSourcesMap.
+ *
+ * @param processorToSeriesMap Key: processor of the virtual storage group. Value: selected series
+ * under the virtual storage group
+ */
+ public void initQueryDataSource(
+ Map<StorageGroupProcessor, List<PartialPath>> processorToSeriesMap,
+ QueryContext context,
+ Filter timeFilter)
+ throws QueryProcessException {
+ for (Map.Entry<StorageGroupProcessor, List<PartialPath>> entry :
+ processorToSeriesMap.entrySet()) {
+ StorageGroupProcessor processor = entry.getKey();
+ List<PartialPath> pathList = entry.getValue();
+
+ long queryId = context.getQueryId();
+ String storageGroupPath = processor.getStorageGroupPath();
+
+ QueryDataSource cachedQueryDataSource =
+ processor.query(pathList, context, filePathsManager, timeFilter);
+ cachedQueryDataSourcesMap
+ .computeIfAbsent(queryId, k -> new HashMap<>())
+ .put(storageGroupPath, cachedQueryDataSource);
}
- return queryDataSource;
}
public QueryDataSource getQueryDataSource(
- PartialPath selectedPath, QueryContext context, Filter filter)
+ PartialPath selectedPath, QueryContext context, Filter timeFilter)
throws StorageEngineException, QueryProcessException {
long queryId = context.getQueryId();
@@ -127,25 +145,17 @@ public class QueryResourceManager {
&& cachedQueryDataSourcesMap.get(queryId).containsKey(storageGroupPath)) {
cachedQueryDataSource = cachedQueryDataSourcesMap.get(queryId).get(storageGroupPath);
} else {
- SingleSeriesExpression singleSeriesExpression =
- new SingleSeriesExpression(selectedPath, filter);
+ // QueryDataSource is not cached earlier in cluster mode
+ StorageGroupProcessor processor =
+ StorageEngine.getInstance().getProcessor(selectedPath.getDevicePath());
cachedQueryDataSource =
- StorageEngine.getInstance().getAllQueryDataSource(singleSeriesExpression);
+ processor.query(
+ Collections.singletonList(selectedPath), context, filePathsManager, timeFilter);
cachedQueryDataSourcesMap
.computeIfAbsent(queryId, k -> new HashMap<>())
.put(storageGroupPath, cachedQueryDataSource);
-
- // used files should be added before mergeLock is unlocked, or they may be deleted by running
- // merge
- filePathsManager.addUsedFilesForQuery(context.getQueryId(), cachedQueryDataSource);
}
- // set query time lower bound according TTL
- long dataTTL = cachedQueryDataSource.getDataTTL();
- long timeLowerBound =
- dataTTL != Long.MAX_VALUE ? System.currentTimeMillis() - dataTTL : Long.MIN_VALUE;
- context.setQueryTimeLowerBound(timeLowerBound);
-
// construct QueryDataSource for selectedPath
QueryDataSource queryDataSource =
new QueryDataSource(
@@ -153,63 +163,12 @@ public class QueryResourceManager {
queryDataSource.setDataTTL(cachedQueryDataSource.getDataTTL());
- TsFileResource cachedUnclosedSeqResource = cachedQueryDataSource.getUnclosedSeqResource();
- if (cachedUnclosedSeqResource != null) {
- try {
- StorageEngine.getInstance().getProcessor(selectedPath.getDevicePath()).closeQueryLock();
- TsFileProcessor processor = cachedUnclosedSeqResource.getUnsealedFileProcessor();
- if (processor != null) {
- queryDataSource.setUnclosedSeqResource(processor.query(selectedPath, context));
- } else {
- // tsFileResource is closed
- queryDataSource.setUnclosedSeqResource(cachedUnclosedSeqResource);
- }
- } catch (IOException e) {
- throw new QueryProcessException(
- String.format(
- "%s: %s get ReadOnlyMemChunk has error",
- storageGroupPath, cachedUnclosedSeqResource.getTsFile().getName()));
- } finally {
- StorageEngine.getInstance().getProcessor(selectedPath.getDevicePath()).closeQueryUnLock();
- }
- }
-
- TsFileResource cachedUnclosedUnseqResource = cachedQueryDataSource.getUnclosedUnseqResource();
- if (cachedUnclosedUnseqResource != null) {
- try {
- StorageEngine.getInstance().getProcessor(selectedPath.getDevicePath()).closeQueryLock();
- TsFileProcessor processor = cachedUnclosedUnseqResource.getUnsealedFileProcessor();
- if (processor != null) {
- queryDataSource.setUnclosedUnseqResource(processor.query(selectedPath, context));
- } else {
- // tsFileResource is closed
- queryDataSource.setUnclosedUnseqResource(cachedUnclosedUnseqResource);
- }
- } catch (IOException e) {
- throw new QueryProcessException(
- String.format(
- "%s: %s get ReadOnlyMemChunk has error",
- storageGroupPath, cachedUnclosedUnseqResource.getTsFile().getName()));
- } finally {
- StorageEngine.getInstance().getProcessor(selectedPath.getDevicePath()).closeQueryUnLock();
- }
- }
-
// calculate the read order of unseqResources
QueryUtils.fillOrderIndexes(queryDataSource, deviceId, context.isAscending());
return queryDataSource;
}
- public void clearCachedQueryDataSource(PartialPath path, QueryContext context)
- throws StorageEngineException {
- long queryId = context.getQueryId();
- String storageGroupPath = StorageEngine.getInstance().getStorageGroupPath(path);
- if (cachedQueryDataSourcesMap.containsKey(queryId)) {
- cachedQueryDataSourcesMap.get(queryId).remove(storageGroupPath);
- }
- }
-
/**
* Whenever the jdbc request is closed normally or abnormally, this method must be invoked. All
* query tokens created by this jdbc request must be cleared.
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByFillDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByFillDataSet.java
index 1805216..6640adb 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByFillDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByFillDataSet.java
@@ -33,6 +33,9 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.TimeFilter;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -41,6 +44,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
public class GroupByFillDataSet extends QueryDataSet {
@@ -53,6 +57,7 @@ public class GroupByFillDataSet extends QueryDataSet {
// last timestamp for each time series
private long[] lastTimeArray;
private TimeValuePair[] firstNotNullTV;
+ private PreviousFill[] previousFillExecutors;
private boolean isPeekEnded = false;
public GroupByFillDataSet(
@@ -68,13 +73,9 @@ public class GroupByFillDataSet extends QueryDataSet {
this.groupByEngineDataSet = groupByEngineDataSet;
this.lastQueryExecutor = lastQueryExecutor;
this.fillTypes = fillTypes;
- List<StorageGroupProcessor> list = StorageEngine.getInstance().mergeLock(paths);
- try {
- initPreviousParis(context, groupByFillPlan);
- initLastTimeArray(context, groupByFillPlan);
- } finally {
- StorageEngine.getInstance().mergeUnLock(list);
- }
+
+ initPreviousParis(context, groupByFillPlan);
+ initLastTimeArray(context, groupByFillPlan);
}
private void initPreviousParis(QueryContext context, GroupByTimeFillPlan groupByFillPlan)
@@ -82,24 +83,25 @@ public class GroupByFillDataSet extends QueryDataSet {
previousValue = new Object[paths.size()];
previousTime = new long[paths.size()];
firstNotNullTV = new TimeValuePair[paths.size()];
+ previousFillExecutors = new PreviousFill[paths.size()];
+ long lowerBound = Long.MAX_VALUE;
for (int i = 0; i < paths.size(); i++) {
PartialPath path = (PartialPath) paths.get(i);
TSDataType dataType = dataTypes.get(i);
- IFill fill;
+ PreviousFill fill;
+ long beforeRange;
if (fillTypes.containsKey(dataType)) {
+ beforeRange = ((PreviousFill) fillTypes.get(dataType)).getBeforeRange();
fill =
new PreviousFill(
dataType,
groupByEngineDataSet.getStartTime(),
- ((PreviousFill) fillTypes.get(dataType)).getBeforeRange(),
+ beforeRange,
((PreviousFill) fillTypes.get(dataType)).isUntilLast());
} else {
- fill =
- new PreviousFill(
- dataType,
- groupByEngineDataSet.getStartTime(),
- IoTDBDescriptor.getInstance().getConfig().getDefaultFillInterval());
+ beforeRange = IoTDBDescriptor.getInstance().getConfig().getDefaultFillInterval();
+ fill = new PreviousFill(dataType, groupByEngineDataSet.getStartTime(), beforeRange);
}
fill.configureFill(
path,
@@ -107,15 +109,40 @@ public class GroupByFillDataSet extends QueryDataSet {
groupByEngineDataSet.getStartTime(),
groupByFillPlan.getAllMeasurementsInDevice(path.getDevice()),
context);
+ previousFillExecutors[i] = fill;
+
+ lowerBound =
+ Math.min(
+ lowerBound,
+ beforeRange == -1
+ ? Long.MIN_VALUE
+ : groupByEngineDataSet.getStartTime() - beforeRange);
+ }
- firstNotNullTV[i] = fill.getFillResult();
- TimeValuePair timeValuePair = firstNotNullTV[i];
- previousValue[i] = null;
- previousTime[i] = Long.MAX_VALUE;
- if (ascending && timeValuePair != null && timeValuePair.getValue() != null) {
- previousValue[i] = timeValuePair.getValue().getValue();
- previousTime[i] = timeValuePair.getTimestamp();
+ Filter timeFilter =
+ FilterFactory.and(
+ TimeFilter.gtEq(lowerBound), TimeFilter.ltEq(groupByEngineDataSet.getStartTime()));
+
+ List<StorageGroupProcessor> list =
+ StorageEngine.getInstance()
+ .mergeLockAndInitQueryDataSource(
+ paths.stream().map(path -> (PartialPath) path).collect(Collectors.toList()),
+ context,
+ timeFilter);
+ try {
+ for (int i = 0; i < paths.size(); i++) {
+ PreviousFill fill = previousFillExecutors[i];
+ firstNotNullTV[i] = fill.getFillResult();
+ TimeValuePair timeValuePair = firstNotNullTV[i];
+ previousValue[i] = null;
+ previousTime[i] = Long.MAX_VALUE;
+ if (ascending && timeValuePair != null && timeValuePair.getValue() != null) {
+ previousValue[i] = timeValuePair.getValue().getValue();
+ previousTime[i] = timeValuePair.getTimestamp();
+ }
}
+ } finally {
+ StorageEngine.getInstance().mergeUnLock(list);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
index e3e8a2b..4711cbb 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithValueFilterDataSet.java
@@ -39,6 +39,9 @@ import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Path;
import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.apache.iotdb.tsfile.read.filter.TimeFilter;
+import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
import org.apache.iotdb.tsfile.read.query.timegenerator.TimeGenerator;
import org.apache.iotdb.tsfile.utils.Pair;
@@ -84,9 +87,17 @@ public class GroupByWithValueFilterDataSet extends GroupByEngineDataSet {
this.allDataReaderList = new ArrayList<>();
this.groupByTimePlan = groupByTimePlan;
+ Filter timeFilter =
+ FilterFactory.and(
+ TimeFilter.gtEq(groupByTimePlan.getStartTime()),
+ TimeFilter.lt(groupByTimePlan.getEndTime()));
+
List<StorageGroupProcessor> list =
StorageEngine.getInstance()
- .mergeLock(paths.stream().map(p -> (PartialPath) p).collect(Collectors.toList()));
+ .mergeLockAndInitQueryDataSource(
+ paths.stream().map(p -> (PartialPath) p).collect(Collectors.toList()),
+ context,
+ timeFilter);
try {
for (int i = 0; i < paths.size(); i++) {
PartialPath path = (PartialPath) paths.get(i);
diff --git a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
index 0c2025f..424915f 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/dataset/groupby/GroupByWithoutValueFilterDataSet.java
@@ -91,7 +91,10 @@ public class GroupByWithoutValueFilterDataSet extends GroupByEngineDataSet {
List<StorageGroupProcessor> list =
StorageEngine.getInstance()
- .mergeLock(paths.stream().map(p -> (PartialPath) p).collect(Collectors.toList()));
+ .mergeLockAndInitQueryDataSource(
+ paths.stream().map(p -> (PartialPath) p).collect(Collectors.toList()),
+ context,
+ timeFilter);
try {
// init resultIndexes, group result indexes by path
for (int i = 0; i < paths.size(); i++) {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
index bb4e53a..d7339a8 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/AggregationExecutor.java
@@ -104,7 +104,9 @@ public class AggregationExecutor {
AggregateResult[] aggregateResultList = new AggregateResult[selectedSeries.size()];
// TODO-Cluster: group the paths by storage group to reduce communications
List<StorageGroupProcessor> list =
- StorageEngine.getInstance().mergeLock(new ArrayList<>(pathToAggrIndexesMap.keySet()));
+ StorageEngine.getInstance()
+ .mergeLockAndInitQueryDataSource(
+ new ArrayList<>(pathToAggrIndexesMap.keySet()), context, timeFilter);
try {
for (Map.Entry<PartialPath, List<Integer>> entry : pathToAggrIndexesMap.entrySet()) {
aggregateOneSeries(
@@ -344,7 +346,8 @@ public class AggregationExecutor {
Map<PartialPath, List<Integer>> pathToAggrIndexesMap =
groupAggregationsBySeries(selectedSeries);
Map<IReaderByTimestamp, List<Integer>> readerToAggrIndexesMap = new HashMap<>();
- List<StorageGroupProcessor> list = StorageEngine.getInstance().mergeLock(selectedSeries);
+ List<StorageGroupProcessor> list =
+ StorageEngine.getInstance().mergeLockAndInitQueryDataSource(selectedSeries, context, null);
try {
for (int i = 0; i < selectedSeries.size(); i++) {
PartialPath path = selectedSeries.get(i);
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
index 2ad3df0..e43c094 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/FillQueryExecutor.java
@@ -31,7 +31,9 @@ import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.QueryResourceManager;
import org.apache.iotdb.db.query.dataset.SingleDataSet;
import org.apache.iotdb.db.query.executor.fill.IFill;
+import org.apache.iotdb.db.query.executor.fill.LinearFill;
import org.apache.iotdb.db.query.executor.fill.PreviousFill;
+import org.apache.iotdb.db.query.executor.fill.ValueFill;
import org.apache.iotdb.db.query.reader.series.ManagedSeriesReader;
import org.apache.iotdb.db.query.reader.series.SeriesRawDataBatchReader;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -40,6 +42,7 @@ import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.read.filter.TimeFilter;
import org.apache.iotdb.tsfile.read.filter.basic.Filter;
+import org.apache.iotdb.tsfile.read.filter.factory.FilterFactory;
import org.apache.iotdb.tsfile.read.query.dataset.QueryDataSet;
import javax.activation.UnsupportedDataTypeException;
@@ -75,7 +78,9 @@ public class FillQueryExecutor {
throws StorageEngineException, QueryProcessException, IOException {
RowRecord record = new RowRecord(queryTime);
- List<StorageGroupProcessor> list = StorageEngine.getInstance().mergeLock(selectedSeries);
+ List<StorageGroupProcessor> list =
+ StorageEngine.getInstance()
+ .mergeLockAndInitQueryDataSource(selectedSeries, context, contructTimeFilter());
try {
List<TimeValuePair> timeValuePairs = getTimeValuePairs(context);
long defaultFillInterval = IoTDBDescriptor.getInstance().getConfig().getDefaultFillInterval();
@@ -131,6 +136,52 @@ public class FillQueryExecutor {
return dataSet;
}
+ private Filter contructTimeFilter() throws UnsupportedDataTypeException {
+ long lowerBound = Long.MAX_VALUE;
+ long upperBound = Long.MIN_VALUE;
+ long defaultFillInterval = IoTDBDescriptor.getInstance().getConfig().getDefaultFillInterval();
+ for (int i = 0; i < selectedSeries.size(); i++) {
+ TSDataType dataType = dataTypes.get(i);
+ if (!typeIFillMap.containsKey(dataType)) {
+ switch (dataType) {
+ case INT32:
+ case INT64:
+ case FLOAT:
+ case DOUBLE:
+ case BOOLEAN:
+ case TEXT:
+ lowerBound =
+ Math.min(
+ lowerBound,
+ defaultFillInterval == -1 ? Long.MIN_VALUE : queryTime - defaultFillInterval);
+ upperBound = Math.max(upperBound, queryTime);
+ break;
+ default:
+ throw new UnsupportedDataTypeException("unsupported data type " + dataType);
+ }
+ } else {
+ IFill fill = typeIFillMap.get(dataType).copy();
+ if (fill instanceof PreviousFill) {
+ long beforeRange = ((PreviousFill) fill).getBeforeRange();
+ lowerBound =
+ Math.min(lowerBound, beforeRange == -1 ? Long.MIN_VALUE : queryTime - beforeRange);
+ upperBound = Math.max(upperBound, queryTime);
+ } else if (fill instanceof LinearFill) {
+ long beforeRange = ((LinearFill) fill).getBeforeRange();
+ long afterRange = ((LinearFill) fill).getAfterRange();
+ lowerBound =
+ Math.min(lowerBound, beforeRange == -1 ? Long.MIN_VALUE : queryTime - beforeRange);
+ upperBound =
+ Math.max(upperBound, afterRange == -1 ? Long.MAX_VALUE : queryTime + afterRange);
+ } else if (fill instanceof ValueFill) {
+ lowerBound = Math.min(lowerBound, queryTime);
+ upperBound = Math.max(upperBound, queryTime);
+ }
+ }
+ }
+ return FilterFactory.and(TimeFilter.gtEq(lowerBound), TimeFilter.ltEq(upperBound));
+ }
+
protected IFill configureFill(
IFill fill,
PartialPath path,
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
index 60c1a79..72b8cf7 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/LastQueryExecutor.java
@@ -172,12 +172,14 @@ public class LastQueryExecutor {
// Acquire query resources for the rest series paths
List<LastPointReader> readerList = new ArrayList<>();
- List<StorageGroupProcessor> list = StorageEngine.getInstance().mergeLock(nonCachedPaths);
+ List<StorageGroupProcessor> list =
+ StorageEngine.getInstance()
+ .mergeLockAndInitQueryDataSource(nonCachedPaths, context, filter);
try {
for (int i = 0; i < nonCachedPaths.size(); i++) {
QueryDataSource dataSource =
QueryResourceManager.getInstance()
- .getQueryDataSourceByPath(nonCachedPaths.get(i), context, filter);
+ .getQueryDataSource(nonCachedPaths.get(i), context, filter);
LastPointReader lastReader =
new LastPointReader(
nonCachedPaths.get(i),
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
index d480d97..03f8e37 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/RawDataQueryExecutor.java
@@ -99,7 +99,8 @@ public class RawDataQueryExecutor {
List<ManagedSeriesReader> readersOfSelectedSeries = new ArrayList<>();
List<StorageGroupProcessor> list =
- StorageEngine.getInstance().mergeLock(queryPlan.getDeduplicatedPaths());
+ StorageEngine.getInstance()
+ .mergeLockAndInitQueryDataSource(queryPlan.getDeduplicatedPaths(), context, timeFilter);
try {
for (int i = 0; i < queryPlan.getDeduplicatedPaths().size(); i++) {
PartialPath path = queryPlan.getDeduplicatedPaths().get(i);
@@ -163,7 +164,8 @@ public class RawDataQueryExecutor {
throws QueryProcessException, StorageEngineException {
List<IReaderByTimestamp> readersOfSelectedSeries = new ArrayList<>();
List<StorageGroupProcessor> list =
- StorageEngine.getInstance().mergeLock(queryPlan.getDeduplicatedPaths());
+ StorageEngine.getInstance()
+ .mergeLockAndInitQueryDataSource(queryPlan.getDeduplicatedPaths(), context, null);
try {
for (int i = 0; i < queryPlan.getDeduplicatedPaths().size(); i++) {
if (cached.get(i)) {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LinearFill.java b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LinearFill.java
index a26109b..47e765b 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LinearFill.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/LinearFill.java
@@ -142,8 +142,7 @@ public class LinearFill extends IFill {
protected TimeValuePair calculatePrecedingPoint()
throws QueryProcessException, StorageEngineException, IOException {
QueryDataSource dataSource =
- QueryResourceManager.getInstance()
- .getQueryDataSourceByPath(seriesPath, context, beforeFilter);
+ QueryResourceManager.getInstance().getQueryDataSource(seriesPath, context, beforeFilter);
LastPointReader lastReader =
new LastPointReader(
seriesPath, dataType, deviceMeasurements, context, dataSource, queryTime, beforeFilter);
@@ -153,9 +152,6 @@ public class LinearFill extends IFill {
protected TimeValuePair calculateSucceedingPoint()
throws IOException, StorageEngineException, QueryProcessException {
-
- QueryResourceManager.getInstance().clearCachedQueryDataSource(seriesPath, context);
-
List<AggregateResult> aggregateResultList = new ArrayList<>();
AggregateResult minTimeResult = new MinTimeAggrResult();
AggregateResult firstValueResult = new FirstValueAggrResult(dataType);
diff --git a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/PreviousFill.java b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/PreviousFill.java
index 979c8fc..1cfd0c9 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/executor/fill/PreviousFill.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/executor/fill/PreviousFill.java
@@ -101,8 +101,7 @@ public class PreviousFill extends IFill {
public TimeValuePair getFillResult()
throws IOException, QueryProcessException, StorageEngineException {
QueryDataSource dataSource =
- QueryResourceManager.getInstance()
- .getQueryDataSourceByPath(seriesPath, context, timeFilter);
+ QueryResourceManager.getInstance().getQueryDataSource(seriesPath, context, timeFilter);
// update filter by TTL
timeFilter = dataSource.updateFilterUsingTTL(timeFilter);
LastPointReader lastReader =
diff --git a/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/metadata/MemChunkMetadataLoader.java b/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/metadata/MemChunkMetadataLoader.java
index 0d7b1d5..6ee4f57 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/metadata/MemChunkMetadataLoader.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/reader/chunk/metadata/MemChunkMetadataLoader.java
@@ -46,11 +46,11 @@ public class MemChunkMetadataLoader implements IChunkMetadataLoader {
@Override
public List<ChunkMetadata> loadChunkMetadataList(TimeseriesMetadata timeseriesMetadata) {
- List<ChunkMetadata> chunkMetadataList = resource.getChunkMetadataList();
+ List<ChunkMetadata> chunkMetadataList = resource.getChunkMetadataList(seriesPath);
DiskChunkMetadataLoader.setDiskChunkLoader(chunkMetadataList, resource, seriesPath, context);
- List<ReadOnlyMemChunk> memChunks = resource.getReadOnlyMemChunk();
+ List<ReadOnlyMemChunk> memChunks = resource.getReadOnlyMemChunk(seriesPath);
if (memChunks != null) {
for (ReadOnlyMemChunk readOnlyMemChunk : memChunks) {
if (!memChunks.isEmpty()) {
diff --git a/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java b/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
index 4cd1f0b..8bba620 100644
--- a/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
+++ b/server/src/main/java/org/apache/iotdb/db/query/timegenerator/ServerTimeGenerator.java
@@ -22,6 +22,7 @@ import org.apache.iotdb.db.engine.StorageEngine;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
import org.apache.iotdb.db.exception.StorageEngineException;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.physical.crud.RawDataQueryPlan;
import org.apache.iotdb.db.query.context.QueryContext;
@@ -64,16 +65,17 @@ public class ServerTimeGenerator extends TimeGenerator {
this.queryPlan = queryPlan;
try {
serverConstructNode(queryPlan.getExpression());
- } catch (IOException e) {
+ } catch (IOException | QueryProcessException e) {
throw new StorageEngineException(e);
}
}
public void serverConstructNode(IExpression expression)
- throws IOException, StorageEngineException {
+ throws IOException, StorageEngineException, QueryProcessException {
List<PartialPath> pathList = new ArrayList<>();
getPartialPathFromExpression(expression, pathList);
- List<StorageGroupProcessor> list = StorageEngine.getInstance().mergeLock(pathList);
+ List<StorageGroupProcessor> list =
+ StorageEngine.getInstance().mergeLockAndInitQueryDataSource(pathList, context, null);
try {
operatorNode = construct(expression);
} finally {
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
index 2610dec..7bbe051 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/FileLoaderUtils.java
@@ -116,7 +116,7 @@ public class FileLoaderUtils {
new DiskChunkMetadataLoader(resource, seriesPath, context, filter));
}
} else {
- timeSeriesMetadata = resource.getTimeSeriesMetadata();
+ timeSeriesMetadata = resource.getTimeSeriesMetadata(seriesPath);
if (timeSeriesMetadata != null) {
timeSeriesMetadata.setChunkMetadataLoader(
new MemChunkMetadataLoader(resource, seriesPath, context, filter));
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/QueryUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/QueryUtils.java
index 90400e6..29fab5a 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/QueryUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/QueryUtils.java
@@ -23,7 +23,6 @@ import org.apache.iotdb.db.engine.modification.Deletion;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
-import org.apache.iotdb.db.query.filter.TsFileFilter;
import org.apache.iotdb.tsfile.file.metadata.ChunkMetadata;
import org.apache.iotdb.tsfile.read.common.TimeRange;
@@ -97,22 +96,9 @@ public class QueryUtils {
}
}
- // remove files that do not satisfy the filter
- public static void filterQueryDataSource(
- QueryDataSource queryDataSource, TsFileFilter fileFilter) {
- if (fileFilter == null) {
- return;
- }
- List<TsFileResource> seqResources = queryDataSource.getSeqResources();
- List<TsFileResource> unseqResources = queryDataSource.getUnseqResources();
- seqResources.removeIf(fileFilter::fileNotSatisfy);
- unseqResources.removeIf(fileFilter::fileNotSatisfy);
- }
-
public static void fillOrderIndexes(
QueryDataSource dataSource, String deviceId, boolean ascending) {
List<TsFileResource> unseqResources = dataSource.getUnseqResources();
- TsFileResource unclosedUnseqResource = dataSource.getUnclosedUnseqResource();
int[] orderIndex = new int[unseqResources.size() + 1];
AtomicInteger index = new AtomicInteger();
Map<Integer, Long> intToOrderTimeMap =
@@ -121,9 +107,6 @@ public class QueryUtils {
Collectors.toMap(
key -> index.getAndIncrement(),
resource -> resource.getOrderTime(deviceId, ascending)));
- if (unclosedUnseqResource != null) {
- intToOrderTimeMap.put(index.get(), unclosedUnseqResource.getOrderTime(deviceId, ascending));
- }
index.set(0);
intToOrderTimeMap.entrySet().stream()
.sorted(
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/merge/ConcurrentMergeTest.java b/server/src/test/java/org/apache/iotdb/db/engine/merge/ConcurrentMergeTest.java
index c9a187b..4e369fc 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/merge/ConcurrentMergeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/merge/ConcurrentMergeTest.java
@@ -46,6 +46,8 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Collections;
+
public class ConcurrentMergeTest {
private static final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
private static Logger logger = LoggerFactory.getLogger(StorageGroupProcessorTest.class);
@@ -113,7 +115,11 @@ public class ConcurrentMergeTest {
}
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(5, queryDataSource.getSeqResources().size());
for (TsFileResource resource : queryDataSource.getSeqResources()) {
Assert.assertTrue(resource.isClosed());
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java
index 200354f..1f48367 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/modification/DeletionFileNodeTest.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.engine.modification.io.LocalTextModificationAccessor;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.db.engine.storagegroup.StorageGroupProcessor;
+import org.apache.iotdb.db.engine.storagegroup.TsFileResource;
import org.apache.iotdb.db.exception.StorageEngineException;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
@@ -126,14 +127,18 @@ public class DeletionFileNodeTest {
new PartialPath(processorName + TsFileConstant.PATH_SEPARATOR + measurements[5]), null);
List<StorageGroupProcessor> list =
StorageEngine.getInstance()
- .mergeLock(Collections.singletonList((PartialPath) expression.getSeriesPath()));
+ .mergeLockAndInitQueryDataSource(
+ Collections.singletonList((PartialPath) expression.getSeriesPath()),
+ TEST_QUERY_CONTEXT,
+ null);
try {
QueryDataSource dataSource =
QueryResourceManager.getInstance()
- .getQueryDataSourceByPath(
+ .getQueryDataSource(
(PartialPath) expression.getSeriesPath(), TEST_QUERY_CONTEXT, null);
+ TsFileResource tsFileResource = dataSource.getSeqResources().get(0);
List<ReadOnlyMemChunk> timeValuePairs =
- dataSource.getSeqResources().get(0).getReadOnlyMemChunk();
+ tsFileResource.getReadOnlyMemChunk((PartialPath) expression.getSeriesPath());
int count = 0;
for (ReadOnlyMemChunk chunk : timeValuePairs) {
IPointReader iterator = chunk.getPointReader();
@@ -250,16 +255,19 @@ public class DeletionFileNodeTest {
List<StorageGroupProcessor> list =
StorageEngine.getInstance()
- .mergeLock(Collections.singletonList((PartialPath) expression.getSeriesPath()));
+ .mergeLockAndInitQueryDataSource(
+ Collections.singletonList((PartialPath) expression.getSeriesPath()),
+ TEST_QUERY_CONTEXT,
+ null);
try {
QueryDataSource dataSource =
QueryResourceManager.getInstance()
- .getQueryDataSourceByPath(
+ .getQueryDataSource(
(PartialPath) expression.getSeriesPath(), TEST_QUERY_CONTEXT, null);
-
+ TsFileResource tsFileResource = dataSource.getUnseqResources().get(0);
List<ReadOnlyMemChunk> timeValuePairs =
- dataSource.getUnseqResources().get(0).getReadOnlyMemChunk();
+ tsFileResource.getReadOnlyMemChunk((PartialPath) expression.getSeriesPath());
int count = 0;
for (ReadOnlyMemChunk chunk : timeValuePairs) {
IPointReader iterator = chunk.getPointReader();
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
index 6493755..b632143 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/StorageGroupProcessorTest.java
@@ -127,18 +127,18 @@ public class StorageGroupProcessorTest {
List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>();
for (TsFileProcessor tsfileProcessor : processor.getWorkUnsequenceTsFileProcessors()) {
tsfileProcessor.query(
- deviceId,
- measurementId,
- TSDataType.INT32,
- TSEncoding.RLE,
- Collections.emptyMap(),
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
new QueryContext(),
tsfileResourcesForQuery);
}
Assert.assertEquals(1, tsfileResourcesForQuery.size());
- Assert.assertEquals(0, tsfileResourcesForQuery.get(0).getChunkMetadataList().size());
- List<ReadOnlyMemChunk> memChunks = tsfileResourcesForQuery.get(0).getReadOnlyMemChunk();
+
+ TsFileResource tsfileResource = tsfileResourcesForQuery.get(0);
+ Assert.assertEquals(
+ 0, tsfileResource.getChunkMetadataList(new PartialPath(deviceId, measurementId)).size());
+ List<ReadOnlyMemChunk> memChunks =
+ tsfileResource.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId));
long time = 16;
for (ReadOnlyMemChunk memChunk : memChunks) {
IPointReader iterator = memChunk.getPointReader();
@@ -166,7 +166,11 @@ public class StorageGroupProcessorTest {
}
processor.syncCloseAllWorkingTsFileProcessors();
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(10, queryDataSource.getSeqResources().size());
for (TsFileResource resource : queryDataSource.getSeqResources()) {
Assert.assertTrue(resource.isClosed());
@@ -195,7 +199,11 @@ public class StorageGroupProcessorTest {
processor.syncCloseAllWorkingTsFileProcessors();
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(0, queryDataSource.getUnseqResources().size());
}
@@ -256,7 +264,11 @@ public class StorageGroupProcessorTest {
processor.syncCloseAllWorkingTsFileProcessors();
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(2, queryDataSource.getSeqResources().size());
Assert.assertEquals(1, queryDataSource.getUnseqResources().size());
@@ -286,7 +298,11 @@ public class StorageGroupProcessorTest {
processor.syncCloseAllWorkingTsFileProcessors();
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(10, queryDataSource.getSeqResources().size());
Assert.assertEquals(10, queryDataSource.getUnseqResources().size());
for (TsFileResource resource : queryDataSource.getSeqResources()) {
@@ -325,7 +341,11 @@ public class StorageGroupProcessorTest {
}
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(10, queryDataSource.getSeqResources().size());
Assert.assertEquals(0, queryDataSource.getUnseqResources().size());
for (TsFileResource resource : queryDataSource.getSeqResources()) {
@@ -406,7 +426,11 @@ public class StorageGroupProcessorTest {
}
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(2, queryDataSource.getSeqResources().size());
Assert.assertEquals(0, queryDataSource.getUnseqResources().size());
@@ -487,7 +511,11 @@ public class StorageGroupProcessorTest {
}
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(2, queryDataSource.getSeqResources().size());
Assert.assertEquals(0, queryDataSource.getUnseqResources().size());
@@ -568,7 +596,11 @@ public class StorageGroupProcessorTest {
}
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(2, queryDataSource.getSeqResources().size());
Assert.assertEquals(0, queryDataSource.getUnseqResources().size());
@@ -606,7 +638,11 @@ public class StorageGroupProcessorTest {
}
QueryDataSource queryDataSource =
- processor.query(new PartialPath(deviceId, measurementId), context, null, null);
+ processor.query(
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ null,
+ null);
Assert.assertEquals(10, queryDataSource.getSeqResources().size());
for (TsFileResource resource : queryDataSource.getSeqResources()) {
Assert.assertTrue(resource.isClosed());
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
index 1f096ee..651f290 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TTLTest.java
@@ -223,7 +223,10 @@ public class TTLTest {
// files before ttl
QueryDataSource dataSource =
storageGroupProcessor.query(
- new PartialPath(sg1, s1), EnvironmentUtils.TEST_QUERY_CONTEXT, null, null);
+ Collections.singletonList(new PartialPath(sg1, s1)),
+ EnvironmentUtils.TEST_QUERY_CONTEXT,
+ null,
+ null);
List<TsFileResource> seqResource = dataSource.getSeqResources();
List<TsFileResource> unseqResource = dataSource.getUnseqResources();
assertEquals(4, seqResource.size());
@@ -234,7 +237,10 @@ public class TTLTest {
// files after ttl
dataSource =
storageGroupProcessor.query(
- new PartialPath(sg1, s1), EnvironmentUtils.TEST_QUERY_CONTEXT, null, null);
+ Collections.singletonList(new PartialPath(sg1, s1)),
+ EnvironmentUtils.TEST_QUERY_CONTEXT,
+ null,
+ null);
seqResource = dataSource.getSeqResources();
unseqResource = dataSource.getUnseqResources();
assertTrue(seqResource.size() < 4);
@@ -269,7 +275,10 @@ public class TTLTest {
storageGroupProcessor.setDataTTL(0);
dataSource =
storageGroupProcessor.query(
- new PartialPath(sg1, s1), EnvironmentUtils.TEST_QUERY_CONTEXT, null, null);
+ Collections.singletonList(new PartialPath(sg1, s1)),
+ EnvironmentUtils.TEST_QUERY_CONTEXT,
+ null,
+ null);
seqResource = dataSource.getSeqResources();
unseqResource = dataSource.getUnseqResources();
assertEquals(0, seqResource.size());
diff --git a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessorTest.java b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessorTest.java
index 4bee9b4..00f3748 100644
--- a/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessorTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/engine/storagegroup/TsFileProcessorTest.java
@@ -25,6 +25,7 @@ import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.db.exception.TsFileProcessorException;
import org.apache.iotdb.db.exception.WriteProcessException;
import org.apache.iotdb.db.exception.metadata.MetadataException;
+import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.rescon.SystemInfo;
@@ -110,7 +111,9 @@ public class TsFileProcessorTest {
SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor);
List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
assertTrue(tsfileResourcesForQuery.isEmpty());
for (int i = 1; i <= 100; i++) {
@@ -122,9 +125,15 @@ public class TsFileProcessorTest {
// query data in memory
tsfileResourcesForQuery.clear();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
- assertFalse(tsfileResourcesForQuery.get(0).getReadOnlyMemChunk().isEmpty());
- List<ReadOnlyMemChunk> memChunks = tsfileResourcesForQuery.get(0).getReadOnlyMemChunk();
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
+
+ TsFileResource tsFileResource = tsfileResourcesForQuery.get(0);
+ assertFalse(
+ tsFileResource.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId)).isEmpty());
+ List<ReadOnlyMemChunk> memChunks =
+ tsFileResource.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId));
for (ReadOnlyMemChunk chunk : memChunks) {
IPointReader iterator = chunk.getPointReader();
for (int num = 1; num <= 100; num++) {
@@ -140,14 +149,20 @@ public class TsFileProcessorTest {
tsfileResourcesForQuery.clear();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
- assertTrue(tsfileResourcesForQuery.get(0).getReadOnlyMemChunk().isEmpty());
- assertEquals(1, tsfileResourcesForQuery.get(0).getChunkMetadataList().size());
- assertEquals(
- measurementId,
- tsfileResourcesForQuery.get(0).getChunkMetadataList().get(0).getMeasurementUid());
- assertEquals(
- dataType, tsfileResourcesForQuery.get(0).getChunkMetadataList().get(0).getDataType());
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
+
+ TsFileResource tsFileResource1 = tsfileResourcesForQuery.get(0);
+ List<ReadOnlyMemChunk> readOnlyMemChunk =
+ tsFileResource1.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId));
+ List<ChunkMetadata> chunkMetadataList =
+ tsFileResource1.getChunkMetadataList(new PartialPath(deviceId, measurementId));
+
+ assertTrue(readOnlyMemChunk.isEmpty());
+ assertEquals(1, chunkMetadataList.size());
+ assertEquals(measurementId, chunkMetadataList.get(0).getMeasurementUid());
+ assertEquals(dataType, chunkMetadataList.get(0).getDataType());
processor.syncClose();
}
@@ -172,7 +187,9 @@ public class TsFileProcessorTest {
SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor);
List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
assertTrue(tsfileResourcesForQuery.isEmpty());
for (int i = 1; i <= 100; i++) {
@@ -184,10 +201,16 @@ public class TsFileProcessorTest {
// query data in memory
tsfileResourcesForQuery.clear();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
- assertFalse(tsfileResourcesForQuery.get(0).getReadOnlyMemChunk().isEmpty());
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
+
+ TsFileResource tsFileResource = tsfileResourcesForQuery.get(0);
+ assertFalse(
+ tsFileResource.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId)).isEmpty());
int num = 1;
- List<ReadOnlyMemChunk> memChunks = tsfileResourcesForQuery.get(0).getReadOnlyMemChunk();
+ List<ReadOnlyMemChunk> memChunks =
+ tsFileResource.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId));
for (ReadOnlyMemChunk chunk : memChunks) {
IPointReader iterator = chunk.getPointReader();
for (; num <= 100; num++) {
@@ -203,14 +226,20 @@ public class TsFileProcessorTest {
tsfileResourcesForQuery.clear();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
- assertTrue(tsfileResourcesForQuery.get(0).getReadOnlyMemChunk().isEmpty());
- assertEquals(1, tsfileResourcesForQuery.get(0).getChunkMetadataList().size());
- assertEquals(
- measurementId,
- tsfileResourcesForQuery.get(0).getChunkMetadataList().get(0).getMeasurementUid());
- assertEquals(
- dataType, tsfileResourcesForQuery.get(0).getChunkMetadataList().get(0).getDataType());
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
+
+ TsFileResource tsFileResource1 = tsfileResourcesForQuery.get(0);
+ List<ReadOnlyMemChunk> readOnlyMemChunk =
+ tsFileResource1.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId));
+ List<ChunkMetadata> chunkMetadataList =
+ tsFileResource1.getChunkMetadataList(new PartialPath(deviceId, measurementId));
+
+ assertTrue(readOnlyMemChunk.isEmpty());
+ assertEquals(1, chunkMetadataList.size());
+ assertEquals(measurementId, chunkMetadataList.get(0).getMeasurementUid());
+ assertEquals(dataType, chunkMetadataList.get(0).getDataType());
RestorableTsFileIOWriter tsFileIOWriter = processor.getWriter();
Map<String, List<ChunkMetadata>> chunkMetaDataListInChunkGroups =
@@ -264,7 +293,9 @@ public class TsFileProcessorTest {
SystemInfo.getInstance().reportStorageGroupStatus(sgInfo, processor);
List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
assertTrue(tsfileResourcesForQuery.isEmpty());
for (int flushId = 0; flushId < 10; flushId++) {
@@ -279,15 +310,20 @@ public class TsFileProcessorTest {
tsfileResourcesForQuery.clear();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
assertFalse(tsfileResourcesForQuery.isEmpty());
- assertTrue(tsfileResourcesForQuery.get(0).getReadOnlyMemChunk().isEmpty());
- assertEquals(10, tsfileResourcesForQuery.get(0).getChunkMetadataList().size());
- assertEquals(
- measurementId,
- tsfileResourcesForQuery.get(0).getChunkMetadataList().get(0).getMeasurementUid());
- assertEquals(
- dataType, tsfileResourcesForQuery.get(0).getChunkMetadataList().get(0).getDataType());
+
+ TsFileResource tsFileResource = tsfileResourcesForQuery.get(0);
+ List<ReadOnlyMemChunk> readOnlyMemChunk =
+ tsFileResource.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId));
+ List<ChunkMetadata> chunkMetadataList =
+ tsFileResource.getChunkMetadataList(new PartialPath(deviceId, measurementId));
+ assertTrue(readOnlyMemChunk.isEmpty());
+ assertEquals(10, chunkMetadataList.size());
+ assertEquals(measurementId, chunkMetadataList.get(0).getMeasurementUid());
+ assertEquals(dataType, chunkMetadataList.get(0).getDataType());
processor.syncClose();
}
@@ -312,7 +348,9 @@ public class TsFileProcessorTest {
List<TsFileResource> tsfileResourcesForQuery = new ArrayList<>();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
assertTrue(tsfileResourcesForQuery.isEmpty());
for (int i = 1; i <= 100; i++) {
@@ -324,10 +362,16 @@ public class TsFileProcessorTest {
// query data in memory
tsfileResourcesForQuery.clear();
processor.query(
- deviceId, measurementId, dataType, encoding, props, context, tsfileResourcesForQuery);
+ Collections.singletonList(new PartialPath(deviceId, measurementId)),
+ context,
+ tsfileResourcesForQuery);
assertFalse(tsfileResourcesForQuery.isEmpty());
- assertFalse(tsfileResourcesForQuery.get(0).getReadOnlyMemChunk().isEmpty());
- List<ReadOnlyMemChunk> memChunks = tsfileResourcesForQuery.get(0).getReadOnlyMemChunk();
+
+ TsFileResource tsFileResource = tsfileResourcesForQuery.get(0);
+ assertFalse(
+ tsFileResource.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId)).isEmpty());
+ List<ReadOnlyMemChunk> memChunks =
+ tsFileResource.getReadOnlyMemChunk(new PartialPath(deviceId, measurementId));
for (ReadOnlyMemChunk chunk : memChunks) {
IPointReader iterator = chunk.getPointReader();
for (int num = 1; num <= 100; num++) {
diff --git a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAlignByDeviceIT.java b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAlignByDeviceIT.java
index 3782763..a535ce9 100644
--- a/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAlignByDeviceIT.java
+++ b/server/src/test/java/org/apache/iotdb/db/integration/IoTDBAlignByDeviceIT.java
@@ -154,7 +154,7 @@ public class IoTDBAlignByDeviceIT {
"1000,root.vehicle.d0,22222,55555,1000.11,null,null,",
"946684800000,root.vehicle.d0,null,100,null,good,null,",
"1,root.vehicle.d1,999,null,null,null,null,",
- "1000,root.vehicle.d1,888,null,null,null,null,",
+ "1000,root.vehicle.d1,888,null,null,null,null,"
};
Class.forName(Config.JDBC_DRIVER_NAME);
diff --git a/server/src/test/java/org/apache/iotdb/db/qp/physical/InsertTabletPlanTest.java b/server/src/test/java/org/apache/iotdb/db/qp/physical/InsertTabletPlanTest.java
index 112e003..e665ec3 100644
--- a/server/src/test/java/org/apache/iotdb/db/qp/physical/InsertTabletPlanTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/qp/physical/InsertTabletPlanTest.java
@@ -30,7 +30,6 @@ import org.apache.iotdb.db.qp.physical.crud.CreateTemplatePlan;
import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.db.qp.physical.crud.QueryPlan;
import org.apache.iotdb.db.qp.physical.crud.SetDeviceTemplatePlan;
-import org.apache.iotdb.db.query.control.QueryResourceManager;
import org.apache.iotdb.db.service.IoTDB;
import org.apache.iotdb.db.utils.EnvironmentUtils;
import org.apache.iotdb.tsfile.exception.filter.QueryFilterOptimizationException;
@@ -210,9 +209,6 @@ public class InsertTabletPlanTest {
EnvironmentUtils.activeDaemon();
queryPlan = (QueryPlan) processor.parseSQLToPhysicalPlan("select * from root.isp.d1");
- QueryResourceManager.getInstance()
- .clearCachedQueryDataSource(
- queryPlan.getPaths().get(0), EnvironmentUtils.TEST_QUERY_CONTEXT);
dataSet = executor.processQuery(queryPlan, EnvironmentUtils.TEST_QUERY_CONTEXT);
Assert.assertEquals(3, dataSet.getPaths().size());
while (dataSet.hasNext()) {