You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2019/06/20 08:24:50 UTC
[incubator-iotdb] branch feature_async_close_tsfile updated: add
javadocs
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch feature_async_close_tsfile
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/feature_async_close_tsfile by this push:
new 8a0c12f add javadocs
8a0c12f is described below
commit 8a0c12f1fc74b56c134d57c88e5fae587992553a
Author: 江天 <jt...@163.com>
AuthorDate: Thu Jun 20 16:22:45 2019 +0800
add javadocs
---
.../db/engine/filenode/FileNodeProcessor.java | 2 +-
.../apache/iotdb/db/writelog/io/ILogReader.java | 18 +++++
.../apache/iotdb/db/writelog/io/ILogWriter.java | 20 ++++++
.../org/apache/iotdb/db/writelog/io/LogWriter.java | 24 ++++---
.../iotdb/db/writelog/io/MultiFileLogReader.java | 4 ++
.../iotdb/db/writelog/io/SingleFileLogReader.java | 19 ++++-
.../writelog/manager/MultiFileLogNodeManager.java | 11 ++-
.../db/writelog/manager/WriteLogNodeManager.java | 11 ++-
.../db/writelog/node/ExclusiveWriteLogNode.java | 83 +++++++---------------
.../iotdb/db/writelog/node/WriteLogNode.java | 30 +++++---
.../iotdb/db/writelog/recover/LogReplayer.java | 26 ++++---
.../recover/SeqTsFileRecoverPerformer.java | 17 ++++-
.../recover/UnseqTsFileRecoverPerformer.java | 12 ++++
.../org/apache/iotdb/db/writelog/RecoverTest.java | 4 +-
14 files changed, 177 insertions(+), 104 deletions(-)
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 a602678..e113827 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
@@ -473,7 +473,7 @@ public class FileNodeProcessor extends Processor implements IStatistic {
String filePath = tsFile.getFilePath();
String logNodePrefix = BufferWriteProcessor.logNodePrefix(processorName);
SeqTsFileRecoverPerformer recoverPerformer =
- new SeqTsFileRecoverPerformer(filePath, logNodePrefix,
+ new SeqTsFileRecoverPerformer(logNodePrefix,
fileSchema, versionController, tsFile);
recoverPerformer.recover();
} catch (ProcessorException e) {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/ILogReader.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/ILogReader.java
index 8cd060b..9d25fc7 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/ILogReader.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/ILogReader.java
@@ -21,11 +21,29 @@ package org.apache.iotdb.db.writelog.io;
import java.io.IOException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+/**
+ * ILogReader is an iterator of PhysicalPlan but throws IOException and provide a close() method.
+ * This class can be used just like java iterators, but remember to close it after use.
+ */
public interface ILogReader {
+ /**
+ * release resources occupied by this object, like file streams.
+ */
void close();
+ /**
+ * return whether there exists next log to be read.
+ * @return whether there exists next log to be read.
+ * @throws IOException
+ */
boolean hasNext() throws IOException;
+ /**
+ * return the next log read from media like a WAL file and covert it to a PhysicalPlan.
+ * @return the next log as a PhysicalPlan
+ * @throws IOException
+ * @throws java.util.NoSuchElementException when there are no more logs
+ */
PhysicalPlan next() throws IOException;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/ILogWriter.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/ILogWriter.java
index 080b41f..535923a 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/ILogWriter.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/ILogWriter.java
@@ -21,11 +21,31 @@ package org.apache.iotdb.db.writelog.io;
import java.io.IOException;
import java.util.List;
+/**
+ * ILogWriter provides functions to write WAL logs that have already been converted to bytes to a
+ * persistent medium.
+ */
public interface ILogWriter {
+ /**
+ * Write given logs to a persistent medium.
+ * NOTICE: the logs may be cached in the OS/FileSystem, if the OS/FileSystem you are using do
+ * not guarantee strong persistency and you want the logs to be persisted immediately, please
+ * call force() after calling this method.
+ * @param logCache WAL logs that have been converted to bytes, each element in this list
+ * represents one log.
+ * @throws IOException
+ */
void write(List<byte[]> logCache) throws IOException;
+ /**
+ * force the OS/FileSystem to flush its cache to make sure logs are persisted.
+ * @throws IOException
+ */
void force() throws IOException;
+ /**
+ * release resources occupied by this object, like file streams.
+ */
void close() throws IOException;
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/LogWriter.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/LogWriter.java
index e2e21e1..5abc925 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/LogWriter.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/LogWriter.java
@@ -28,11 +28,15 @@ import java.util.zip.CRC32;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+/**
+ * LogWriter writes the binarized logs into a file using FileChannel together with check sums of
+ * each log calculated using CRC32.
+ */
public class LogWriter implements ILogWriter {
private File logFile;
private FileOutputStream fileOutputStream;
- private FileChannel outputStream;
+ private FileChannel channel;
private CRC32 checkSummer = new CRC32();
private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
@@ -46,9 +50,9 @@ public class LogWriter implements ILogWriter {
@Override
public void write(List<byte[]> logCache) throws IOException {
- if (outputStream == null) {
+ if (channel == null) {
fileOutputStream = new FileOutputStream(logFile, true);
- outputStream = fileOutputStream.getChannel();
+ channel = fileOutputStream.getChannel();
}
int totalSize = 0;
for (byte[] bytes : logCache) {
@@ -63,26 +67,26 @@ public class LogWriter implements ILogWriter {
buffer.put(bytes);
}
buffer.flip();
- outputStream.write(buffer);
+ channel.write(buffer);
if (config.getForceWalPeriodInMs() == 0) {
- outputStream.force(true);
+ channel.force(true);
}
}
@Override
public void force() throws IOException {
- if (outputStream != null) {
- outputStream.force(true);
+ if (channel != null) {
+ channel.force(true);
}
}
@Override
public void close() throws IOException {
- if (outputStream != null) {
+ if (channel != null) {
fileOutputStream.close();
fileOutputStream = null;
- outputStream.close();
- outputStream = null;
+ channel.close();
+ channel = null;
}
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/MultiFileLogReader.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/MultiFileLogReader.java
index 2c25117..62dc3b5 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/MultiFileLogReader.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/MultiFileLogReader.java
@@ -24,6 +24,10 @@ import java.io.IOException;
import java.util.NoSuchElementException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+/**
+ * MultiFileLogReader constructs SingleFileLogReaders for a list of WAL files, and retrieve logs
+ * from the files one-by-one.
+ */
public class MultiFileLogReader implements ILogReader {
private SingleFileLogReader currentReader;
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/SingleFileLogReader.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/SingleFileLogReader.java
index bd854e3..445c573 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/SingleFileLogReader.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/io/SingleFileLogReader.java
@@ -26,22 +26,32 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.NoSuchElementException;
import java.util.zip.CRC32;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.qp.physical.transfer.PhysicalPlanLogTransfer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * SingleFileLogReader reads binarized WAL logs from a file through a DataInputStream by scanning
+ * the file from head to tail.
+ */
public class SingleFileLogReader implements ILogReader {
private static final Logger logger = LoggerFactory.getLogger(SingleFileLogReader.class);
+ private int bufferSize = IoTDBDescriptor.getInstance().getConfig().getMaxLogEntrySize();
public static final int LEAST_LOG_SIZE = 12; // size + checksum
+
private DataInputStream logStream;
private String filepath;
- private int bufferSize = 4 * 1024 * 1024;
+
private byte[] buffer = new byte[bufferSize];
private CRC32 checkSummer = new CRC32();
private PhysicalPlan planBuffer = null;
+ // used to indicate the position of the broken log
+ private int idx;
+
public SingleFileLogReader(File logFile) throws FileNotFoundException {
open(logFile);
}
@@ -66,7 +76,8 @@ public class SingleFileLogReader implements ILogReader {
checkSummer.reset();
checkSummer.update(buffer, 0, logSize);
if (checkSummer.getValue() != checkSum) {
- throw new IOException("The check sum is incorrect!");
+ throw new IOException(String.format("The check sum of the No.%d log is incorrect! In file: "
+ + "%d Calculated: %d.", idx, checkSum, checkSummer.getValue()));
}
planBuffer = PhysicalPlanLogTransfer.logToPlan(buffer);
return true;
@@ -80,6 +91,7 @@ public class SingleFileLogReader implements ILogReader {
PhysicalPlan ret = planBuffer;
planBuffer = null;
+ idx ++;
return ret;
}
@@ -89,7 +101,7 @@ public class SingleFileLogReader implements ILogReader {
try {
logStream.close();
} catch (IOException e) {
- logger.error("Cannot setCloseMark log file {}", filepath, e);
+ logger.error("Cannot close log file {}", filepath, e);
}
}
}
@@ -97,5 +109,6 @@ public class SingleFileLogReader implements ILogReader {
public void open(File logFile) throws FileNotFoundException {
logStream = new DataInputStream(new BufferedInputStream(new FileInputStream(logFile)));
this.filepath = logFile.getPath();
+ idx = 0;
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
index 7d3d213..313c6fb 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
@@ -32,6 +32,10 @@ import org.apache.iotdb.db.writelog.node.WriteLogNode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * MultiFileLogNodeManager manages all ExclusiveWriteLogNodes, each manages WALs of a TsFile
+ * (either seq or unseq).
+ */
public class MultiFileLogNodeManager implements WriteLogNodeManager, IService {
private static final Logger logger = LoggerFactory.getLogger(MultiFileLogNodeManager.class);
@@ -46,7 +50,10 @@ public class MultiFileLogNodeManager implements WriteLogNodeManager, IService {
logger.info("WAL force thread exits.");
return;
}
- logger.debug("Timed force starts, {} nodes to be flushed", nodeMap.size());
+ if (logger.isDebugEnabled()) {
+ logger.debug("Timed force starts, {} nodes to be flushed", nodeMap.size());
+ }
+
for (WriteLogNode node : nodeMap.values()) {
try {
node.forceSync();
@@ -115,7 +122,7 @@ public class MultiFileLogNodeManager implements WriteLogNodeManager, IService {
try {
node.close();
} catch (IOException e) {
- logger.error("{} failed to setCloseMark", node.toString(), e);
+ logger.error("failed to close {}", node, e);
}
}
nodeMap.clear();
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/manager/WriteLogNodeManager.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/manager/WriteLogNodeManager.java
index 825cf39..e61cf73 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/manager/WriteLogNodeManager.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/manager/WriteLogNodeManager.java
@@ -27,17 +27,16 @@ import org.apache.iotdb.db.writelog.node.WriteLogNode;
public interface WriteLogNodeManager {
/**
- * Get a WriteLogNode by a identifier like "{storageGroupName}-bufferwrite/overflow". The
- * WriteLogNode will be automatically created if not exist and restoreFilePath and
- * processorStoreFilePath are provided, if either restoreFilePath or processorStoreFilePath is not
- * provided and the LogNode does not exist, null is returned.
+ * Get a WriteLogNode by a identifier like "{storageGroupName}-{BufferWrite/Overflow}-{
+ * nameOfTsFile}". The WriteLogNode will be automatically created if not exist.
*
- * @param identifier -identifier, the format: "{storageGroupName}-bufferwrite/overflow"
+ * @param identifier -identifier, the format: "{storageGroupName}-{BufferWrite/Overflow}-{
+ * nameOfTsFile}"
*/
WriteLogNode getNode(String identifier) throws IOException;
/**
- * Delete a log node. If the node log does not exist, this will be an empty operation.
+ * Delete a log node. If the log node does not exist, this will be an empty operation.
*
* @param identifier -identifier
*/
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
index 82e8017..7473557 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
@@ -34,16 +34,13 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * This WriteLogNode is used to manage write ahead logs of BufferWrite or Overflow of a single
- * StorageGroup.
+ * This WriteLogNode is used to manage write ahead logs of a TsFile.
*/
public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<ExclusiveWriteLogNode> {
public static final String WAL_FILE_NAME = "wal";
private static final Logger logger = LoggerFactory.getLogger(ExclusiveWriteLogNode.class);
- /**
- * This should be the same as the corresponding StorageGroup's name.
- */
+
private String identifier;
private String logDirectory;
@@ -56,8 +53,6 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
private ReadWriteLock lock = new ReentrantReadWriteLock();
- private ReadWriteLock forceLock = new ReentrantReadWriteLock();
-
private long fileId;
private long lastFlushedId = -1;
@@ -72,13 +67,10 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
Directories.getInstance().getWALFolder() + File.separator + this.identifier;
new File(logDirectory).mkdirs();
}
-
- /*
- * Return value is of no use in this implementation.
- */
+
@Override
public void write(PhysicalPlan plan) throws IOException {
- lockForWrite();
+ lock.writeLock().lock();
try {
long start = System.currentTimeMillis();
byte[] logBytes = PhysicalPlanLogTransfer.planToLog(plan);
@@ -92,7 +84,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
logger.info("WAL write cost {} ms", elapse);
}
} finally {
- unlockForWrite();
+ lock.writeLock().unlock();
}
}
@@ -100,8 +92,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
public void close() {
sync();
forceWal();
- lockForOther();
- lockForForceOther();
+ lock.writeLock().lock();
long start = System.currentTimeMillis();
try {
if (this.currentFileWriter != null) {
@@ -110,14 +101,14 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
}
logger.debug("Log node {} closed successfully", identifier);
} catch (IOException e) {
- logger.error("Cannot setCloseMark log node {} because:", identifier, e);
+ logger.error("Cannot close log node {} because:", identifier, e);
+ } finally {
+ lock.writeLock().unlock();
}
long elapse = System.currentTimeMillis() - start;
if (elapse > 1000) {
- logger.info("WAL setCloseMark cost {} ms", elapse);
+ logger.info("WAL log node {} close cost {} ms", identifier, elapse);
}
- unlockForForceOther();
- unlockForOther();
}
@Override
@@ -129,7 +120,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
@Override
public void notifyStartFlush() {
- lockForWrite();
+ lock.writeLock().lock();
try {
long start = System.currentTimeMillis();
close();
@@ -139,13 +130,13 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
logger.info("WAL notifyStartFlush cost {} ms", elapse);
}
} finally {
- unlockForWrite();
+ lock.writeLock().unlock();
}
}
@Override
public void notifyEndFlush() {
- lockForWrite();
+ lock.writeLock().lock();
try {
long start = System.currentTimeMillis();
File logFile = new File(logDirectory, WAL_FILE_NAME + ++lastFlushedId);
@@ -155,7 +146,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
logger.info("WAL notifyEndFlush cost {} ms", elapse);
}
} finally {
- unlockForWrite();
+ lock.writeLock().unlock();
}
}
@@ -171,7 +162,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
@Override
public void delete() throws IOException {
- lockForOther();
+ lock.writeLock().lock();
try {
long start = System.currentTimeMillis();
logCache.clear();
@@ -184,7 +175,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
logger.info("WAL delete cost {} ms", elapse);
}
} finally {
- unlockForOther();
+ lock.writeLock().unlock();
}
}
@@ -193,34 +184,9 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
File[] logFiles = new File(logDirectory).listFiles();
return new MultiFileLogReader(logFiles);
}
-
- private void lockForWrite() {
- lock.writeLock().lock();
- }
-
- // other means sync and delete
- private void lockForOther() {
- lock.writeLock().lock();
- }
-
- private void unlockForWrite() {
- lock.writeLock().unlock();
- }
-
- private void unlockForOther() {
- lock.writeLock().unlock();
- }
-
- private void lockForForceOther() {
- forceLock.writeLock().lock();
- }
-
- private void unlockForForceOther() {
- forceLock.writeLock().unlock();
- }
-
+
private void sync() {
- lockForOther();
+ lock.writeLock().lock();
try {
long start = System.currentTimeMillis();
logger.debug("Log node {} starts sync, {} logs to be synced", identifier, logCache.size());
@@ -239,12 +205,12 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
logger.info("WAL sync cost {} ms", elapse);
}
} finally {
- unlockForOther();
+ lock.writeLock().unlock();
}
}
private void forceWal() {
- lockForForceOther();
+ lock.writeLock().lock();
try {
long start = System.currentTimeMillis();
logger.debug("Log node {} starts force, {} logs to be forced", identifier, logCache.size());
@@ -261,7 +227,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
logger.info("WAL forceWal cost {} ms", elapse);
}
} finally {
- unlockForForceOther();
+ lock.writeLock().unlock();
}
}
@@ -269,10 +235,11 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
if (!logFile.exists()) {
logger.info("Log file does not exist");
} else {
- if (!logFile.delete()) {
- logger.error("Old log file of {} cannot be deleted", identifier);
- } else {
+ try {
+ FileUtils.forceDelete(logFile);
logger.info("Log node {} cleaned old file", identifier);
+ } catch (IOException e) {
+ logger.error("Old log file {} of {} cannot be deleted", logFile.getName(), identifier, e);
}
}
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
index a4b2d2e..b3a135b 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/node/WriteLogNode.java
@@ -22,19 +22,22 @@ import java.io.IOException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.writelog.io.ILogReader;
+/**
+ * WriteLogNode is the minimum unit of managing WALs.
+ */
public interface WriteLogNode {
/**
- * Write a log which implements LogSerializable. First, the log will be conveyed to byte[] by
- * codec. Then the byte[] will be put into a cache. If necessary, the logs in the cache will be
+ * Write a wal for a PhysicalPlan. First, the PhysicalPlan will be conveyed to byte[].
+ * Then the byte[] will be put into a cache. When the cache is full, the logs in the cache will be
* synced to disk.
*
- * @param plan -plan
+ * @param plan - a PhysicalPlan
*/
void write(PhysicalPlan plan) throws IOException;
/**
- * Sync and setCloseMark streams.
+ * Sync and close streams.
*/
void close() throws IOException;
@@ -44,13 +47,14 @@ public interface WriteLogNode {
void forceSync() throws IOException;
/**
- * When a FileNode attempts to start a flushMetadata, this method must be called to rename log file.
+ * When data that have WALs in this node start to be flushed, this method must be called to
+ * change the working WAL file.
*/
void notifyStartFlush() throws IOException;
/**
- * When the flushMetadata of a FlieNode ends, this method must be called to check if log file needs
- * cleaning.
+ * When data that have WALs in this node end flushing, this method must be called to check and
+ * remove the out-dated logs file.
*/
void notifyEndFlush();
@@ -62,17 +66,21 @@ public interface WriteLogNode {
String getIdentifier();
/**
- * return the directory where wal file is placed.
+ * return the directory where wal files of this node are placed.
*
- * @return The directory where wal file is placed.
+ * @return The directory where wal files of this node are placed.
*/
String getLogDirectory();
/**
- * Abandon all logs in this node and delete the log directory. The caller should guarantee that NO
- * MORE WRITE is coming.
+ * Abandon all logs in this node and delete the log directory. Calling write() after calling
+ * this method is undefined.
*/
void delete() throws IOException;
+ /**
+ * return an ILogReader which can iterate each log in this log node.
+ * @return an ILogReader which can iterate each log in this log node.
+ */
ILogReader getLogReader();
}
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
index ae3ea82..822ada6 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/LogReplayer.java
@@ -40,22 +40,27 @@ import org.apache.iotdb.tsfile.write.record.TSRecord;
import org.apache.iotdb.tsfile.write.record.datapoint.DataPoint;
import org.apache.iotdb.tsfile.write.schema.FileSchema;
+/**
+ * LogReplayer finds the logNode of the TsFile given by insertFilePath and logNodePrefix, reads
+ * the WALs from the logNode and redoes them into a given MemTable and ModificationFile.
+ */
public class LogReplayer {
- private String processorName;
+ private String logNodePrefix;
private String insertFilePath;
private ModificationFile modFile;
private VersionController versionController;
private TsFileResource currentTsFileResource;
+ // fileSchema is used to get the measurement data type
private FileSchema fileSchema;
private IMemTable recoverMemTable;
- public LogReplayer(String processorName, String insertFilePath,
+ public LogReplayer(String logNodePrefix, String insertFilePath,
ModificationFile modFile,
VersionController versionController,
TsFileResource currentTsFileResource,
FileSchema fileSchema, IMemTable memTable) {
- this.processorName = processorName;
+ this.logNodePrefix = logNodePrefix;
this.insertFilePath = insertFilePath;
this.modFile = modFile;
this.versionController = versionController;
@@ -64,11 +69,16 @@ public class LogReplayer {
this.recoverMemTable = memTable;
}
+ /**
+ * finds the logNode of the TsFile given by insertFilePath and logNodePrefix, reads
+ * the WALs from the logNode and redoes them into a given MemTable and ModificationFile.
+ * @throws ProcessorException
+ */
public void replayLogs() throws ProcessorException {
WriteLogNode logNode;
try {
logNode = MultiFileLogNodeManager.getInstance().getNode(
- processorName + new File(insertFilePath).getName());
+ logNodePrefix + new File(insertFilePath).getName());
} catch (IOException e) {
throw new ProcessorException(e);
}
@@ -77,9 +87,9 @@ public class LogReplayer {
while (logReader.hasNext()) {
PhysicalPlan plan = logReader.next();
if (plan instanceof InsertPlan) {
- replayInsert((InsertPlan) plan, recoverMemTable);
+ replayInsert((InsertPlan) plan);
} else if (plan instanceof DeletePlan) {
- replayDelete((DeletePlan) plan, recoverMemTable);
+ replayDelete((DeletePlan) plan);
}
}
} catch (IOException e) {
@@ -87,7 +97,7 @@ public class LogReplayer {
}
}
- private void replayDelete(DeletePlan deletePlan, IMemTable recoverMemTable) throws IOException {
+ private void replayDelete(DeletePlan deletePlan) throws IOException {
List<Path> paths = deletePlan.getPaths();
for (Path path : paths) {
recoverMemTable.delete(path.getDevice(), path.getMeasurement(), deletePlan.getDeleteTime());
@@ -96,7 +106,7 @@ public class LogReplayer {
}
}
- private void replayInsert(InsertPlan insertPlan, IMemTable recoverMemTable) {
+ private void replayInsert(InsertPlan insertPlan) {
TSRecord tsRecord = new TSRecord(insertPlan.getTime(), insertPlan.getDeviceId());
if (currentTsFileResource != null) {
// the last chunk group may contain the same data with the logs, ignore such logs
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/SeqTsFileRecoverPerformer.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/SeqTsFileRecoverPerformer.java
index 1329c7e..ee99733 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/SeqTsFileRecoverPerformer.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/SeqTsFileRecoverPerformer.java
@@ -35,6 +35,10 @@ import org.apache.iotdb.tsfile.file.metadata.ChunkMetaData;
import org.apache.iotdb.tsfile.write.schema.FileSchema;
import org.apache.iotdb.tsfile.write.writer.NativeRestorableIOWriter;
+/**
+ * SeqTsFileRecoverPerformer recovers a SeqTsFile to correct status, redoes the WALs since last
+ * crash and removes the redone logs.
+ */
public class SeqTsFileRecoverPerformer {
private String insertFilePath;
@@ -44,16 +48,23 @@ public class SeqTsFileRecoverPerformer {
private LogReplayer logReplayer;
private TsFileResource tsFileResource;
- public SeqTsFileRecoverPerformer(String insertFilePath, String logNodePrefix,
+ public SeqTsFileRecoverPerformer(String logNodePrefix,
FileSchema fileSchema, VersionController versionController,
TsFileResource currentTsFileResource) {
- this.insertFilePath = insertFilePath;
+ this.insertFilePath = currentTsFileResource.getFilePath();
this.logNodePrefix = logNodePrefix;
this.fileSchema = fileSchema;
this.versionController = versionController;
this.tsFileResource = currentTsFileResource;
}
+ /**
+ * 1. recover the TsFile by NativeRestorableIOWriter and truncate position of last recovery
+ * 2. redo the WALs to recover unpersisted data
+ * 3. flush and close the file
+ * 4. clean WALs
+ * @throws ProcessorException
+ */
public void recover() throws ProcessorException {
IMemTable recoverMemTable = new PrimitiveMemTable();
this.logReplayer = new LogReplayer(logNodePrefix, insertFilePath, tsFileResource.getModFile(),
@@ -64,7 +75,7 @@ public class SeqTsFileRecoverPerformer {
return;
}
NativeRestorableIOWriter restorableTsFileIOWriter = recoverFile(insertFile);
- // due to failure, the last ChunkGroup may contain the same data with the WALs, so the time
+ // due to failure, the last ChunkGroup may contain the same data as the WALs, so the time
// map must be updated first to avoid duplicated insertion
for (ChunkGroupMetaData chunkGroupMetaData : restorableTsFileIOWriter.getChunkGroupMetaDatas()) {
for (ChunkMetaData chunkMetaData : chunkGroupMetaData.getChunkMetaDataList()) {
diff --git a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/UnseqTsFileRecoverPerformer.java b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/UnseqTsFileRecoverPerformer.java
index 6861265..ea577fa 100644
--- a/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/UnseqTsFileRecoverPerformer.java
+++ b/iotdb/src/main/java/org/apache/iotdb/db/writelog/recover/UnseqTsFileRecoverPerformer.java
@@ -28,6 +28,10 @@ import org.apache.iotdb.db.exception.ProcessorException;
import org.apache.iotdb.db.writelog.manager.MultiFileLogNodeManager;
import org.apache.iotdb.tsfile.write.schema.FileSchema;
+
+/**
+ * UnseqTsFileRecoverPerformer redoes the WALs since last crash and removes the redone logs.
+ */
public class UnseqTsFileRecoverPerformer {
private FileSchema fileSchema;
@@ -39,6 +43,14 @@ public class UnseqTsFileRecoverPerformer {
this.fileSchema = fileSchema;
}
+ /**
+ * The UnseqTsFile itself has already been recovered by the OverResource, and duplicated
+ * operation will not break the file, so merely redoing and flush are enough.
+ * 1. redo the WALs to recover unpersisted data
+ * 2. flush
+ * 3. clean WALs
+ * @throws ProcessorException
+ */
public void recover() throws ProcessorException {
IMemTable memTable = new PrimitiveMemTable();
String logNodePrefix = resource.logNodePrefix();
diff --git a/iotdb/src/test/java/org/apache/iotdb/db/writelog/RecoverTest.java b/iotdb/src/test/java/org/apache/iotdb/db/writelog/RecoverTest.java
index 92fdc60..5a10175 100644
--- a/iotdb/src/test/java/org/apache/iotdb/db/writelog/RecoverTest.java
+++ b/iotdb/src/test/java/org/apache/iotdb/db/writelog/RecoverTest.java
@@ -106,8 +106,8 @@ public class RecoverTest {
logNode.write(deletePlan);
logNode.forceSync();
- SeqTsFileRecoverPerformer performer = new SeqTsFileRecoverPerformer(
- insertFile.getPath(), deviceId, schema, SysTimeVersionController.INSTANCE, tsFileResource);
+ SeqTsFileRecoverPerformer performer = new SeqTsFileRecoverPerformer(deviceId, schema,
+ SysTimeVersionController.INSTANCE, tsFileResource);
// used to check if logs are replayed in order
performer.recover();