This is an automated email from the ASF dual-hosted git repository.
haonan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 03d37b0 [IOTDB-958] Fix MLog bug and add switch for MTree snapshot
(#1863)
03d37b0 is described below
commit 03d37b0bcf3a1be6f3ca9f4336dd3f2f793d0eeb
Author: Zesong Sun <[email protected]>
AuthorDate: Mon Oct 26 17:39:37 2020 +0800
[IOTDB-958] Fix MLog bug and add switch for MTree snapshot (#1863)
Co-authored-by: JackieTien97 <[email protected]>
---
.../resources/conf/iotdb-engine.properties | 7 +++-
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 13 +++++++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 2 +
.../org/apache/iotdb/db/metadata/MLogWriter.java | 45 ++++++++++------------
.../org/apache/iotdb/db/metadata/MManager.java | 20 +++++-----
5 files changed, 52 insertions(+), 35 deletions(-)
diff --git a/server/src/assembly/resources/conf/iotdb-engine.properties
b/server/src/assembly/resources/conf/iotdb-engine.properties
index 3045cce..183ba86 100644
--- a/server/src/assembly/resources/conf/iotdb-engine.properties
+++ b/server/src/assembly/resources/conf/iotdb-engine.properties
@@ -207,11 +207,16 @@ tag_attribute_total_size=700
# if enable partial insert, one measurement failure will not impact other
measurements
enable_partial_insert=true
-# The least interval line numbers of mlog.txt when creating a checkpoint and
saving snapshot of MTree. Unit: line numbers
+# Whether to enable MTree snapshot. Default false in 0.11.0
+enable_mtree_snapshot=false
+
+# The least interval line numbers of mlog.txt when creating a checkpoint and
saving snapshot of MTree.
+# Only take effect when enable_mtree_snapshot=true. Unit: line numbers
mtree_snapshot_interval=100000
# Threshold interval time of MTree modification. Unit: second. Default: 1
hour(3600 seconds)
# If the last modification time is less than this threshold, MTree snapshot
will not be created
+# Only take effect when enable_mtree_snapshot=true.
mtree_snapshot_threshold_time=3600
####################
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index 89f053b..f873ca5 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -661,6 +661,11 @@ public class IoTDBConfig {
private boolean enablePartition = false;
/**
+ * whether enable MTree snapshot
+ */
+ private boolean enableMTreeSnapshot = false;
+
+ /**
* Interval line number of mlog.txt when creating a checkpoint and saving
snapshot of mtree
*/
private int mtreeSnapshotInterval = 100000;
@@ -726,6 +731,14 @@ public class IoTDBConfig {
this.enablePartition = enablePartition;
}
+ public boolean isEnableMTreeSnapshot() {
+ return enableMTreeSnapshot;
+ }
+
+ public void setEnableMTreeSnapshot(boolean enableMTreeSnapshot) {
+ this.enableMTreeSnapshot = enableMTreeSnapshot;
+ }
+
public int getMtreeSnapshotInterval() {
return mtreeSnapshotInterval;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 4fe9ecb..7f9e9b3 100644
--- a/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++ b/server/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -381,6 +381,8 @@ public class IoTDBDescriptor {
Boolean.parseBoolean(properties.getProperty("enable_partial_insert",
String.valueOf(conf.isEnablePartialInsert()))));
+ conf.setEnableMTreeSnapshot(Boolean.parseBoolean(properties.getProperty(
+ "enable_mtree_snapshot",
Boolean.toString(conf.isEnableMTreeSnapshot()))));
conf.setMtreeSnapshotInterval(Integer.parseInt(properties.getProperty(
"mtree_snapshot_interval",
Integer.toString(conf.getMtreeSnapshotInterval()))));
conf.setMtreeSnapshotThresholdTime(Integer.parseInt(properties.getProperty(
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MLogWriter.java
b/server/src/main/java/org/apache/iotdb/db/metadata/MLogWriter.java
index d6745f0..5f77527 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MLogWriter.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MLogWriter.java
@@ -29,6 +29,7 @@ import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Files;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
import org.apache.iotdb.db.qp.physical.sys.CreateTimeSeriesPlan;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
@@ -38,11 +39,12 @@ import org.slf4j.LoggerFactory;
public class MLogWriter {
private static final Logger logger =
LoggerFactory.getLogger(MLogWriter.class);
- private static final String STRING_TYPE = "%s,%s,%s";
- private File logFile;
+ private static final String STRING_TYPE = "%s,%s,%s" +
System.lineSeparator();
+ private static final String LINE_SEPARATOR = System.lineSeparator();
+ private final File logFile;
private FileOutputStream fileOutputStream;
private FileChannel channel;
- private int lineNumber;
+ private final AtomicInteger lineNumber;
public MLogWriter(String schemaDir, String logFileName) throws IOException {
File metadataDir = SystemFileFactory.INSTANCE.getFile(schemaDir);
@@ -57,10 +59,10 @@ public class MLogWriter {
logFile = SystemFileFactory.INSTANCE.getFile(schemaDir + File.separator +
logFileName);
fileOutputStream = new FileOutputStream(logFile, true);
channel = fileOutputStream.getChannel();
+ lineNumber = new AtomicInteger(0);
}
public void close() throws IOException {
- channel.close();
fileOutputStream.close();
}
@@ -92,51 +94,50 @@ public class MLogWriter {
if (offset >= 0) {
buf.append(offset);
}
- buf.append(System.getProperty("line.separator"));
+ buf.append(LINE_SEPARATOR);
channel.write(ByteBuffer.wrap(buf.toString().getBytes()));
- ++lineNumber;
+ lineNumber.incrementAndGet();
}
public void deleteTimeseries(String path) throws IOException {
- String outputStr = MetadataOperationType.DELETE_TIMESERIES + "," + path;
+ String outputStr = MetadataOperationType.DELETE_TIMESERIES + "," + path +
LINE_SEPARATOR;
ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
channel.write(buff);
- newLine();
}
public void setStorageGroup(String storageGroup) throws IOException {
- String outputStr = MetadataOperationType.SET_STORAGE_GROUP + "," +
storageGroup;
+ String outputStr = MetadataOperationType.SET_STORAGE_GROUP + "," +
storageGroup + LINE_SEPARATOR;
ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
channel.write(buff);
- newLine();
+ lineNumber.incrementAndGet();
}
public void deleteStorageGroup(String storageGroup) throws IOException {
- String outputStr = MetadataOperationType.DELETE_STORAGE_GROUP + "," +
storageGroup;
+ String outputStr = MetadataOperationType.DELETE_STORAGE_GROUP + "," +
storageGroup + LINE_SEPARATOR;
ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
channel.write(buff);
- newLine();
+ lineNumber.incrementAndGet();
}
public void setTTL(String storageGroup, long ttl) throws IOException {
String outputStr = String.format(STRING_TYPE,
MetadataOperationType.SET_TTL, storageGroup, ttl);
ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
channel.write(buff);
- newLine();
+ lineNumber.incrementAndGet();
}
public void changeOffset(String path, long offset) throws IOException {
String outputStr = String.format(STRING_TYPE,
MetadataOperationType.CHANGE_OFFSET, path, offset);
ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
channel.write(buff);
- newLine();
+ lineNumber.incrementAndGet();
}
public void changeAlias(String path, String alias) throws IOException {
String outputStr = String.format(STRING_TYPE,
MetadataOperationType.CHANGE_ALIAS, path, alias);
ByteBuffer buff = ByteBuffer.wrap(outputStr.getBytes());
channel.write(buff);
- newLine();
+ lineNumber.incrementAndGet();
}
public static void upgradeMLog(String schemaDir, String logFileName) throws
IOException {
@@ -158,7 +159,7 @@ public class MLogWriter {
}
// upgrading
try (BufferedReader reader = new BufferedReader(new FileReader(logFile));
- BufferedWriter writer = new BufferedWriter(new FileWriter(tmpLogFile,
true));) {
+ BufferedWriter writer = new BufferedWriter(new FileWriter(tmpLogFile,
true))) {
String line;
while ((line = reader.readLine()) != null) {
StringBuilder buf = new StringBuilder();
@@ -179,23 +180,17 @@ public class MLogWriter {
Files.delete(logFile.toPath());
fileOutputStream = new FileOutputStream(logFile, true);
channel = fileOutputStream.getChannel();
- lineNumber = 0;
- }
-
- private void newLine() throws IOException {
- channel.write(ByteBuffer.wrap(System.lineSeparator().getBytes()));
- channel.force(true);
- ++lineNumber;
+ lineNumber.set(0);
}
int getLineNumber() {
- return lineNumber;
+ return lineNumber.get();
}
/**
* only used for initialize a mlog file writer.
*/
void setLineNumber(int number) {
- lineNumber = number;
+ lineNumber.set(number);
}
}
diff --git a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
index f4c6c3f..6886259 100644
--- a/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/metadata/MManager.java
@@ -94,12 +94,12 @@ public class MManager {
public static final String TIME_SERIES_TREE_HEADER = "=== Timeseries Tree
===\n\n";
private static final String TAG_FORMAT = "tag key is %s, tag value is %s,
tlog offset is %d";
- private static final String DEBUG_MSG = "%s : TimeSeries %s is removed from
tag inverted index, ";
+ private static final String DEBUG_MSG = "%s : TimeSeries %s is removed from
tag inverted index, ";
private static final String DEBUG_MSG_1 = "%s: TimeSeries %s's tag info has
been removed from tag inverted index ";
private static final String PREVIOUS_CONDITION = "before deleting it, tag
key is %s, tag value is %s, tlog offset is %d, contains key %b";
-
+
private static final Logger logger = LoggerFactory.getLogger(MManager.class);
-
+
/**
* A thread will check whether the MTree is modified lately each such
interval. Unit: second
*/
@@ -159,11 +159,13 @@ public class MManager {
}
};
- timedCreateMTreeSnapshotThread =
Executors.newSingleThreadScheduledExecutor(r -> new Thread(r,
- "timedCreateMTreeSnapshotThread"));
- timedCreateMTreeSnapshotThread
- .scheduleAtFixedRate(this::checkMTreeModified,
MTREE_SNAPSHOT_THREAD_CHECK_TIME,
- MTREE_SNAPSHOT_THREAD_CHECK_TIME, TimeUnit.SECONDS);
+ if (config.isEnableMTreeSnapshot()) {
+ timedCreateMTreeSnapshotThread =
Executors.newSingleThreadScheduledExecutor(r -> new Thread(r,
+ "timedCreateMTreeSnapshotThread"));
+ timedCreateMTreeSnapshotThread
+ .scheduleAtFixedRate(this::checkMTreeModified,
MTREE_SNAPSHOT_THREAD_CHECK_TIME,
+ MTREE_SNAPSHOT_THREAD_CHECK_TIME, TimeUnit.SECONDS);
+ }
}
/**
@@ -273,7 +275,7 @@ public class MManager {
tagLogFile = null;
}
initialized = false;
- if (timedCreateMTreeSnapshotThread != null) {
+ if (config.isEnableMTreeSnapshot() && timedCreateMTreeSnapshotThread !=
null) {
timedCreateMTreeSnapshotThread.shutdownNow();
timedCreateMTreeSnapshotThread = null;
}
