You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by hx...@apache.org on 2019/03/10 15:03:06 UTC
[incubator-iotdb] 01/01: rename OverflowSupport to OverflowMemtable;
rename IntervalFileNode to TsFileResource; delete OverflowSeriesImpl
This is an automated email from the ASF dual-hosted git repository.
hxd pushed a commit to branch refactor_overflow
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 4b35aa3c3a950421ee0792352d32714b2cc38aad
Author: xiangdong huang <sa...@gmail.com>
AuthorDate: Sun Mar 10 23:02:43 2019 +0800
rename OverflowSupport to OverflowMemtable; rename IntervalFileNode to TsFileResource; delete OverflowSeriesImpl
---
.../iotdb/db/engine/filenode/FileNodeManager.java | 4 +-
.../db/engine/filenode/FileNodeProcessor.java | 232 ++++++++++-----------
.../db/engine/filenode/FileNodeProcessorStore.java | 26 +--
.../{IntervalFileNode.java => TsFileResource.java} | 18 +-
...{OverflowSupport.java => OverflowMemtable.java} | 24 +--
.../db/engine/overflow/io/OverflowProcessor.java | 8 +-
.../db/engine/overflow/io/OverflowSeriesImpl.java | 79 -------
.../querycontext/GlobalSortedSeriesDataSource.java | 11 +-
.../db/postback/receiver/ServerServiceImpl.java | 4 +-
.../db/query/control/OpenedFilePathsManager.java | 6 +-
.../db/query/factory/SeriesReaderFactory.java | 10 +-
.../query/reader/sequence/SealedTsFilesReader.java | 14 +-
.../db/engine/filenode/SerializeUtilTest.java | 16 +-
...wSupportTest.java => OverflowMemtableTest.java} | 4 +-
.../engine/overflow/io/OverflowResourceTest.java | 2 +-
.../db/engine/overflow/io/OverflowTestUtils.java | 2 +-
16 files changed, 184 insertions(+), 276 deletions(-)
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java
index 4135de3..eec4942 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeManager.java
@@ -700,7 +700,7 @@ public class FileNodeManager implements IStatistic, IService {
* @param fileNodeName the seriesPath of storage group
* @param appendFile the appended tsfile information
*/
- public boolean appendFileToFileNode(String fileNodeName, IntervalFileNode appendFile,
+ public boolean appendFileToFileNode(String fileNodeName, TsFileResource appendFile,
String appendFilePath) throws FileNodeManagerException {
FileNodeProcessor fileNodeProcessor = getProcessor(fileNodeName, true);
try {
@@ -729,7 +729,7 @@ public class FileNodeManager implements IStatistic, IService {
* @param fileNodeName the seriesPath of storage group
* @param appendFile the appended tsfile information
*/
- public List<String> getOverlapFilesFromFileNode(String fileNodeName, IntervalFileNode appendFile,
+ public List<String> getOverlapFilesFromFileNode(String fileNodeName, TsFileResource appendFile,
String uuid) throws FileNodeManagerException {
FileNodeProcessor fileNodeProcessor = getProcessor(fileNodeName, true);
List<String> overlapFiles;
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
index 0d43be8..6f56ebc 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
@@ -76,7 +76,6 @@ import org.apache.iotdb.db.monitor.StatMonitor;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.factory.SeriesReaderFactory;
import org.apache.iotdb.db.query.reader.IReader;
-import org.apache.iotdb.db.utils.FileSchemaUtils;
import org.apache.iotdb.db.utils.MemUtils;
import org.apache.iotdb.db.utils.QueryUtils;
import org.apache.iotdb.db.utils.TimeValuePair;
@@ -98,11 +97,8 @@ import org.apache.iotdb.tsfile.write.chunk.ChunkWriterImpl;
import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.record.datapoint.LongDataPoint;
import org.apache.iotdb.tsfile.write.schema.FileSchema;
-import org.apache.iotdb.tsfile.write.schema.JsonConverter;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.apache.iotdb.tsfile.write.writer.TsFileIOWriter;
-import org.json.JSONArray;
-import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -125,10 +121,10 @@ public class FileNodeProcessor extends Processor implements IStatistic {
private volatile boolean isOverflowed;
private Map<String, Long> lastUpdateTimeMap;
private Map<String, Long> flushLastUpdateTimeMap;
- private Map<String, List<IntervalFileNode>> invertedIndexOfFiles;
- private IntervalFileNode emptyIntervalFileNode;
- private IntervalFileNode currentIntervalFileNode;
- private List<IntervalFileNode> newFileNodes;
+ private Map<String, List<TsFileResource>> invertedIndexOfFiles;
+ private TsFileResource emptyTsFileResource;
+ private TsFileResource currentTsFileResource;
+ private List<TsFileResource> newFileNodes;
private FileNodeProcessorStatus isMerging;
// this is used when work->merge operation
private int numOfMergeFile;
@@ -187,21 +183,21 @@ public class FileNodeProcessor extends Processor implements IStatistic {
if (!newFileNodes.isEmpty()) {
// end time with one start time
Map<String, Long> endTimeMap = new HashMap<>();
- for (Entry<String, Long> startTime : currentIntervalFileNode.getStartTimeMap().entrySet()) {
+ for (Entry<String, Long> startTime : currentTsFileResource.getStartTimeMap().entrySet()) {
String deviceId = startTime.getKey();
endTimeMap.put(deviceId, lastUpdateTimeMap.get(deviceId));
}
- currentIntervalFileNode.setEndTimeMap(endTimeMap);
+ currentTsFileResource.setEndTimeMap(endTimeMap);
}
}
};
private Action overflowFlushAction = () -> {
- // update the new IntervalFileNode List and emptyIntervalFile.
+ // update the new TsFileResource List and emptyIntervalFile.
// Notice: thread safe
synchronized (fileNodeProcessorStore) {
fileNodeProcessorStore.setOverflowed(isOverflowed);
- fileNodeProcessorStore.setEmptyIntervalFileNode(emptyIntervalFileNode);
+ fileNodeProcessorStore.setEmptyTsFileResource(emptyTsFileResource);
fileNodeProcessorStore.setNewFileNodes(newFileNodes);
}
};
@@ -264,7 +260,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
}
// TODO deep clone the lastupdate time
lastUpdateTimeMap = fileNodeProcessorStore.getLastUpdateTimeMap();
- emptyIntervalFileNode = fileNodeProcessorStore.getEmptyIntervalFileNode();
+ emptyTsFileResource = fileNodeProcessorStore.getEmptyTsFileResource();
newFileNodes = fileNodeProcessorStore.getNewFileNodes();
isMerging = fileNodeProcessorStore.getFileNodeProcessorStatus();
numOfMergeFile = fileNodeProcessorStore.getNumOfMergeFile();
@@ -350,10 +346,10 @@ public class FileNodeProcessor extends Processor implements IStatistic {
*/
void addIntervalFileNode(String baseDir, String fileName) throws ActionException {
- IntervalFileNode intervalFileNode = new IntervalFileNode(OverflowChangeType.NO_CHANGE, baseDir,
+ TsFileResource tsFileResource = new TsFileResource(OverflowChangeType.NO_CHANGE, baseDir,
fileName);
- this.currentIntervalFileNode = intervalFileNode;
- newFileNodes.add(intervalFileNode);
+ this.currentTsFileResource = tsFileResource;
+ newFileNodes.add(tsFileResource);
fileNodeProcessorStore.setNewFileNodes(newFileNodes);
flushFileNodeProcessorAction.act();
}
@@ -364,12 +360,12 @@ public class FileNodeProcessor extends Processor implements IStatistic {
* @param deviceId device ID
*/
void setIntervalFileNodeStartTime(String deviceId) {
- if (currentIntervalFileNode.getStartTime(deviceId) == -1) {
- currentIntervalFileNode.setStartTime(deviceId, flushLastUpdateTimeMap.get(deviceId));
+ if (currentTsFileResource.getStartTime(deviceId) == -1) {
+ currentTsFileResource.setStartTime(deviceId, flushLastUpdateTimeMap.get(deviceId));
if (!invertedIndexOfFiles.containsKey(deviceId)) {
invertedIndexOfFiles.put(deviceId, new ArrayList<>());
}
- invertedIndexOfFiles.get(deviceId).add(currentIntervalFileNode);
+ invertedIndexOfFiles.get(deviceId).add(currentTsFileResource);
}
}
@@ -378,7 +374,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
*/
public void clearFileNode() {
isOverflowed = false;
- emptyIntervalFileNode = new IntervalFileNode(OverflowChangeType.NO_CHANGE, null);
+ emptyTsFileResource = new TsFileResource(OverflowChangeType.NO_CHANGE, null);
newFileNodes = new ArrayList<>();
isMerging = FileNodeProcessorStatus.NONE;
numOfMergeFile = 0;
@@ -386,14 +382,14 @@ public class FileNodeProcessor extends Processor implements IStatistic {
fileNodeProcessorStore.setFileNodeProcessorStatus(isMerging);
fileNodeProcessorStore.setNewFileNodes(newFileNodes);
fileNodeProcessorStore.setNumOfMergeFile(numOfMergeFile);
- fileNodeProcessorStore.setEmptyIntervalFileNode(emptyIntervalFileNode);
+ fileNodeProcessorStore.setEmptyTsFileResource(emptyTsFileResource);
}
- private void addAllFileIntoIndex(List<IntervalFileNode> fileList) {
+ private void addAllFileIntoIndex(List<TsFileResource> fileList) {
// clear map
invertedIndexOfFiles.clear();
// add all file to index
- for (IntervalFileNode fileNode : fileList) {
+ for (TsFileResource fileNode : fileList) {
if (fileNode.getStartTimeMap().isEmpty()) {
continue;
}
@@ -437,7 +433,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
//
// add the current file
//
- currentIntervalFileNode = newFileNodes.get(newFileNodes.size() - 1);
+ currentTsFileResource = newFileNodes.get(newFileNodes.size() - 1);
// this bufferwrite file is not close by normal operation
String damagedFilePath = newFileNodes.get(newFileNodes.size() - 1).getFilePath();
@@ -626,17 +622,17 @@ public class FileNodeProcessor extends Processor implements IStatistic {
WARN_NO_SUCH_OVERFLOWED_FILE
+ "the data is [device:{},time:{}]",
getProcessorName(), deviceId, timestamp);
- emptyIntervalFileNode.setStartTime(deviceId, 0L);
- emptyIntervalFileNode.setEndTime(deviceId, getLastUpdateTime(deviceId));
- emptyIntervalFileNode.changeTypeToChanged(isMerging);
+ emptyTsFileResource.setStartTime(deviceId, 0L);
+ emptyTsFileResource.setEndTime(deviceId, getLastUpdateTime(deviceId));
+ emptyTsFileResource.changeTypeToChanged(isMerging);
} else {
- List<IntervalFileNode> temp = invertedIndexOfFiles.get(deviceId);
+ List<TsFileResource> temp = invertedIndexOfFiles.get(deviceId);
int index = searchIndexNodeByTimestamp(deviceId, timestamp, temp);
changeTypeToChanged(temp.get(index), deviceId);
}
}
- private void changeTypeToChanged(IntervalFileNode fileNode, String deviceId) {
+ private void changeTypeToChanged(TsFileResource fileNode, String deviceId) {
fileNode.changeTypeToChanged(isMerging);
if (isMerging == FileNodeProcessorStatus.MERGING_WRITE) {
fileNode.addMergeChanged(deviceId);
@@ -652,11 +648,11 @@ public class FileNodeProcessor extends Processor implements IStatistic {
WARN_NO_SUCH_OVERFLOWED_FILE
+ "the data is [device:{}, start time:{}, end time:{}]",
getProcessorName(), deviceId, startTime, endTime);
- emptyIntervalFileNode.setStartTime(deviceId, 0L);
- emptyIntervalFileNode.setEndTime(deviceId, getLastUpdateTime(deviceId));
- emptyIntervalFileNode.changeTypeToChanged(isMerging);
+ emptyTsFileResource.setStartTime(deviceId, 0L);
+ emptyTsFileResource.setEndTime(deviceId, getLastUpdateTime(deviceId));
+ emptyTsFileResource.changeTypeToChanged(isMerging);
} else {
- List<IntervalFileNode> temp = invertedIndexOfFiles.get(deviceId);
+ List<TsFileResource> temp = invertedIndexOfFiles.get(deviceId);
int left = searchIndexNodeByTimestamp(deviceId, startTime, temp);
int right = searchIndexNodeByTimestamp(deviceId, endTime, temp);
for (int i = left; i <= right; i++) {
@@ -674,11 +670,11 @@ public class FileNodeProcessor extends Processor implements IStatistic {
WARN_NO_SUCH_OVERFLOWED_FILE
+ "the data is [device:{}, delete time:{}]",
getProcessorName(), deviceId, timestamp);
- emptyIntervalFileNode.setStartTime(deviceId, 0L);
- emptyIntervalFileNode.setEndTime(deviceId, getLastUpdateTime(deviceId));
- emptyIntervalFileNode.changeTypeToChanged(isMerging);
+ emptyTsFileResource.setStartTime(deviceId, 0L);
+ emptyTsFileResource.setEndTime(deviceId, getLastUpdateTime(deviceId));
+ emptyTsFileResource.changeTypeToChanged(isMerging);
} else {
- List<IntervalFileNode> temp = invertedIndexOfFiles.get(deviceId);
+ List<TsFileResource> temp = invertedIndexOfFiles.get(deviceId);
int index = searchIndexNodeByTimestamp(deviceId, timestamp, temp);
for (int i = 0; i <= index; i++) {
temp.get(i).changeTypeToChanged(isMerging);
@@ -695,7 +691,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
* @return index of interval
*/
private int searchIndexNodeByTimestamp(String deviceId, long timestamp,
- List<IntervalFileNode> fileList) {
+ List<TsFileResource> fileList) {
int index = 1;
while (index < fileList.size()) {
if (timestamp < fileList.get(index).getStartTime(deviceId)) {
@@ -768,11 +764,11 @@ public class FileNodeProcessor extends Processor implements IStatistic {
throw new FileNodeProcessorException(e);
}
// tsfile dataØØ
- List<IntervalFileNode> bufferwriteDataInFiles = new ArrayList<>();
- for (IntervalFileNode intervalFileNode : newFileNodes) {
- // add the same intervalFileNode, but not the same reference
- if (intervalFileNode.isClosed()) {
- bufferwriteDataInFiles.add(intervalFileNode.backUp());
+ List<TsFileResource> bufferwriteDataInFiles = new ArrayList<>();
+ for (TsFileResource tsFileResource : newFileNodes) {
+ // add the same tsFileResource, but not the same reference
+ if (tsFileResource.isClosed()) {
+ bufferwriteDataInFiles.add(tsFileResource.backUp());
}
}
Pair<ReadOnlyMemChunk, List<ChunkMetaData>> bufferwritedata = new Pair<>(null, null);
@@ -798,7 +794,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
try {
List<Modification> pathModifications = context.getPathModifications(
- currentIntervalFileNode.getModFile(), deviceId
+ currentTsFileResource.getModFile(), deviceId
+ IoTDBConstant.PATH_SEPARATOR + measurementId
);
if (!pathModifications.isEmpty()) {
@@ -823,7 +819,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
* @param appendFile the appended tsfile information
* @param appendFilePath the seriesPath of appended file
*/
- public void appendFile(IntervalFileNode appendFile, String appendFilePath)
+ public void appendFile(TsFileResource appendFile, String appendFilePath)
throws FileNodeProcessorException {
try {
if (!new File(appendFile.getFilePath()).getParentFile().exists()) {
@@ -869,12 +865,12 @@ public class FileNodeProcessor extends Processor implements IStatistic {
*
* @param appendFile the appended tsfile information
*/
- public List<String> getOverlapFiles(IntervalFileNode appendFile, String uuid)
+ public List<String> getOverlapFiles(TsFileResource appendFile, String uuid)
throws FileNodeProcessorException {
List<String> overlapFiles = new ArrayList<>();
try {
- for (IntervalFileNode intervalFileNode : newFileNodes) {
- getOverlapFiles(appendFile, intervalFileNode, uuid, overlapFiles);
+ for (TsFileResource tsFileResource : newFileNodes) {
+ getOverlapFiles(appendFile, tsFileResource, uuid, overlapFiles);
}
} catch (IOException e) {
LOGGER.error("Failed to get overlap tsfiles which conflict with the appendFile.");
@@ -883,24 +879,24 @@ public class FileNodeProcessor extends Processor implements IStatistic {
return overlapFiles;
}
- private void getOverlapFiles(IntervalFileNode appendFile, IntervalFileNode intervalFileNode,
+ private void getOverlapFiles(TsFileResource appendFile, TsFileResource tsFileResource,
String uuid, List<String> overlapFiles) throws IOException {
for (Entry<String, Long> entry : appendFile.getStartTimeMap().entrySet()) {
- if (intervalFileNode.getStartTimeMap().containsKey(entry.getKey()) &&
- intervalFileNode.getEndTime(entry.getKey()) >= entry.getValue()
- && intervalFileNode.getStartTime(entry.getKey()) <= appendFile
+ if (tsFileResource.getStartTimeMap().containsKey(entry.getKey()) &&
+ tsFileResource.getEndTime(entry.getKey()) >= entry.getValue()
+ && tsFileResource.getStartTime(entry.getKey()) <= appendFile
.getEndTime(entry.getKey())) {
String relativeFilePath = "postback" + File.separator + uuid + File.separator + "backup"
- + File.separator + intervalFileNode.getRelativePath();
+ + File.separator + tsFileResource.getRelativePath();
File newFile = new File(
- Directories.getInstance().getTsFileFolder(intervalFileNode.getBaseDirIndex()),
+ Directories.getInstance().getTsFileFolder(tsFileResource.getBaseDirIndex()),
relativeFilePath);
if (!newFile.getParentFile().exists()) {
newFile.getParentFile().mkdirs();
}
java.nio.file.Path link = FileSystems.getDefault().getPath(newFile.getPath());
java.nio.file.Path target = FileSystems.getDefault()
- .getPath(intervalFileNode.getFilePath());
+ .getPath(tsFileResource.getFilePath());
Files.createLink(link, target);
overlapFiles.add(newFile.getPath());
break;
@@ -1028,12 +1024,12 @@ public class FileNodeProcessor extends Processor implements IStatistic {
// change status from work to merge
isMerging = FileNodeProcessorStatus.MERGING_WRITE;
// check the empty file
- Map<String, Long> startTimeMap = emptyIntervalFileNode.getStartTimeMap();
+ Map<String, Long> startTimeMap = emptyTsFileResource.getStartTimeMap();
mergeCheckEmptyFile(startTimeMap);
- for (IntervalFileNode intervalFileNode : newFileNodes) {
- if (intervalFileNode.getOverflowChangeType() != OverflowChangeType.NO_CHANGE) {
- intervalFileNode.setOverflowChangeType(OverflowChangeType.CHANGED);
+ for (TsFileResource tsFileResource : newFileNodes) {
+ if (tsFileResource.getOverflowChangeType() != OverflowChangeType.NO_CHANGE) {
+ tsFileResource.setOverflowChangeType(OverflowChangeType.CHANGED);
}
}
@@ -1042,7 +1038,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
fileNodeProcessorStore.setOverflowed(isOverflowed);
fileNodeProcessorStore.setFileNodeProcessorStatus(isMerging);
fileNodeProcessorStore.setNewFileNodes(newFileNodes);
- fileNodeProcessorStore.setEmptyIntervalFileNode(emptyIntervalFileNode);
+ fileNodeProcessorStore.setEmptyTsFileResource(emptyTsFileResource);
// flush this filenode information
try {
writeStoreToDisk(fileNodeProcessorStore);
@@ -1054,17 +1050,17 @@ public class FileNodeProcessor extends Processor implements IStatistic {
}
}
// add numOfMergeFile to control the number of the merge file
- List<IntervalFileNode> backupIntervalFiles;
+ List<TsFileResource> backupIntervalFiles;
backupIntervalFiles = switchFileNodeToMerge();
//
// clear empty file
//
boolean needEmtpy = false;
- if (emptyIntervalFileNode.getOverflowChangeType() != OverflowChangeType.NO_CHANGE) {
+ if (emptyTsFileResource.getOverflowChangeType() != OverflowChangeType.NO_CHANGE) {
needEmtpy = true;
}
- emptyIntervalFileNode.clear();
+ emptyTsFileResource.clear();
// attention
try {
overflowProcessor.switchWorkToMerge();
@@ -1081,7 +1077,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
// query tsfile data and overflow data, and merge them
int numOfMergeFiles = 0;
int allNeedMergeFiles = backupIntervalFiles.size();
- for (IntervalFileNode backupIntervalFile : backupIntervalFiles) {
+ for (TsFileResource backupIntervalFile : backupIntervalFiles) {
numOfMergeFiles++;
if (backupIntervalFile.getOverflowChangeType() == OverflowChangeType.CHANGED) {
// query data and merge
@@ -1133,10 +1129,10 @@ public class FileNodeProcessor extends Processor implements IStatistic {
}
private void mergeCheckEmptyFile(Map<String, Long> startTimeMap) {
- if (emptyIntervalFileNode.getOverflowChangeType() == OverflowChangeType.NO_CHANGE) {
+ if (emptyTsFileResource.getOverflowChangeType() == OverflowChangeType.NO_CHANGE) {
return;
}
- Iterator<Entry<String, Long>> iterator = emptyIntervalFileNode.getEndTimeMap().entrySet()
+ Iterator<Entry<String, Long>> iterator = emptyTsFileResource.getEndTimeMap().entrySet()
.iterator();
while (iterator.hasNext()) {
Entry<String, Long> entry = iterator.next();
@@ -1147,32 +1143,32 @@ public class FileNodeProcessor extends Processor implements IStatistic {
iterator.remove();
}
}
- if (emptyIntervalFileNode.checkEmpty()) {
- emptyIntervalFileNode.clear();
+ if (emptyTsFileResource.checkEmpty()) {
+ emptyTsFileResource.clear();
} else {
if (!newFileNodes.isEmpty()) {
- IntervalFileNode first = newFileNodes.get(0);
- for (String deviceId : emptyIntervalFileNode.getStartTimeMap().keySet()) {
- first.setStartTime(deviceId, emptyIntervalFileNode.getStartTime(deviceId));
- first.setEndTime(deviceId, emptyIntervalFileNode.getEndTime(deviceId));
+ TsFileResource first = newFileNodes.get(0);
+ for (String deviceId : emptyTsFileResource.getStartTimeMap().keySet()) {
+ first.setStartTime(deviceId, emptyTsFileResource.getStartTime(deviceId));
+ first.setEndTime(deviceId, emptyTsFileResource.getEndTime(deviceId));
first.setOverflowChangeType(OverflowChangeType.CHANGED);
}
- emptyIntervalFileNode.clear();
+ emptyTsFileResource.clear();
} else {
- emptyIntervalFileNode.setOverflowChangeType(OverflowChangeType.CHANGED);
+ emptyTsFileResource.setOverflowChangeType(OverflowChangeType.CHANGED);
}
}
}
- private List<IntervalFileNode> switchFileNodeToMerge() throws FileNodeProcessorException {
- List<IntervalFileNode> result = new ArrayList<>();
- if (emptyIntervalFileNode.getOverflowChangeType() != OverflowChangeType.NO_CHANGE) {
+ private List<TsFileResource> switchFileNodeToMerge() throws FileNodeProcessorException {
+ List<TsFileResource> result = new ArrayList<>();
+ if (emptyTsFileResource.getOverflowChangeType() != OverflowChangeType.NO_CHANGE) {
// add empty
- result.add(emptyIntervalFileNode.backUp());
+ result.add(emptyTsFileResource.backUp());
if (!newFileNodes.isEmpty()) {
throw new FileNodeProcessorException(
String.format("The status of empty file is %s, but the new file list is not empty",
- emptyIntervalFileNode.getOverflowChangeType()));
+ emptyTsFileResource.getOverflowChangeType()));
}
return result;
}
@@ -1181,43 +1177,43 @@ public class FileNodeProcessor extends Processor implements IStatistic {
throw new FileNodeProcessorException(
"No file was changed when merging, the filenode is " + getProcessorName());
}
- for (IntervalFileNode intervalFileNode : newFileNodes) {
- updateFileNode(intervalFileNode, result);
+ for (TsFileResource tsFileResource : newFileNodes) {
+ updateFileNode(tsFileResource, result);
}
return result;
}
- private void updateFileNode(IntervalFileNode intervalFileNode, List<IntervalFileNode> result) {
- if (intervalFileNode.getOverflowChangeType() == OverflowChangeType.NO_CHANGE) {
- result.add(intervalFileNode.backUp());
+ private void updateFileNode(TsFileResource tsFileResource, List<TsFileResource> result) {
+ if (tsFileResource.getOverflowChangeType() == OverflowChangeType.NO_CHANGE) {
+ result.add(tsFileResource.backUp());
} else {
Map<String, Long> startTimeMap = new HashMap<>();
Map<String, Long> endTimeMap = new HashMap<>();
- for (String deviceId : intervalFileNode.getEndTimeMap().keySet()) {
- List<IntervalFileNode> temp = invertedIndexOfFiles.get(deviceId);
- int index = temp.indexOf(intervalFileNode);
+ for (String deviceId : tsFileResource.getEndTimeMap().keySet()) {
+ List<TsFileResource> temp = invertedIndexOfFiles.get(deviceId);
+ int index = temp.indexOf(tsFileResource);
int size = temp.size();
// start time
if (index == 0) {
startTimeMap.put(deviceId, 0L);
} else {
- startTimeMap.put(deviceId, intervalFileNode.getStartTime(deviceId));
+ startTimeMap.put(deviceId, tsFileResource.getStartTime(deviceId));
}
// end time
if (index < size - 1) {
endTimeMap.put(deviceId, temp.get(index + 1).getStartTime(deviceId) - 1);
} else {
- endTimeMap.put(deviceId, intervalFileNode.getEndTime(deviceId));
+ endTimeMap.put(deviceId, tsFileResource.getEndTime(deviceId));
}
}
- IntervalFileNode node = new IntervalFileNode(startTimeMap, endTimeMap,
- intervalFileNode.getOverflowChangeType(), intervalFileNode.getBaseDirIndex(),
- intervalFileNode.getRelativePath());
+ TsFileResource node = new TsFileResource(startTimeMap, endTimeMap,
+ tsFileResource.getOverflowChangeType(), tsFileResource.getBaseDirIndex(),
+ tsFileResource.getRelativePath());
result.add(node);
}
}
- private void switchMergeToWaiting(List<IntervalFileNode> backupIntervalFiles, boolean needEmpty)
+ private void switchMergeToWaiting(List<TsFileResource> backupIntervalFiles, boolean needEmpty)
throws FileNodeProcessorException {
LOGGER.info("The status of filenode processor {} switches from {} to {}.", getProcessorName(),
FileNodeProcessorStatus.MERGING_WRITE, FileNodeProcessorStatus.WAITING);
@@ -1227,10 +1223,10 @@ public class FileNodeProcessor extends Processor implements IStatistic {
oldMultiPassLock = newMultiPassLock;
newMultiPassTokenSet = new HashSet<>();
newMultiPassLock = new ReentrantReadWriteLock(false);
- List<IntervalFileNode> result = new ArrayList<>();
+ List<TsFileResource> result = new ArrayList<>();
int beginIndex = 0;
if (needEmpty) {
- IntervalFileNode empty = backupIntervalFiles.get(0);
+ TsFileResource empty = backupIntervalFiles.get(0);
if (!empty.checkEmpty()) {
updateEmpty(empty, result);
beginIndex++;
@@ -1240,8 +1236,8 @@ public class FileNodeProcessor extends Processor implements IStatistic {
addAllFileIntoIndex(backupIntervalFiles);
// check the merge changed file
for (int i = beginIndex; i < backupIntervalFiles.size(); i++) {
- IntervalFileNode newFile = newFileNodes.get(i - beginIndex);
- IntervalFileNode temp = backupIntervalFiles.get(i);
+ TsFileResource newFile = newFileNodes.get(i - beginIndex);
+ TsFileResource temp = backupIntervalFiles.get(i);
if (newFile.getOverflowChangeType() == OverflowChangeType.MERGING_CHANGE) {
updateMergeChanged(newFile, temp);
}
@@ -1251,7 +1247,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
}
// add new file when merge
for (int i = backupIntervalFiles.size() - beginIndex; i < newFileNodes.size(); i++) {
- IntervalFileNode fileNode = newFileNodes.get(i);
+ TsFileResource fileNode = newFileNodes.get(i);
if (fileNode.isClosed()) {
result.add(fileNode.backUp());
} else {
@@ -1264,13 +1260,13 @@ public class FileNodeProcessor extends Processor implements IStatistic {
// reconstruct the index
addAllFileIntoIndex(newFileNodes);
// clear merge changed
- for (IntervalFileNode fileNode : newFileNodes) {
+ for (TsFileResource fileNode : newFileNodes) {
fileNode.clearMergeChanged();
}
synchronized (fileNodeProcessorStore) {
fileNodeProcessorStore.setFileNodeProcessorStatus(isMerging);
- fileNodeProcessorStore.setEmptyIntervalFileNode(emptyIntervalFileNode);
+ fileNodeProcessorStore.setEmptyTsFileResource(emptyTsFileResource);
fileNodeProcessorStore.setNewFileNodes(newFileNodes);
try {
writeStoreToDisk(fileNodeProcessorStore);
@@ -1289,10 +1285,10 @@ public class FileNodeProcessor extends Processor implements IStatistic {
}
}
- private void updateEmpty(IntervalFileNode empty, List<IntervalFileNode> result) {
+ private void updateEmpty(TsFileResource empty, List<TsFileResource> result) {
for (String deviceId : empty.getStartTimeMap().keySet()) {
if (invertedIndexOfFiles.containsKey(deviceId)) {
- IntervalFileNode temp = invertedIndexOfFiles.get(deviceId).get(0);
+ TsFileResource temp = invertedIndexOfFiles.get(deviceId).get(0);
if (temp.getMergeChanged().contains(deviceId)) {
empty.setOverflowChangeType(OverflowChangeType.CHANGED);
break;
@@ -1303,7 +1299,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
result.add(empty.backUp());
}
- private void updateMergeChanged(IntervalFileNode newFile, IntervalFileNode temp) {
+ private void updateMergeChanged(TsFileResource newFile, TsFileResource temp) {
for (String deviceId : newFile.getMergeChanged()) {
if (temp.getStartTimeMap().containsKey(deviceId)) {
temp.setOverflowChangeType(OverflowChangeType.CHANGED);
@@ -1359,7 +1355,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
synchronized (fileNodeProcessorStore) {
fileNodeProcessorStore.setFileNodeProcessorStatus(isMerging);
fileNodeProcessorStore.setNewFileNodes(newFileNodes);
- fileNodeProcessorStore.setEmptyIntervalFileNode(emptyIntervalFileNode);
+ fileNodeProcessorStore.setEmptyTsFileResource(emptyTsFileResource);
writeStoreToDisk(fileNodeProcessorStore);
}
} catch (IOException e) {
@@ -1400,7 +1396,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
}
private void collectBufferWriteFiles(Set<String> bufferFiles) {
- for (IntervalFileNode bufferFileNode : newFileNodes) {
+ for (TsFileResource bufferFileNode : newFileNodes) {
String bufferFilePath = bufferFileNode.getFilePath();
if (bufferFilePath != null) {
bufferFiles.add(bufferFilePath);
@@ -1423,14 +1419,14 @@ public class FileNodeProcessor extends Processor implements IStatistic {
}
private void changeFileNodes() {
- for (IntervalFileNode fileNode : newFileNodes) {
+ for (TsFileResource fileNode : newFileNodes) {
if (fileNode.getOverflowChangeType() != OverflowChangeType.NO_CHANGE) {
fileNode.setOverflowChangeType(OverflowChangeType.CHANGED);
}
}
}
- private String queryAndWriteDataForMerge(IntervalFileNode backupIntervalFile)
+ private String queryAndWriteDataForMerge(TsFileResource backupIntervalFile)
throws IOException, FileNodeProcessorException, PathErrorException {
Map<String, Long> startTimeMap = new HashMap<>();
Map<String, Long> endTimeMap = new HashMap<>();
@@ -1751,7 +1747,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
public void close() throws FileNodeProcessorException {
closeBufferWrite();
closeOverflow();
- for (IntervalFileNode fileNode : newFileNodes) {
+ for (TsFileResource fileNode : newFileNodes) {
if (fileNode.getModFile() != null) {
try {
fileNode.getModFile().close();
@@ -1773,7 +1769,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
}
closeBufferWrite();
closeOverflow();
- for (IntervalFileNode fileNode : newFileNodes) {
+ for (TsFileResource fileNode : newFileNodes) {
if (fileNode.getModFile() != null) {
try {
fileNode.getModFile().close();
@@ -1819,7 +1815,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
try {
processorStore = serializeUtil.deserialize(fileNodeRestoreFilePath)
.orElse(new FileNodeProcessorStore(false, new HashMap<>(),
- new IntervalFileNode(OverflowChangeType.NO_CHANGE, null),
+ new TsFileResource(OverflowChangeType.NO_CHANGE, null),
new ArrayList<>(), FileNodeProcessorStatus.NONE, 0));
} catch (IOException e) {
throw new FileNodeProcessorException(e);
@@ -1875,12 +1871,12 @@ public class FileNodeProcessor extends Processor implements IStatistic {
private void deleteBufferWriteFiles(String deviceId, Deletion deletion,
List<ModificationFile> updatedModFiles) throws IOException {
- if (currentIntervalFileNode != null && currentIntervalFileNode.containsDevice(deviceId)) {
- currentIntervalFileNode.getModFile().write(deletion);
- updatedModFiles.add(currentIntervalFileNode.getModFile());
+ if (currentTsFileResource != null && currentTsFileResource.containsDevice(deviceId)) {
+ currentTsFileResource.getModFile().write(deletion);
+ updatedModFiles.add(currentTsFileResource.getModFile());
}
- for (IntervalFileNode fileNode : newFileNodes) {
- if (fileNode != currentIntervalFileNode && fileNode.containsDevice(deviceId)
+ for (TsFileResource fileNode : newFileNodes) {
+ if (fileNode != currentTsFileResource && fileNode.containsDevice(deviceId)
&& fileNode.getStartTime(deviceId) <= deletion.getTimestamp()) {
fileNode.getModFile().write(deletion);
updatedModFiles.add(fileNode.getModFile());
@@ -1953,8 +1949,8 @@ public class FileNodeProcessor extends Processor implements IStatistic {
Objects.equals(lastUpdateTimeMap, that.lastUpdateTimeMap) &&
Objects.equals(flushLastUpdateTimeMap, that.flushLastUpdateTimeMap) &&
Objects.equals(invertedIndexOfFiles, that.invertedIndexOfFiles) &&
- Objects.equals(emptyIntervalFileNode, that.emptyIntervalFileNode) &&
- Objects.equals(currentIntervalFileNode, that.currentIntervalFileNode) &&
+ Objects.equals(emptyTsFileResource, that.emptyTsFileResource) &&
+ Objects.equals(currentTsFileResource, that.currentTsFileResource) &&
Objects.equals(newFileNodes, that.newFileNodes) &&
isMerging == that.isMerging &&
Objects.equals(fileNodeProcessorStore, that.fileNodeProcessorStore) &&
@@ -1978,7 +1974,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
public int hashCode() {
return Objects.hash(super.hashCode(), statStorageDeltaName, statParamsHashMap, isOverflowed,
lastUpdateTimeMap, flushLastUpdateTimeMap, invertedIndexOfFiles,
- emptyIntervalFileNode, currentIntervalFileNode, newFileNodes, isMerging,
+ emptyTsFileResource, currentTsFileResource, newFileNodes, isMerging,
numOfMergeFile, fileNodeProcessorStore, fileNodeRestoreFilePath, baseDirPath,
lastMergeTime, bufferWriteProcessor, overflowProcessor, oldMultiPassTokenSet,
newMultiPassTokenSet, oldMultiPassLock, newMultiPassLock, shouldRecovery, parameters,
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessorStore.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessorStore.java
index ed8944f..db56364 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessorStore.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessorStore.java
@@ -26,7 +26,7 @@ import java.util.Map;
/**
* FileNodeProcessorStore is used to store information about FileNodeProcessor's status.
* lastUpdateTime is changed and stored by BufferWrite flush or BufferWrite close.
- * emptyIntervalFileNode and newFileNodes are changed and stored by Overflow flush and
+ * emptyTsFileResource and newFileNodes are changed and stored by Overflow flush and
* Overflow close. fileNodeProcessorState is changed and stored by the change of FileNodeProcessor's
* status such as "work->merge merge->wait wait->work". numOfMergeFile is changed
* and stored when FileNodeProcessor's status changes from work to merge.
@@ -39,8 +39,8 @@ public class FileNodeProcessorStore implements Serializable {
private boolean isOverflowed;
private Map<String, Long> lastUpdateTimeMap;
- private IntervalFileNode emptyIntervalFileNode;
- private List<IntervalFileNode> newFileNodes;
+ private TsFileResource emptyTsFileResource;
+ private List<TsFileResource> newFileNodes;
private int numOfMergeFile;
private FileNodeProcessorStatus fileNodeProcessorStatus;
@@ -48,19 +48,19 @@ public class FileNodeProcessorStore implements Serializable {
* Constructor of FileNodeProcessorStore.
* @param isOverflowed whether this FileNode contains unmerged Overflow operations.
* @param lastUpdateTimeMap the timestamp of last data point of each device in this FileNode.
- * @param emptyIntervalFileNode a place holder when the FileNode contains no TsFile.
+ * @param emptyTsFileResource a place holder when the FileNode contains no TsFile.
* @param newFileNodes TsFiles in the FileNode.
* @param fileNodeProcessorStatus the status of the FileNode.
* @param numOfMergeFile the number of files already merged in one merge operation.
*/
public FileNodeProcessorStore(boolean isOverflowed, Map<String, Long> lastUpdateTimeMap,
- IntervalFileNode emptyIntervalFileNode,
- List<IntervalFileNode> newFileNodes,
+ TsFileResource emptyTsFileResource,
+ List<TsFileResource> newFileNodes,
FileNodeProcessorStatus fileNodeProcessorStatus,
int numOfMergeFile) {
this.isOverflowed = isOverflowed;
this.lastUpdateTimeMap = lastUpdateTimeMap;
- this.emptyIntervalFileNode = emptyIntervalFileNode;
+ this.emptyTsFileResource = emptyTsFileResource;
this.newFileNodes = newFileNodes;
this.fileNodeProcessorStatus = fileNodeProcessorStatus;
this.numOfMergeFile = numOfMergeFile;
@@ -90,19 +90,19 @@ public class FileNodeProcessorStore implements Serializable {
this.lastUpdateTimeMap = lastUpdateTimeMap;
}
- public IntervalFileNode getEmptyIntervalFileNode() {
- return emptyIntervalFileNode;
+ public TsFileResource getEmptyTsFileResource() {
+ return emptyTsFileResource;
}
- public void setEmptyIntervalFileNode(IntervalFileNode emptyIntervalFileNode) {
- this.emptyIntervalFileNode = emptyIntervalFileNode;
+ public void setEmptyTsFileResource(TsFileResource emptyTsFileResource) {
+ this.emptyTsFileResource = emptyTsFileResource;
}
- public List<IntervalFileNode> getNewFileNodes() {
+ public List<TsFileResource> getNewFileNodes() {
return newFileNodes;
}
- public void setNewFileNodes(List<IntervalFileNode> newFileNodes) {
+ public void setNewFileNodes(List<TsFileResource> newFileNodes) {
this.newFileNodes = newFileNodes;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/IntervalFileNode.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/TsFileResource.java
similarity index 92%
rename from iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/IntervalFileNode.java
rename to iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/TsFileResource.java
index 6a63b61..5f0b376 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/IntervalFileNode.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/TsFileResource.java
@@ -32,7 +32,7 @@ import org.apache.iotdb.db.engine.modification.ModificationFile;
/**
* This class is used to store one bufferwrite file status.<br>
*/
-public class IntervalFileNode implements Serializable {
+public class TsFileResource implements Serializable {
private static final long serialVersionUID = -4309683416067212549L;
@@ -45,7 +45,7 @@ public class IntervalFileNode implements Serializable {
private transient ModificationFile modFile;
- public IntervalFileNode(Map<String, Long> startTimeMap, Map<String, Long> endTimeMap,
+ public TsFileResource(Map<String, Long> startTimeMap, Map<String, Long> endTimeMap,
OverflowChangeType type, int baseDirIndex, String relativePath) {
this.overflowChangeType = type;
@@ -65,7 +65,7 @@ public class IntervalFileNode implements Serializable {
* @param type whether this file is affected by overflow and how it is affected.
* @param relativePath the path of the file relative to the FileNode.
*/
- public IntervalFileNode(OverflowChangeType type, int baseDirIndex, String relativePath) {
+ public TsFileResource(OverflowChangeType type, int baseDirIndex, String relativePath) {
this.overflowChangeType = type;
this.baseDirIndex = baseDirIndex;
@@ -78,7 +78,7 @@ public class IntervalFileNode implements Serializable {
+ relativePath + ModificationFile.FILE_SUFFIX);
}
- public IntervalFileNode(OverflowChangeType type, String baseDir, String relativePath) {
+ public TsFileResource(OverflowChangeType type, String baseDir, String relativePath) {
this.overflowChangeType = type;
this.baseDirIndex = Directories.getInstance().getTsFileFolderIndex(baseDir);
@@ -91,7 +91,7 @@ public class IntervalFileNode implements Serializable {
+ relativePath + ModificationFile.FILE_SUFFIX);
}
- public IntervalFileNode(OverflowChangeType type, String relativePath) {
+ public TsFileResource(OverflowChangeType type, String relativePath) {
this(type, 0, relativePath);
}
@@ -220,11 +220,11 @@ public class IntervalFileNode implements Serializable {
}
- public IntervalFileNode backUp() {
+ public TsFileResource backUp() {
Map<String, Long> startTimeMapCopy = new HashMap<>(this.startTimeMap);
Map<String, Long> endTimeMapCopy = new HashMap<>(this.endTimeMap);
- return new IntervalFileNode(startTimeMapCopy, endTimeMapCopy, overflowChangeType,
+ return new TsFileResource(startTimeMapCopy, endTimeMapCopy, overflowChangeType,
baseDirIndex, relativePath);
}
@@ -244,7 +244,7 @@ public class IntervalFileNode implements Serializable {
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- IntervalFileNode fileNode = (IntervalFileNode) o;
+ TsFileResource fileNode = (TsFileResource) o;
return baseDirIndex == fileNode.baseDirIndex &&
overflowChangeType == fileNode.overflowChangeType &&
Objects.equals(relativePath, fileNode.relativePath) &&
@@ -257,7 +257,7 @@ public class IntervalFileNode implements Serializable {
public String toString() {
return String.format(
- "IntervalFileNode [relativePath=%s,overflowChangeType=%s, startTimeMap=%s,"
+ "TsFileResource [relativePath=%s,overflowChangeType=%s, startTimeMap=%s,"
+ " endTimeMap=%s, mergeChanged=%s]",
relativePath, overflowChangeType, startTimeMap, endTimeMap, mergeChanged);
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowSupport.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowMemtable.java
similarity index 79%
rename from iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowSupport.java
rename to iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowMemtable.java
index 8a0a97e..5fadb54 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowSupport.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowMemtable.java
@@ -24,27 +24,28 @@ import org.apache.iotdb.db.engine.memtable.IMemTable;
import org.apache.iotdb.db.engine.memtable.PrimitiveMemTable;
import org.apache.iotdb.db.engine.querycontext.ReadOnlyMemChunk;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.statistics.LongStatistics;
+import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
import org.apache.iotdb.tsfile.read.common.BatchData;
import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
/**
* This class is used to store and query all overflow data in memory.<br>
- * This just represent someone storage group.<br>
*/
-public class OverflowSupport {
+public class OverflowMemtable {
/**
* store update and delete data
*/
- private Map<String, Map<String, OverflowSeriesImpl>> indexTrees;
+ private Map<String, Map<String, LongStatistics>> indexTrees;
/**
* store insert data
*/
private IMemTable memTable;
- public OverflowSupport() {
+ public OverflowMemtable() {
indexTrees = new HashMap<>();
memTable = new PrimitiveMemTable();
}
@@ -68,9 +69,9 @@ public class OverflowSupport {
indexTrees.put(deviceId, new HashMap<>());
}
if (!indexTrees.get(deviceId).containsKey(measurementId)) {
- indexTrees.get(deviceId).put(measurementId, new OverflowSeriesImpl(measurementId, dataType));
+ indexTrees.get(deviceId).put(measurementId, new LongStatistics());
}
- indexTrees.get(deviceId).get(measurementId).update(startTime, endTime);
+ indexTrees.get(deviceId).get(measurementId).updateStats(startTime, endTime);
}
public void delete(String deviceId, String measurementId, long timestamp, boolean isFlushing) {
@@ -87,20 +88,11 @@ public class OverflowSupport {
return memTable.query(deviceId, measurementId, dataType, props);
}
- public BatchData queryOverflowUpdateInMemory(String deviceId, String measurementId,
- TSDataType dataType) {
- if (indexTrees.containsKey(deviceId) && indexTrees.get(deviceId).containsKey(measurementId)
- && indexTrees.get(deviceId).get(measurementId).getDataType().equals(dataType)) {
- return indexTrees.get(deviceId).get(measurementId).query();
- }
- return null;
- }
-
public boolean isEmptyOfOverflowSeriesMap() {
return indexTrees.isEmpty();
}
- public Map<String, Map<String, OverflowSeriesImpl>> getOverflowSeriesMap() {
+ public Map<String, Map<String, LongStatistics>> getOverflowSeriesMap() {
return indexTrees;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
index 7b9c7bf..b4669ec 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowProcessor.java
@@ -73,8 +73,8 @@ public class OverflowProcessor extends Processor {
private OverflowResource workResource;
private OverflowResource mergeResource;
- private OverflowSupport workSupport;
- private OverflowSupport flushSupport;
+ private OverflowMemtable workSupport;
+ private OverflowMemtable flushSupport;
private volatile Future<Boolean> flushFuture = new ImmediateFuture<>(true);
private volatile boolean isMerge;
@@ -113,7 +113,7 @@ public class OverflowProcessor extends Processor {
// recover file
recovery(processorDataDir);
// memory
- workSupport = new OverflowSupport();
+ workSupport = new OverflowMemtable();
overflowFlushAction = parameters.get(FileNodeConstants.OVERFLOW_FLUSH_ACTION);
filenodeFlushAction = parameters
.get(FileNodeConstants.FILENODE_PROCESSOR_FLUSH_ACTION);
@@ -380,7 +380,7 @@ public class OverflowProcessor extends Processor {
queryFlushLock.lock();
try {
flushSupport = workSupport;
- workSupport = new OverflowSupport();
+ workSupport = new OverflowMemtable();
} finally {
queryFlushLock.unlock();
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowSeriesImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowSeriesImpl.java
deleted file mode 100644
index 8aac499..0000000
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/overflow/io/OverflowSeriesImpl.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.iotdb.db.engine.overflow.io;
-
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.apache.iotdb.tsfile.file.metadata.statistics.LongStatistics;
-import org.apache.iotdb.tsfile.file.metadata.statistics.Statistics;
-import org.apache.iotdb.tsfile.read.common.BatchData;
-
-/**
- * This class is only used to store and query overflow overflowIndex {@code IIntervalTreeOperator}
- * data in memory.
- */
-public class OverflowSeriesImpl {
-
- /**
- * The data of update and delete in memory for this time series.
- */
- private String measurementId;
- private TSDataType dataType;
- private Statistics<Long> statistics;
- private int valueCount;
-
- public OverflowSeriesImpl(String measurementId, TSDataType dataType) {
- this.measurementId = measurementId;
- this.dataType = dataType;
- statistics = new LongStatistics();
- }
-
- public void update(long startTime, long endTime) {
- statistics.updateStats(startTime, endTime);
- valueCount++;
- }
-
- public void delete(long timestamp) {
- statistics.updateStats(timestamp, timestamp);
- valueCount++;
- }
-
- public BatchData query() {
- return null;
- }
-
- public long getSize() {
- return 0;
- }
-
- public String getMeasurementId() {
- return measurementId;
- }
-
- public TSDataType getDataType() {
- return dataType;
- }
-
- public Statistics<Long> getStatistics() {
- return statistics;
- }
-
- public int getValueCount() {
- return valueCount;
- }
-}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/engine/querycontext/GlobalSortedSeriesDataSource.java b/iotdb/src/main/java/org/apache/iotdb/db/engine/querycontext/GlobalSortedSeriesDataSource.java
index c78ee57..176a30e 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/querycontext/GlobalSortedSeriesDataSource.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/querycontext/GlobalSortedSeriesDataSource.java
@@ -19,8 +19,7 @@
package org.apache.iotdb.db.engine.querycontext;
import java.util.List;
-import org.apache.iotdb.db.engine.filenode.IntervalFileNode;
-import org.apache.iotdb.db.engine.memtable.TimeValuePairSorter;
+import org.apache.iotdb.db.engine.filenode.TsFileResource;
import org.apache.iotdb.tsfile.read.common.Path;
public class GlobalSortedSeriesDataSource {
@@ -28,7 +27,7 @@ public class GlobalSortedSeriesDataSource {
private Path seriesPath;
// sealed tsfile
- private List<IntervalFileNode> sealedTsFiles;
+ private List<TsFileResource> sealedTsFiles;
// unsealed tsfile
private UnsealedTsFile unsealedTsFile;
@@ -36,7 +35,7 @@ public class GlobalSortedSeriesDataSource {
// seq mem-table
private ReadOnlyMemChunk readableChunk;
- public GlobalSortedSeriesDataSource(Path seriesPath, List<IntervalFileNode> sealedTsFiles,
+ public GlobalSortedSeriesDataSource(Path seriesPath, List<TsFileResource> sealedTsFiles,
UnsealedTsFile unsealedTsFile,
ReadOnlyMemChunk readableChunk) {
this.seriesPath = seriesPath;
@@ -50,11 +49,11 @@ public class GlobalSortedSeriesDataSource {
return sealedTsFiles != null && !sealedTsFiles.isEmpty();
}
- public List<IntervalFileNode> getSealedTsFiles() {
+ public List<TsFileResource> getSealedTsFiles() {
return sealedTsFiles;
}
- public void setSealedTsFiles(List<IntervalFileNode> sealedTsFiles) {
+ public void setSealedTsFiles(List<TsFileResource> sealedTsFiles) {
this.sealedTsFiles = sealedTsFiles;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerServiceImpl.java b/iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerServiceImpl.java
index adac27c..7bbdb16 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerServiceImpl.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/postback/receiver/ServerServiceImpl.java
@@ -39,7 +39,7 @@ import org.apache.iotdb.db.conf.IoTDBConstant;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.conf.directories.Directories;
import org.apache.iotdb.db.engine.filenode.FileNodeManager;
-import org.apache.iotdb.db.engine.filenode.IntervalFileNode;
+import org.apache.iotdb.db.engine.filenode.TsFileResource;
import org.apache.iotdb.db.engine.filenode.OverflowChangeType;
import org.apache.iotdb.db.exception.FileNodeManagerException;
import org.apache.iotdb.db.utils.PostbackUtils;
@@ -660,7 +660,7 @@ public class ServerServiceImpl implements ServerService.Iface {
// create a new fileNode
String header = postbackPath + uuid.get() + File.separator + "data" + File.separator;
String relativePath = path.substring(header.length());
- IntervalFileNode fileNode = new IntervalFileNode(startTimeMap, endTimeMap,
+ TsFileResource fileNode = new TsFileResource(startTimeMap, endTimeMap,
OverflowChangeType.NO_CHANGE,
Directories.getInstance().getNextFolderIndexForTsFile(), relativePath);
// call interface of load external file
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/query/control/OpenedFilePathsManager.java b/iotdb/src/main/java/org/apache/iotdb/db/query/control/OpenedFilePathsManager.java
index b2c1452..585d85d 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/control/OpenedFilePathsManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/control/OpenedFilePathsManager.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.query.control;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
-import org.apache.iotdb.db.engine.filenode.IntervalFileNode;
+import org.apache.iotdb.db.engine.filenode.TsFileResource;
import org.apache.iotdb.db.engine.querycontext.OverflowInsertFile;
import org.apache.iotdb.db.engine.querycontext.QueryDataSource;
@@ -66,8 +66,8 @@ public class OpenedFilePathsManager {
* Add the unique file paths to closedFilePathsMap and unclosedFilePathsMap.
*/
public void addUsedFilesForCurrentRequestThread(long jobId, QueryDataSource dataSource) {
- for (IntervalFileNode intervalFileNode : dataSource.getSeqDataSource().getSealedTsFiles()) {
- String sealedFilePath = intervalFileNode.getFilePath();
+ for (TsFileResource tsFileResource : dataSource.getSeqDataSource().getSealedTsFiles()) {
+ String sealedFilePath = tsFileResource.getFilePath();
addFilePathToMap(jobId, sealedFilePath, true);
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/query/factory/SeriesReaderFactory.java b/iotdb/src/main/java/org/apache/iotdb/db/query/factory/SeriesReaderFactory.java
index 51ce94b..0e14b4b 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/factory/SeriesReaderFactory.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/factory/SeriesReaderFactory.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.db.query.factory;
import java.io.IOException;
import java.util.List;
-import org.apache.iotdb.db.engine.filenode.IntervalFileNode;
+import org.apache.iotdb.db.engine.filenode.TsFileResource;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.engine.querycontext.OverflowInsertFile;
import org.apache.iotdb.db.engine.querycontext.OverflowSeriesDataSource;
@@ -131,7 +131,7 @@ public class SeriesReaderFactory {
* This method is used to construct reader for merge process in IoTDB. To merge only one TsFile
* data and one UnSeqFile data.
*/
- public IReader createSeriesReaderForMerge(IntervalFileNode intervalFileNode,
+ public IReader createSeriesReaderForMerge(TsFileResource tsFileResource,
OverflowSeriesDataSource overflowSeriesDataSource,
SingleSeriesExpression singleSeriesExpression,
QueryContext context)
@@ -139,12 +139,12 @@ public class SeriesReaderFactory {
logger.debug("Create seriesReaders for merge. SeriesFilter = {}. TsFilePath = {}",
singleSeriesExpression,
- intervalFileNode.getFilePath());
+ tsFileResource.getFilePath());
PriorityMergeReader priorityMergeReader = new PriorityMergeReader();
// Sequence reader
- IReader seriesInTsFileReader = createSealedTsFileReaderForMerge(intervalFileNode,
+ IReader seriesInTsFileReader = createSealedTsFileReaderForMerge(tsFileResource,
singleSeriesExpression, context);
priorityMergeReader.addReaderWithPriority(seriesInTsFileReader, 1);
@@ -156,7 +156,7 @@ public class SeriesReaderFactory {
return priorityMergeReader;
}
- private IReader createSealedTsFileReaderForMerge(IntervalFileNode fileNode,
+ private IReader createSealedTsFileReaderForMerge(TsFileResource fileNode,
SingleSeriesExpression singleSeriesExpression,
QueryContext context)
throws IOException {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/SealedTsFilesReader.java b/iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/SealedTsFilesReader.java
index a8465bf..d83af0f 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/SealedTsFilesReader.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/query/reader/sequence/SealedTsFilesReader.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.db.query.reader.sequence;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
-import org.apache.iotdb.db.engine.filenode.IntervalFileNode;
+import org.apache.iotdb.db.engine.filenode.TsFileResource;
import org.apache.iotdb.db.engine.modification.Modification;
import org.apache.iotdb.db.query.context.QueryContext;
import org.apache.iotdb.db.query.control.FileReaderManager;
@@ -45,7 +45,7 @@ import org.apache.iotdb.tsfile.read.reader.series.FileSeriesReaderWithoutFilter;
public class SealedTsFilesReader implements IReader {
private Path seriesPath;
- private List<IntervalFileNode> sealedTsFiles;
+ private List<TsFileResource> sealedTsFiles;
private int usedIntervalFileIndex;
private FileSeriesReader seriesReader;
private Filter filter;
@@ -53,7 +53,7 @@ public class SealedTsFilesReader implements IReader {
private boolean hasCachedData;
private QueryContext context;
- public SealedTsFilesReader(Path seriesPath, List<IntervalFileNode> sealedTsFiles, Filter filter,
+ public SealedTsFilesReader(Path seriesPath, List<TsFileResource> sealedTsFiles, Filter filter,
QueryContext context) {
this(seriesPath, sealedTsFiles, context);
this.filter = filter;
@@ -63,7 +63,7 @@ public class SealedTsFilesReader implements IReader {
/**
* init with seriesPath and sealedTsFiles.
*/
- public SealedTsFilesReader(Path seriesPath, List<IntervalFileNode> sealedTsFiles,
+ public SealedTsFilesReader(Path seriesPath, List<TsFileResource> sealedTsFiles,
QueryContext context) {
this.seriesPath = seriesPath;
this.sealedTsFiles = sealedTsFiles;
@@ -109,7 +109,7 @@ public class SealedTsFilesReader implements IReader {
while (!flag && usedIntervalFileIndex < sealedTsFiles.size()) {
// init until reach a satisfied reader
if (seriesReader == null || !seriesReader.hasNextBatch()) {
- IntervalFileNode fileNode = sealedTsFiles.get(usedIntervalFileIndex++);
+ TsFileResource fileNode = sealedTsFiles.get(usedIntervalFileIndex++);
if (singleTsFileSatisfied(fileNode)) {
initSingleTsFileReader(fileNode, context);
} else {
@@ -155,7 +155,7 @@ public class SealedTsFilesReader implements IReader {
}
}
- private boolean singleTsFileSatisfied(IntervalFileNode fileNode) {
+ private boolean singleTsFileSatisfied(TsFileResource fileNode) {
if (filter == null) {
return true;
@@ -166,7 +166,7 @@ public class SealedTsFilesReader implements IReader {
return filter.satisfyStartEndTime(startTime, endTime);
}
- private void initSingleTsFileReader(IntervalFileNode fileNode, QueryContext context)
+ private void initSingleTsFileReader(TsFileResource fileNode, QueryContext context)
throws IOException {
// to avoid too many opened files
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/filenode/SerializeUtilTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/filenode/SerializeUtilTest.java
index 5411b46..c750439 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/filenode/SerializeUtilTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/filenode/SerializeUtilTest.java
@@ -83,13 +83,13 @@ public class SerializeUtilTest {
@Test
public void testFileStore() {
- IntervalFileNode emptyIntervalFileNode = new IntervalFileNode(OverflowChangeType.NO_CHANGE,
+ TsFileResource emptyTsFileResource = new TsFileResource(OverflowChangeType.NO_CHANGE,
null);
- List<IntervalFileNode> newFilenodes = new ArrayList<>();
+ List<TsFileResource> newFilenodes = new ArrayList<>();
String deviceId = "d0.s0";
for (int i = 1; i <= 3; i++) {
// i * 100, i * 100 + 99
- IntervalFileNode node = new IntervalFileNode(OverflowChangeType.NO_CHANGE,
+ TsFileResource node = new TsFileResource(OverflowChangeType.NO_CHANGE,
"bufferfiletest" + i);
node.setStartTime(deviceId, i * 100);
node.setEndTime(deviceId, i * 100 + 99);
@@ -100,7 +100,7 @@ public class SerializeUtilTest {
lastUpdateTimeMap.put(deviceId, (long) 500);
FileNodeProcessorStore fileNodeProcessorStore = new FileNodeProcessorStore(false,
lastUpdateTimeMap,
- emptyIntervalFileNode, newFilenodes, fileNodeProcessorState, 0);
+ emptyTsFileResource, newFilenodes, fileNodeProcessorState, 0);
SerializeUtil<FileNodeProcessorStore> serializeUtil = new SerializeUtil<>();
@@ -114,13 +114,13 @@ public class SerializeUtilTest {
try {
FileNodeProcessorStore fileNodeProcessorStore2 = serializeUtil.deserialize(filePath)
.orElse(new FileNodeProcessorStore(false, new HashMap<>(),
- new IntervalFileNode(OverflowChangeType.NO_CHANGE, null),
- new ArrayList<IntervalFileNode>(),
+ new TsFileResource(OverflowChangeType.NO_CHANGE, null),
+ new ArrayList<TsFileResource>(),
FileNodeProcessorStatus.NONE, 0));
assertEquals(fileNodeProcessorStore.getLastUpdateTimeMap(),
fileNodeProcessorStore2.getLastUpdateTimeMap());
- assertEquals(fileNodeProcessorStore.getEmptyIntervalFileNode(),
- fileNodeProcessorStore2.getEmptyIntervalFileNode());
+ assertEquals(fileNodeProcessorStore.getEmptyTsFileResource(),
+ fileNodeProcessorStore2.getEmptyTsFileResource());
assertEquals(fileNodeProcessorStore.getNewFileNodes(),
fileNodeProcessorStore2.getNewFileNodes());
assertEquals(fileNodeProcessorStore.getNumOfMergeFile(),
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowSupportTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowMemtableTest.java
similarity index 97%
rename from iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowSupportTest.java
rename to iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowMemtableTest.java
index 3cc74db..7a4ea05 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowSupportTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowMemtableTest.java
@@ -28,9 +28,9 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-public class OverflowSupportTest {
+public class OverflowMemtableTest {
- private OverflowSupport support = new OverflowSupport();
+ private OverflowMemtable support = new OverflowMemtable();
private String deviceId1 = "d1";
private String deviceId2 = "d2";
private String measurementId1 = "s1";
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowResourceTest.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowResourceTest.java
index 818dd9a..13eee7b 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowResourceTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowResourceTest.java
@@ -44,7 +44,7 @@ public class OverflowResourceTest {
private String positionFileName = "positionFile";
private String filePath = "overflow";
private String dataPath = "1";
- private OverflowSupport support = new OverflowSupport();
+ private OverflowMemtable support = new OverflowMemtable();
@Before
public void setUp() throws Exception {
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowTestUtils.java b/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowTestUtils.java
index 4b386ca..5b40870 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowTestUtils.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/engine/overflow/io/OverflowTestUtils.java
@@ -47,7 +47,7 @@ public class OverflowTestUtils {
return fileSchema;
}
- public static void produceInsertData(OverflowSupport support) {
+ public static void produceInsertData(OverflowMemtable support) {
support.insert(getData(deviceId1, measurementId1, dataType1, String.valueOf(1), 1));
support.insert(getData(deviceId1, measurementId1, dataType1, String.valueOf(3), 3));
support.insert(getData(deviceId1, measurementId1, dataType1, String.valueOf(2), 2));