You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by qi...@apache.org on 2019/06/19 06:37:16 UTC
[incubator-iotdb] 02/02: add recover filenode processor
This is an automated email from the ASF dual-hosted git repository.
qiaojialin pushed a commit to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
commit 6e183f6db4b1d5d178b98104277245a522acb965
Merge: e8dc717 00feb5f
Author: qiaojialin <64...@qq.com>
AuthorDate: Wed Jun 19 14:36:59 2019 +0800
add recover filenode processor
.../bufferwriteV2/BufferWriteProcessorV2.java | 1 -
.../db/engine/filenode/FileNodeProcessor.java | 4 +-
.../filenodeV2/FileNodeProcessorStoreV2.java | 50 ++++++------
.../db/engine/filenodeV2/FileNodeProcessorV2.java | 93 +++++++++++++++-------
.../db/engine/filenodeV2/TsFileResourceV2.java | 3 +
5 files changed, 98 insertions(+), 53 deletions(-)
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwriteV2/BufferWriteProcessorV2.java
index ecf28ea,ecf28ea..144c66e
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwriteV2/BufferWriteProcessorV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/bufferwriteV2/BufferWriteProcessorV2.java
@@@ -34,7 -34,7 +34,6 @@@ import org.apache.iotdb.tsfile.file.met
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.utils.Pair;
import org.apache.iotdb.tsfile.write.schema.FileSchema;
--import org.apache.iotdb.tsfile.write.writer.NativeRestorableIOWriter;
public class BufferWriteProcessorV2 extends AbstractUnsealedDataFileProcessorV2 {
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/filenode/FileNodeProcessor.java
index 0d670a7,0d670a7..54191e9
--- 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
@@@ -202,9 -202,9 +202,9 @@@ public class FileNodeProcessor extends
// @Override
// public void act() {
// synchronized (fileNodeProcessorStore) {
--// fileNodeProcessorStore.setLastUpdateTimeMap(lastUpdateTimeMap);
++// fileNodeProcessorStore.setLatestTimeMap(lastUpdateTimeMap);
// addLastTimeToIntervalFile();
--// fileNodeProcessorStore.setNewFileNodes(newFileNodes);
++// fileNodeProcessorStore.setSequenceFileList(newFileNodes);
// }
// }
//
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorStoreV2.java
index 7ae9776,7ae9776..b136511
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorStoreV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorStoreV2.java
@@@ -34,7 -34,7 +34,7 @@@ import org.apache.iotdb.tsfile.utils.Re
/**
* FileNodeProcessorStore is used to store information about FileNodeProcessor's status.
* lastUpdateTime is changed and stored by BufferWrite flushMetadata or BufferWrite close.
-- * emptyTsFileResource and newFileNodes are changed and stored by Overflow flushMetadata and
++ * emptyTsFileResource and sequenceFileList are changed and stored by Overflow flushMetadata 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.
@@@ -44,8 -44,8 +44,9 @@@ public class FileNodeProcessorStoreV2 i
private static final long serialVersionUID = -54525372941897565L;
private boolean isOverflowed;
-- private Map<String, Long> lastUpdateTimeMap;
-- private List<TsFileResourceV2> newFileNodes;
++ private Map<String, Long> latestTimeMap;
++ private List<TsFileResourceV2> sequenceFileList;
++ private List<TsFileResourceV2> unSequenceFileList;
private int numOfMergeFile;
private FileNodeProcessorStatus fileNodeProcessorStatus;
@@@ -53,18 -53,18 +54,17 @@@
* 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 newFileNodes TsFiles in the FileNode.
++ * @param latestTimeMap the timestamp of last data point of each device in this FileNode.
++ * @param sequenceFileList TsFiles in the FileNode.
* @param fileNodeProcessorStatus the status of the FileNode.
* @param numOfMergeFile the number of files already merged in one merge operation.
*/
-- public FileNodeProcessorStoreV2(boolean isOverflowed, Map<String, Long> lastUpdateTimeMap,
-- List<TsFileResourceV2> newFileNodes,
-- FileNodeProcessorStatus fileNodeProcessorStatus,
++ public FileNodeProcessorStoreV2(boolean isOverflowed, Map<String, Long> latestTimeMap,
++ List<TsFileResourceV2> sequenceFileList, FileNodeProcessorStatus fileNodeProcessorStatus,
int numOfMergeFile) {
this.isOverflowed = isOverflowed;
-- this.lastUpdateTimeMap = lastUpdateTimeMap;
-- this.newFileNodes = newFileNodes;
++ this.latestTimeMap = latestTimeMap;
++ this.sequenceFileList = sequenceFileList;
this.fileNodeProcessorStatus = fileNodeProcessorStatus;
this.numOfMergeFile = numOfMergeFile;
}
@@@ -72,14 -72,14 +72,14 @@@
public void serialize(OutputStream outputStream) throws IOException {
ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
ReadWriteIOUtils.write(this.isOverflowed, byteArrayOutputStream);
-- // lastUpdateTimeMap
-- ReadWriteIOUtils.write(lastUpdateTimeMap.size(), byteArrayOutputStream);
-- for (Entry<String, Long> entry : lastUpdateTimeMap.entrySet()) {
++ // latestTimeMap
++ ReadWriteIOUtils.write(latestTimeMap.size(), byteArrayOutputStream);
++ for (Entry<String, Long> entry : latestTimeMap.entrySet()) {
ReadWriteIOUtils.write(entry.getKey(), byteArrayOutputStream);
ReadWriteIOUtils.write(entry.getValue(), byteArrayOutputStream);
}
-- ReadWriteIOUtils.write(this.newFileNodes.size(), byteArrayOutputStream);
-- for (TsFileResourceV2 tsFileResource : this.newFileNodes) {
++ ReadWriteIOUtils.write(this.sequenceFileList.size(), byteArrayOutputStream);
++ for (TsFileResourceV2 tsFileResource : this.sequenceFileList) {
tsFileResource.serialize(byteArrayOutputStream);
}
ReadWriteIOUtils.write(this.numOfMergeFile, byteArrayOutputStream);
@@@ -126,20 -126,20 +126,24 @@@
this.fileNodeProcessorStatus = fileNodeProcessorStatus;
}
-- public Map<String, Long> getLastUpdateTimeMap() {
-- return new HashMap<>(lastUpdateTimeMap);
++ public Map<String, Long> getLatestTimeMap() {
++ return new HashMap<>(latestTimeMap);
}
-- public void setLastUpdateTimeMap(Map<String, Long> lastUpdateTimeMap) {
-- this.lastUpdateTimeMap = lastUpdateTimeMap;
++ public void setLatestTimeMap(Map<String, Long> latestTimeMap) {
++ this.latestTimeMap = latestTimeMap;
}
-- public List<TsFileResourceV2> getNewFileNodes() {
-- return newFileNodes;
++ public List<TsFileResourceV2> getSequenceFileList() {
++ return sequenceFileList;
}
-- public void setNewFileNodes(List<TsFileResourceV2> newFileNodes) {
-- this.newFileNodes = newFileNodes;
++ public void setSequenceFileList(List<TsFileResourceV2> sequenceFileList) {
++ this.sequenceFileList = sequenceFileList;
++ }
++
++ public List<TsFileResourceV2> getUnSequenceFileList() {
++ return unSequenceFileList;
}
public int getNumOfMergeFile() {
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java
index e5902c1,89a12e0..f28bdfd
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/FileNodeProcessorV2.java
@@@ -22,7 -22,7 +22,6 @@@ import java.io.File
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
--import java.io.UTFDataFormatException;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.HashMap;
@@@ -43,9 -43,9 +42,12 @@@ import org.apache.iotdb.db.engine.versi
import org.apache.iotdb.db.engine.version.VersionController;
import org.apache.iotdb.db.exception.FileNodeProcessorException;
import org.apache.iotdb.db.metadata.MManager;
--import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
++import org.apache.iotdb.tsfile.file.metadata.enums.CompressionType;
++import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
++import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.schema.FileSchema;
++import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@@ -61,10 -61,10 +63,10 @@@ public class FileNodeProcessorV2
private FileSchema fileSchema;
-- /**
-- * device -> tsfile list
-- */
-- private Map<String, List<TsFileResourceV2>> invertedIndexOfFiles = new HashMap<>();
++// /**
++// * device -> tsfile list
++// */
++// private Map<String, List<TsFileResourceV2>> invertedIndexOfFiles = new HashMap<>();
// for bufferwrite
//includes sealed and unsealed tsfiles
@@@ -73,14 -73,14 +75,15 @@@
private CopyOnWriteLinkedList<BufferWriteProcessorV2> closingBufferWriteProcessor = new CopyOnWriteLinkedList<>();
// for overflow
-- private List<TsFileResourceV2> unsequenceFileList;
++ private List<TsFileResourceV2> unSequenceFileList;
private OverflowProcessorV2 workOverflowProcessor = null;
- private CopyOnWriteLinkedList<BufferWriteProcessorV2> closingOverflowProcessor = new CopyOnWriteLinkedList<>();
++
+ private CopyOnWriteLinkedList<OverflowProcessorV2> closingOverflowProcessor = new CopyOnWriteLinkedList<>();
/**
* device -> global latest timestamp of each device
*/
-- private Map<String, Long> latestTimeMap = new HashMap<>();
++ private Map<String, Long> latestTimeMap;
/**
* device -> largest timestamp of the latest memtable to be submitted to asyncFlush
@@@ -97,7 -97,7 +100,6 @@@
private FileNodeProcessorStoreV2 fileNodeProcessorStore;
private final Object fileNodeRestoreLock = new Object();
--
public FileNodeProcessorV2(String baseDir, String storageGroup) throws FileNodeProcessorException {
this.storageGroup = storageGroup;
lock = new ReentrantReadWriteLock();
@@@ -105,8 -105,8 +107,7 @@@
File storageGroupDir = new File(baseDir + storageGroup);
if (!storageGroupDir.exists()) {
storageGroupDir.mkdir();
-- LOGGER.info(
-- "The directory of the storage group {} doesn't exist. Create a new " +
++ LOGGER.info("The directory of the storage group {} doesn't exist. Create a new " +
"directory {}", storageGroup, storageGroupDir.getAbsolutePath());
}
@@@ -119,18 -119,19 +120,21 @@@
LOGGER.info("The restore directory of the filenode processor {} doesn't exist. Create new " +
"directory {}", storageGroup, restoreFolder.getAbsolutePath());
}
-- fileNodeRestoreFilePath = new File(restoreFolder, storageGroup + RESTORE_FILE_SUFFIX)
-- .getPath();
++
++ fileNodeRestoreFilePath = new File(restoreFolder, storageGroup + RESTORE_FILE_SUFFIX).getPath();
++
try {
-- fileNodeProcessorStore = readStoreFromDisk();
++ fileNodeProcessorStore = readStoreFromDiskOrCreate();
} catch (FileNodeProcessorException e) {
- LOGGER.error("The fileNode processor {} encountered an error when recoverying restore " +
- LOGGER.error(
- "The fileNode processor {} encountered an error when recoverying restore " +
++ LOGGER.error("The fileNode processor {} encountered an error when recovering restore " +
"information.", storageGroup);
throw new FileNodeProcessorException(e);
}
-- // TODO deep clone the lastupdate time, change the getNewFileNodes to V2
--// sequenceFileList = fileNodeProcessorStore.getNewFileNodes();
-- invertedIndexOfFiles = new HashMap<>();
++
++ // TODO deep clone the lastupdate time, change the getSequenceFileList to V2
++ sequenceFileList = fileNodeProcessorStore.getSequenceFileList();
++ unSequenceFileList = fileNodeProcessorStore.getUnSequenceFileList();
++ latestTimeMap = fileNodeProcessorStore.getLatestTimeMap();
/**
* version controller
@@@ -141,10 -142,10 +145,42 @@@
throw new FileNodeProcessorException(e);
}
++ // construct the file schema
++ this.fileSchema = constructFileSchema(storageGroup);
+ }
+
++ private FileSchema constructFileSchema(String storageGroupName) {
++ List<MeasurementSchema> columnSchemaList;
++ columnSchemaList = mManager.getSchemaForFileName(storageGroupName);
++
++ FileSchema schema = new FileSchema();
++ for (MeasurementSchema measurementSchema : columnSchemaList) {
++ schema.registerMeasurement(measurementSchema);
++ }
++ return schema;
++
++ }
+
- private FileNodeProcessorStoreV2 readStoreFromDisk() throws FileNodeProcessorException {
++
++ /**
++ * add time series.
++ */
++ public void addTimeSeries(String measurementId, TSDataType dataType, TSEncoding encoding,
++ CompressionType compressor, Map<String, String> props) {
++ lock.writeLock().lock();
++ try {
++ fileSchema.registerMeasurement(new MeasurementSchema(measurementId, dataType, encoding,
++ compressor, props));
++ } finally {
++ lock.writeLock().unlock();
++ }
+ }
+
+
- private FileNodeProcessorStoreV2 readStoreFromDisk() throws FileNodeProcessorException {
++ /**
++ * read file node store from disk or create a new one
++ */
++ private FileNodeProcessorStoreV2 readStoreFromDiskOrCreate() throws FileNodeProcessorException {
synchronized (fileNodeRestoreLock) {
File restoreFile = new File(fileNodeRestoreFilePath);
@@@ -176,34 -177,8 +212,35 @@@
}
}
- private void writeUnsealedDataFile(AbstractUnsealedDataFileProcessorV2 udfProcessor, TSRecord tsRecord, boolean sequence)
- throws IOException {
++
+ public boolean insert(TSRecord tsRecord) {
+ lock.writeLock().lock();
+ boolean result = true;
+
+ try {
+ // init map
+ latestTimeMap.putIfAbsent(tsRecord.deviceId, Long.MIN_VALUE);
+ latestFlushTimeMap.putIfAbsent(tsRecord.deviceId, Long.MIN_VALUE);
+
+ // write to sequence or unsequence file
+ if (tsRecord.time > latestFlushTimeMap.get(tsRecord.deviceId)) {
- writeUnsealedDataFile(workBufferWriteProcessor, tsRecord, true);
++ result = writeUnsealedDataFile(workBufferWriteProcessor, tsRecord, true);
+ } else {
- writeUnsealedDataFile(workOverflowProcessor, tsRecord, false);
++ result = writeUnsealedDataFile(workOverflowProcessor, tsRecord, false);
+ }
+ } catch (Exception e) {
+ LOGGER.error("insert tsRecord to unsealed data file failed, because {}", e.getMessage(), e);
+ result = false;
+ } finally {
+ lock.writeLock().unlock();
+ }
+
+ return result;
+ }
+
+
- private void writeUnsealedDataFile(AbstractUnsealedDataFileProcessorV2 udfProcessor,
++ private boolean writeUnsealedDataFile(AbstractUnsealedDataFileProcessorV2 udfProcessor,
+ TSRecord tsRecord, boolean sequence) throws IOException {
boolean result;
// create a new BufferWriteProcessor
if (udfProcessor == null) {
@@@ -219,8 -193,9 +256,8 @@@
String filePath = Paths.get(baseDir, storageGroup, tsRecord.time + "").toString();
udfProcessor = new OverflowProcessorV2(storageGroup, new File(filePath),
fileSchema, versionController, this::closeBufferWriteProcessorCallBack);
-- unsequenceFileList.add(udfProcessor.getTsFileResource());
++ unSequenceFileList.add(udfProcessor.getTsFileResource());
}
- // TODO check if the disk is full
}
// write BufferWrite
@@@ -235,9 -210,32 +272,11 @@@
if (udfProcessor.shouldFlush()) {
flushAndCheckClose(udfProcessor, sequence);
}
- }
-
- public boolean insert(TSRecord tsRecord) {
- lock.writeLock().lock();
- boolean result = true;
-
- try {
-
- // init map
- latestTimeMap.putIfAbsent(tsRecord.deviceId, Long.MIN_VALUE);
- latestFlushTimeMap.putIfAbsent(tsRecord.deviceId, Long.MIN_VALUE);
-
- if (tsRecord.time > latestFlushTimeMap.get(tsRecord.deviceId)) {
- writeUnsealedDataFile(workBufferWriteProcessor, tsRecord, true);
- } else {
- writeUnsealedDataFile(workOverflowProcessor, tsRecord, false);
- }
- } catch (Exception e) {
-
- } finally {
- lock.writeLock().unlock();
- }
+
+ return result;
}
+
/**
* ensure there must be a flush thread submitted after close() is called,
* therefore the close task will be executed by a flush thread.
@@@ -281,7 -279,7 +320,7 @@@
private void closeBufferWriteProcessorCallBack(Object bufferWriteProcessor) {
closingBufferWriteProcessor.remove((BufferWriteProcessorV2) bufferWriteProcessor);
synchronized (fileNodeProcessorStore) {
-- fileNodeProcessorStore.setLastUpdateTimeMap(latestTimeMap);
++ fileNodeProcessorStore.setLatestTimeMap(latestTimeMap);
if (!sequenceFileList.isEmpty()) {
// end time with one start time
@@@ -293,7 -291,7 +332,7 @@@
}
resource.setEndTimeMap(endTimeMap);
}
-- fileNodeProcessorStore.setNewFileNodes(sequenceFileList);
++ fileNodeProcessorStore.setSequenceFileList(sequenceFileList);
try {
writeStoreToDisk(fileNodeProcessorStore);
} catch (FileNodeProcessorException e) {
diff --cc iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/TsFileResourceV2.java
index b49c00a,b49c00a..d587dc1
--- a/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/TsFileResourceV2.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/engine/filenodeV2/TsFileResourceV2.java
@@@ -71,6 -71,6 +71,9 @@@ public class TsFileResourceV2
this.endTimeMap = endTimeMap;
}
++ public Map<String, Long> getEndTimeMap() {
++ return endTimeMap;
++ }
public void serialize(OutputStream outputStream) throws IOException {
// ReadWriteIOUtils.write(this.overflowChangeType.serialize(), outputStream);