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 2020/10/12 07:05:40 UTC
[iotdb] 01/01: use differential logs in wal
This is an automated email from the ASF dual-hosted git repository.
jiangtian pushed a commit to branch improve_wal
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit b060f392a9d74d6bf782a51347c49241171dc159
Author: jt <jt...@163.com>
AuthorDate: Mon Oct 12 15:04:02 2020 +0800
use differential logs in wal
---
.../apache/iotdb/db/qp/physical/PhysicalPlan.java | 109 ++++++++++++++++
.../iotdb/db/qp/physical/crud/InsertRowPlan.java | 52 +++++++-
.../db/qp/physical/crud/InsertTabletPlan.java | 108 ++++++++++++++--
.../java/org/apache/iotdb/db/tools/WalChecker.java | 2 +
.../org/apache/iotdb/db/utils/CommonUtils.java | 22 ++++
.../iotdb/db/writelog/io/BatchLogReader.java | 10 +-
...Reader.java => DifferentialBatchLogReader.java} | 50 +++-----
.../io/DifferentialSingleFileLogReader.java | 29 +++++
.../iotdb/db/writelog/io/ILogReaderFactory.java | 24 ++++
.../iotdb/db/writelog/io/MultiFileLogReader.java | 6 +-
.../iotdb/db/writelog/io/SingleFileLogReader.java | 11 +-
.../writelog/manager/MultiFileLogNodeManager.java | 3 +-
.../db/writelog/node/DifferentialWriteLogNode.java | 140 +++++++++++++++++++++
.../db/writelog/node/ExclusiveWriteLogNode.java | 23 ++--
.../db/writelog/io/MultiFileLogReaderTest.java | 2 +-
15 files changed, 528 insertions(+), 63 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
index 9d26ac0..594c195 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/PhysicalPlan.java
@@ -22,7 +22,9 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
+import java.util.Queue;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.logical.Operator;
@@ -130,6 +132,26 @@ public abstract class PhysicalPlan {
throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
}
+ /**
+ * Serialize the plan into the given buffer. This is provided for WAL, so fields that can be
+ * recovered will not be serialized.
+ *
+ * @param buffer
+ */
+ public void serialize(ByteBuffer buffer, PhysicalPlan base, int baseIndex) {
+ throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
+ }
+
+ /**
+ * Deserialize the plan from the given buffer. This is provided for WAL, and must be used with
+ * serializeToWAL.
+ *
+ * @param buffer
+ */
+ public void deserialize(ByteBuffer buffer, PhysicalPlan base) throws IllegalPathException {
+ throw new UnsupportedOperationException(SERIALIZATION_UNIMPLEMENTED);
+ }
+
protected void putString(ByteBuffer buffer, String value) {
if (value == null) {
buffer.putInt(NULL_VALUE_LEN);
@@ -174,6 +196,11 @@ public abstract class PhysicalPlan {
throw new IOException("unrecognized log type " + typeNum);
}
PhysicalPlanType type = PhysicalPlanType.values()[typeNum];
+ return create(type, buffer);
+ }
+
+ public static PhysicalPlan create(PhysicalPlanType type, ByteBuffer buffer)
+ throws IllegalPathException, IOException {
PhysicalPlan plan;
// TODO-Cluster: support more plans
switch (type) {
@@ -274,6 +301,53 @@ public abstract class PhysicalPlan {
}
return plan;
}
+
+ public static PhysicalPlan create(ByteBuffer buffer,
+ Queue<PhysicalPlan> planWindow) throws IOException,
+ IllegalPathException {
+ int typeNum = buffer.get();
+ if (typeNum >= PhysicalPlanType.values().length) {
+ throw new IOException("unrecognized log type " + typeNum);
+ }
+ PhysicalPlanType type = PhysicalPlanType.values()[typeNum];
+ PhysicalPlan plan;
+ int index;
+ switch (type) {
+ case INSERT:
+ plan = new InsertRowPlan();
+ index = buffer.getInt();
+ if (index < 0) {
+ plan.deserialize(buffer);
+ } else {
+ InsertRowPlan baseInsertRowPlan = (InsertRowPlan) getPlan(planWindow, index);
+ plan.deserialize(buffer, baseInsertRowPlan);
+ }
+ break;
+ case BATCHINSERT:
+ plan = new InsertTabletPlan();
+ index = buffer.getInt();
+ if (index < 0) {
+ plan.deserialize(buffer);
+ } else {
+ InsertTabletPlan baseInsertTabletPlan = (InsertTabletPlan) getPlan(planWindow, index);
+ plan.deserialize(buffer, baseInsertTabletPlan);
+ }
+ break;
+ default:
+ plan = create(type, buffer);
+ }
+ return plan;
+ }
+
+ private static PhysicalPlan getPlan(Queue<PhysicalPlan> planWindow, int index) {
+ Iterator<PhysicalPlan> iterator = planWindow.iterator();
+ PhysicalPlan physicalPlan = null;
+ while (iterator.hasNext() && index >= 0) {
+ physicalPlan = iterator.next();
+ index --;
+ }
+ return physicalPlan;
+ }
}
public enum PhysicalPlanType {
@@ -282,5 +356,40 @@ public abstract class PhysicalPlan {
DELETE_STORAGE_GROUP, SHOW_TIMESERIES, DELETE_TIMESERIES, LOAD_CONFIGURATION
}
+ protected void putDiffTime(long time, long base, ByteBuffer buffer) {
+ long timeDiff = time - base;
+ TimeDiffType diffType = TimeDiffType.fromDiff(timeDiff);
+ buffer.put((byte) diffType.ordinal());
+ switch (diffType) {
+ case INT:
+ buffer.put((byte) TimeDiffType.INT.ordinal());
+ break;
+ case BYTE:
+ buffer.put((byte) timeDiff);
+ break;
+ case SHORT:
+ buffer.putShort((short) timeDiff);
+ break;
+ case LONG:
+ default:
+ // NOTICE HERE
+ buffer.putLong(time);
+ }
+ }
+
+ public enum TimeDiffType {
+ BYTE, SHORT, INT, LONG;
+ public static TimeDiffType fromDiff(long timeDiff) {
+ if (Byte.MIN_VALUE <= timeDiff && timeDiff <= Byte.MAX_VALUE) {
+ return BYTE;
+ } else if (Short.MIN_VALUE <= timeDiff && timeDiff <= Short.MAX_VALUE) {
+ return SHORT;
+ } else if (Integer.MIN_VALUE <= timeDiff && timeDiff <= Integer.MAX_VALUE) {
+ return INT;
+ } else {
+ return LONG;
+ }
+ }
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java
index 1271ecd..5b719ce 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertRowPlan.java
@@ -34,6 +34,7 @@ import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.metadata.mnode.MeasurementMNode;
import org.apache.iotdb.db.qp.logical.Operator;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.utils.CommonUtils;
import org.apache.iotdb.db.utils.TestOnly;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
@@ -71,8 +72,10 @@ public class InsertRowPlan extends InsertPlan {
this.measurements = measurementList;
this.dataTypes = new TSDataType[measurements.length];
// We need to create an Object[] for the data type casting, because we can not set Float, Long to String[i]
- this.values = new Object[measurements.length];
- System.arraycopy(insertValues, 0, values, 0, measurements.length);
+ if (insertValues != null) {
+ this.values = new Object[measurements.length];
+ System.arraycopy(insertValues, 0, values, 0, measurements.length);
+ }
isNeedInferType = true;
}
@@ -285,7 +288,7 @@ public class InsertRowPlan extends InsertPlan {
}
}
- private void putValues(ByteBuffer buffer) throws QueryProcessException {
+ public void putValues(ByteBuffer buffer) throws QueryProcessException {
for (int i = 0; i < values.length; i++) {
// types are not determined, the situation mainly occurs when the plan uses string values
// and is forwarded to other nodes
@@ -411,6 +414,49 @@ public class InsertRowPlan extends InsertPlan {
}
@Override
+ public void serialize(ByteBuffer buffer, PhysicalPlan base, int baseIndex) {
+ InsertRowPlan baseInsertRowPlan = (InsertRowPlan) base;
+ int type = PhysicalPlanType.INSERT.ordinal();
+ buffer.put((byte) type);
+ buffer.putInt(baseIndex);
+
+ putDiffTime(this.getTime(), baseInsertRowPlan.getTime(), buffer);
+
+ buffer
+ .putInt(this.getMeasurements().length - (this.getFailedMeasurements() == null ? 0 :
+ this.getFailedMeasurements().size()));
+
+ try {
+ this.putValues(buffer);
+ } catch (QueryProcessException e) {
+ logger.error("Cannot put values of {} into logBuffer", this, e);
+ }
+
+ // the types are not inferred before the plan is serialized
+ buffer.put((byte) (this.isNeedInferType() ? 1 : 0));
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer, PhysicalPlan base) {
+ InsertRowPlan baseInsertRowPlan = (InsertRowPlan) base;
+
+ this.time = buffer.getLong();
+ this.deviceId = baseInsertRowPlan.deviceId;
+
+ this.measurements = baseInsertRowPlan.measurements;
+
+ this.dataTypes = new TSDataType[measurements.length];
+ this.values = new Object[measurements.length];
+ try {
+ fillValues(buffer);
+ } catch (QueryProcessException e) {
+ logger.error("Cannot fill values of {} from logBuffer", this, e);
+ }
+
+ isNeedInferType = buffer.get() == 1;
+ }
+
+ @Override
public String toString() {
return "deviceId: " + deviceId + ", time: " + time;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java
index f7db816..5c824ff 100644
--- a/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java
+++ b/server/src/main/java/org/apache/iotdb/db/qp/physical/crud/InsertTabletPlan.java
@@ -18,6 +18,7 @@
*/
package org.apache.iotdb.db.qp.physical.crud;
+import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -26,9 +27,15 @@ import java.util.List;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.metadata.PartialPath;
import org.apache.iotdb.db.qp.logical.Operator.OperatorType;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
import org.apache.iotdb.db.utils.QueryDataSetUtils;
+import org.apache.iotdb.tsfile.common.conf.TSFileDescriptor;
+import org.apache.iotdb.tsfile.encoding.decoder.Decoder;
+import org.apache.iotdb.tsfile.encoding.encoder.Encoder;
+import org.apache.iotdb.tsfile.encoding.encoder.TSEncodingBuilder;
import org.apache.iotdb.tsfile.exception.write.UnSupportedDataTypeException;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSEncoding;
import org.apache.iotdb.tsfile.read.TimeValuePair;
import org.apache.iotdb.tsfile.utils.Binary;
import org.apache.iotdb.tsfile.utils.BytesUtils;
@@ -39,9 +46,12 @@ import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsDouble;
import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsFloat;
import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsInt;
import org.apache.iotdb.tsfile.utils.TsPrimitiveType.TsLong;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class InsertTabletPlan extends InsertPlan {
+ private static final Logger logger = LoggerFactory.getLogger(InsertTabletPlan.class);
private static final String DATATYPE_UNSUPPORTED = "Data type %s is not supported.";
private long[] times; // times should be sorted. It is done in the session API.
@@ -137,9 +147,7 @@ public class InsertTabletPlan extends InsertPlan {
stream.writeInt(end - start);
if (timeBuffer == null) {
- for (int i = start; i < end; i++) {
- stream.writeLong(times[i]);
- }
+ serializeTimes(stream);
} else {
stream.write(timeBuffer.array());
timeBuffer = null;
@@ -174,12 +182,45 @@ public class InsertTabletPlan extends InsertPlan {
}
}
+ serializeTimeValue(buffer);
+ }
+
+ @Override
+ public void serialize(ByteBuffer buffer, PhysicalPlan base, int baseIndex) {
+ int type = PhysicalPlanType.BATCHINSERT.ordinal();
+ buffer.put((byte) type);
+ buffer.putInt(baseIndex);
+
+ buffer
+ .putInt(this.getMeasurements().length - (this.getFailedMeasurements() == null ? 0 :
+ this.getFailedMeasurements().size()));
+
+ serializeTimeValue(buffer);
+ }
+
+ @Override
+ public void deserialize(ByteBuffer buffer, PhysicalPlan base) {
+ InsertTabletPlan baseInsertTabletPlan = (InsertTabletPlan) base;
+
+ this.deviceId = baseInsertTabletPlan.deviceId;
+
+ this.measurements = baseInsertTabletPlan.getMeasurements();
+
+ this.dataTypes = baseInsertTabletPlan.dataTypes;
+
+ int rows = buffer.getInt();
+ rowCount = rows;
+ this.times = new long[rows];
+ deserializeTimes(buffer, rows);
+
+ columns = QueryDataSetUtils.readValuesFromBuffer(buffer, dataTypes, this.measurements.length, rows);
+ }
+
+ public void serializeTimeValue(ByteBuffer buffer) {
buffer.putInt(end - start);
if (timeBuffer == null) {
- for (int i = start; i < end; i++) {
- buffer.putLong(times[i]);
- }
+ serializeTimes(buffer);
} else {
buffer.put(timeBuffer.array());
timeBuffer = null;
@@ -193,6 +234,41 @@ public class InsertTabletPlan extends InsertPlan {
}
}
+ private void serializeTimes(ByteBuffer buffer) {
+ byte[] bytes = serializeTimesToArray();
+ buffer.putInt(bytes.length);
+ buffer.put(bytes);
+ }
+
+ private void serializeTimes(DataOutputStream stream) throws IOException {
+ byte[] bytes = serializeTimesToArray();
+ stream.writeInt(bytes.length);
+ stream.write(bytes);
+ }
+
+ private byte[] serializeTimesToArray() {
+ Encoder timeEncoder = getTimeEncoder();
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+
+ for (int i = start; i < end; i++) {
+ timeEncoder.encode(times[i], byteArrayOutputStream);
+ }
+ try {
+ timeEncoder.flush(byteArrayOutputStream);
+ } catch (IOException e) {
+ logger.error("Cannot encode time of {}", this);
+ }
+ return byteArrayOutputStream.toByteArray();
+ }
+
+ public Encoder getTimeEncoder() {
+ TSEncoding timeEncoding = TSEncoding
+ .valueOf(TSFileDescriptor.getInstance().getConfig().getTimeEncoder());
+ TSDataType timeType = TSDataType
+ .valueOf(TSFileDescriptor.getInstance().getConfig().getTimeSeriesDataType());
+ return TSEncodingBuilder.getEncodingBuilder(timeEncoding).getEncoder(timeType);
+ }
+
private void serializeValues(DataOutputStream outputStream) throws IOException {
for (int i = 0; i < measurements.length; i++) {
serializeColumn(dataTypes[i], columns[i], outputStream, start, end);
@@ -328,11 +404,29 @@ public class InsertTabletPlan extends InsertPlan {
int rows = buffer.getInt();
rowCount = rows;
this.times = new long[rows];
- times = QueryDataSetUtils.readTimesFromBuffer(buffer, rows);
+ deserializeTimes(buffer, rows);
columns = QueryDataSetUtils.readValuesFromBuffer(buffer, dataTypes, measurementSize, rows);
}
+ private void deserializeTimes(ByteBuffer buffer, int number) {
+ Decoder defaultTimeDecoder = Decoder.getDecoderByType(
+ TSEncoding.valueOf(TSFileDescriptor.getInstance().getConfig().getTimeEncoder()),
+ TSDataType.INT64);
+ int timeSize = buffer.getInt();
+ byte[] bytes = new byte[timeSize];
+ buffer.get(bytes);
+
+ int i = 0;
+ try {
+ while (defaultTimeDecoder.hasNext(buffer) && i < number) {
+ times[i++] = defaultTimeDecoder.readLong(buffer);
+ }
+ } catch (IOException e) {
+ logger.error("Cannot decode time of {}", this);
+ }
+ }
+
public void setDataTypes(List<Integer> dataTypes) {
this.dataTypes = new TSDataType[dataTypes.size()];
for (int i = 0; i < dataTypes.size(); i++) {
diff --git a/server/src/main/java/org/apache/iotdb/db/tools/WalChecker.java b/server/src/main/java/org/apache/iotdb/db/tools/WalChecker.java
index 04215ff..468e423 100644
--- a/server/src/main/java/org/apache/iotdb/db/tools/WalChecker.java
+++ b/server/src/main/java/org/apache/iotdb/db/tools/WalChecker.java
@@ -18,10 +18,12 @@
*/
package org.apache.iotdb.db.tools;
+import static org.apache.iotdb.db.writelog.node.DifferentialWriteLogNode.WINDOW_LENGTH;
import static org.apache.iotdb.db.writelog.node.ExclusiveWriteLogNode.WAL_FILE_NAME;
import java.io.File;
import java.io.IOException;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
diff --git a/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java b/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java
index 7bd7bf2..4fa0e96 100644
--- a/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java
+++ b/server/src/main/java/org/apache/iotdb/db/utils/CommonUtils.java
@@ -22,10 +22,14 @@ import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.util.Queue;
import java.util.stream.Stream;
import org.apache.iotdb.db.exception.query.QueryProcessException;
import org.apache.iotdb.db.qp.constant.SQLConstant;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertRowPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.fileSystem.FSFactoryProducer;
import org.apache.iotdb.tsfile.utils.Binary;
@@ -139,4 +143,22 @@ public class CommonUtils {
}
throw new QueryProcessException("The BOOLEAN should be true/TRUE, false/FALSE or 0/1");
}
+
+ public static void updatePlanWindow(PhysicalPlan plan, int windowLength,
+ Queue<PhysicalPlan> planWindow) {
+ if (planWindow.size() >= windowLength) {
+ planWindow.remove();
+ }
+ // remove unnecessary fields as bases to reduce memory footprint
+ if (plan instanceof InsertRowPlan) {
+ InsertRowPlan insertRowPlan = (InsertRowPlan) plan;
+ plan = new InsertRowPlan(insertRowPlan.getDeviceId(),
+ insertRowPlan.getTime(), insertRowPlan.getMeasurements(), null);
+ } else if (plan instanceof InsertTabletPlan) {
+ InsertTabletPlan insertTabletPlan = (InsertTabletPlan) plan;
+ plan = new InsertTabletPlan(insertTabletPlan.getDeviceId(),
+ insertTabletPlan.getMeasurements());
+ }
+ planWindow.add(plan);
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/io/BatchLogReader.java b/server/src/main/java/org/apache/iotdb/db/writelog/io/BatchLogReader.java
index e0bf503..84095fd 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/io/BatchLogReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/io/BatchLogReader.java
@@ -37,16 +37,20 @@ public class BatchLogReader implements ILogReader{
private static Logger logger = LoggerFactory.getLogger(BatchLogReader.class);
- private Iterator<PhysicalPlan> planIterator;
+ Iterator<PhysicalPlan> planIterator;
- private boolean fileCorrupted = false;
+ boolean fileCorrupted = false;
+
+ BatchLogReader() {
+
+ }
BatchLogReader(ByteBuffer buffer) {
List<PhysicalPlan> logs = readLogs(buffer);
this.planIterator = logs.iterator();
}
- private List<PhysicalPlan> readLogs(ByteBuffer buffer) {
+ List<PhysicalPlan> readLogs(ByteBuffer buffer) {
List<PhysicalPlan> plans = new ArrayList<>();
while (buffer.position() != buffer.limit()) {
try {
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/io/BatchLogReader.java b/server/src/main/java/org/apache/iotdb/db/writelog/io/DifferentialBatchLogReader.java
similarity index 66%
copy from server/src/main/java/org/apache/iotdb/db/writelog/io/BatchLogReader.java
copy to server/src/main/java/org/apache/iotdb/db/writelog/io/DifferentialBatchLogReader.java
index e0bf503..3b97b6e 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/io/BatchLogReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/io/DifferentialBatchLogReader.java
@@ -17,40 +17,42 @@
* under the License.
*/
+
package org.apache.iotdb.db.writelog.io;
+import static org.apache.iotdb.db.writelog.node.DifferentialWriteLogNode.WINDOW_LENGTH;
+
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.Iterator;
import java.util.List;
+import java.util.Queue;
import org.apache.iotdb.db.exception.metadata.IllegalPathException;
import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan.Factory;
+import org.apache.iotdb.db.utils.CommonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/**
- * BatchedLogReader reads logs from a binary batch of log in the format of ByteBuffer. The
- * ByteBuffer must be readable.
- */
-public class BatchLogReader implements ILogReader{
-
- private static Logger logger = LoggerFactory.getLogger(BatchLogReader.class);
-
- private Iterator<PhysicalPlan> planIterator;
+public class DifferentialBatchLogReader extends BatchLogReader {
- private boolean fileCorrupted = false;
+ private static final Logger logger = LoggerFactory.getLogger(DifferentialBatchLogReader.class);
+ private Queue<PhysicalPlan> planWindow;
- BatchLogReader(ByteBuffer buffer) {
+ public DifferentialBatchLogReader(ByteBuffer buffer, Queue<PhysicalPlan> planWindow) {
+ this.planWindow = planWindow;
List<PhysicalPlan> logs = readLogs(buffer);
this.planIterator = logs.iterator();
}
- private List<PhysicalPlan> readLogs(ByteBuffer buffer) {
+ @Override
+ List<PhysicalPlan> readLogs(ByteBuffer buffer) {
List<PhysicalPlan> plans = new ArrayList<>();
while (buffer.position() != buffer.limit()) {
try {
- plans.add(PhysicalPlan.Factory.create(buffer));
+ PhysicalPlan physicalPlan = Factory.create(buffer, planWindow);
+ plans.add(physicalPlan);
+ CommonUtils.updatePlanWindow(physicalPlan, WINDOW_LENGTH, planWindow);
} catch (IOException | IllegalPathException e) {
logger.error("Cannot deserialize PhysicalPlans from ByteBuffer, ignore remaining logs", e);
fileCorrupted = true;
@@ -59,24 +61,4 @@ public class BatchLogReader implements ILogReader{
}
return plans;
}
-
-
- @Override
- public void close() {
- // nothing to be closed
- }
-
- @Override
- public boolean hasNext() {
- return planIterator.hasNext();
- }
-
- @Override
- public PhysicalPlan next() {
- return planIterator.next();
- }
-
- public boolean isFileCorrupted() {
- return fileCorrupted;
- }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/io/DifferentialSingleFileLogReader.java b/server/src/main/java/org/apache/iotdb/db/writelog/io/DifferentialSingleFileLogReader.java
new file mode 100644
index 0000000..73a4ee5
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/io/DifferentialSingleFileLogReader.java
@@ -0,0 +1,29 @@
+/*
+ * * 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 [...]
+ */
+
+package org.apache.iotdb.db.writelog.io;
+
+import static org.apache.iotdb.db.writelog.node.DifferentialWriteLogNode.WINDOW_LENGTH;
+
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.nio.ByteBuffer;
+import java.util.ArrayDeque;
+import java.util.Queue;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+
+public class DifferentialSingleFileLogReader extends SingleFileLogReader {
+
+ private Queue<PhysicalPlan> planWindow;
+
+ public DifferentialSingleFileLogReader(File logFile) throws FileNotFoundException {
+ super(logFile);
+ this.planWindow = new ArrayDeque<>(WINDOW_LENGTH);
+ }
+
+ @Override
+ BatchLogReader getBatchLogReader(ByteBuffer byteBuffer) {
+ return new DifferentialBatchLogReader(byteBuffer, planWindow);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/io/ILogReaderFactory.java b/server/src/main/java/org/apache/iotdb/db/writelog/io/ILogReaderFactory.java
new file mode 100644
index 0000000..210ab9b
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/io/ILogReaderFactory.java
@@ -0,0 +1,24 @@
+/*
+ * 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.writelog.io;
+
+import java.io.File;
+
+
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/io/MultiFileLogReader.java b/server/src/main/java/org/apache/iotdb/db/writelog/io/MultiFileLogReader.java
index cf7b25f..bb5f7f8 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/io/MultiFileLogReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/io/MultiFileLogReader.java
@@ -33,9 +33,11 @@ public class MultiFileLogReader implements ILogReader {
private SingleFileLogReader currentReader;
private File[] files;
private int fileIdx = 0;
+ private SingleFileLogReader.Factory fileLogReaderFactory;
- public MultiFileLogReader(File[] files) {
+ public MultiFileLogReader(File[] files, SingleFileLogReader.Factory fileLogReaderFactory) {
this.files = files;
+ this.fileLogReaderFactory = fileLogReaderFactory;
}
@Override
@@ -51,7 +53,7 @@ public class MultiFileLogReader implements ILogReader {
return false;
}
if (currentReader == null) {
- currentReader = new SingleFileLogReader(files[fileIdx++]);
+ currentReader = fileLogReaderFactory.create(files[fileIdx++]);
}
if (currentReader.hasNext()) {
return true;
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/io/SingleFileLogReader.java b/server/src/main/java/org/apache/iotdb/db/writelog/io/SingleFileLogReader.java
index d504cbc..e01c530 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/io/SingleFileLogReader.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/io/SingleFileLogReader.java
@@ -88,7 +88,7 @@ public class SingleFileLogReader implements ILogReader {
+ "%d Calculated: %d.", idx, checkSum, checkSummer.getValue()));
}
- batchLogReader = new BatchLogReader(ByteBuffer.wrap(buffer));
+ batchLogReader = getBatchLogReader(ByteBuffer.wrap(buffer));
fileCorrupted = fileCorrupted || batchLogReader.isFileCorrupted();
} catch (Exception e) {
logger.error("Cannot read more PhysicalPlans from {} because", filepath, e);
@@ -98,6 +98,10 @@ public class SingleFileLogReader implements ILogReader {
return true;
}
+ BatchLogReader getBatchLogReader(ByteBuffer byteBuffer) {
+ return new BatchLogReader(byteBuffer);
+ }
+
@Override
public PhysicalPlan next() {
if (!hasNext()){
@@ -129,4 +133,9 @@ public class SingleFileLogReader implements ILogReader {
public boolean isFileCorrupted() {
return fileCorrupted;
}
+
+ public interface Factory {
+
+ SingleFileLogReader create(File file) throws FileNotFoundException;
+ }
}
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java b/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
index f07d69f..83e8687 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/manager/MultiFileLogNodeManager.java
@@ -29,6 +29,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.exception.StartupException;
import org.apache.iotdb.db.service.IService;
import org.apache.iotdb.db.service.ServiceType;
+import org.apache.iotdb.db.writelog.node.DifferentialWriteLogNode;
import org.apache.iotdb.db.writelog.node.ExclusiveWriteLogNode;
import org.apache.iotdb.db.writelog.node.WriteLogNode;
import org.slf4j.Logger;
@@ -78,7 +79,7 @@ public class MultiFileLogNodeManager implements WriteLogNodeManager, IService {
public WriteLogNode getNode(String identifier) {
WriteLogNode node = nodeMap.get(identifier);
if (node == null) {
- node = new ExclusiveWriteLogNode(identifier);
+ node = new DifferentialWriteLogNode(identifier);
WriteLogNode oldNode = nodeMap.putIfAbsent(identifier, node);
if (oldNode != null) {
return oldNode;
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/node/DifferentialWriteLogNode.java b/server/src/main/java/org/apache/iotdb/db/writelog/node/DifferentialWriteLogNode.java
new file mode 100644
index 0000000..280181f
--- /dev/null
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/node/DifferentialWriteLogNode.java
@@ -0,0 +1,140 @@
+/*
+ * 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.writelog.node;
+
+import java.io.File;
+import java.nio.BufferOverflowException;
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.Queue;
+import org.apache.iotdb.db.engine.fileSystem.SystemFileFactory;
+import org.apache.iotdb.db.qp.physical.PhysicalPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertPlan;
+import org.apache.iotdb.db.qp.physical.crud.InsertTabletPlan;
+import org.apache.iotdb.db.utils.CommonUtils;
+import org.apache.iotdb.db.writelog.io.DifferentialSingleFileLogReader;
+import org.apache.iotdb.db.writelog.io.ILogReader;
+import org.apache.iotdb.db.writelog.io.MultiFileLogReader;
+import org.apache.iotdb.tsfile.utils.Pair;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DifferentialWriteLogNode extends ExclusiveWriteLogNode {
+
+ private static final Logger logger = LoggerFactory.getLogger(DifferentialWriteLogNode.class);
+ // TODO: make WINDOW_LENGTH a config
+ public static final int WINDOW_LENGTH = 100;
+ private Queue<PhysicalPlan> planWindow;
+
+ /**
+ * constructor of ExclusiveWriteLogNode.
+ *
+ * @param identifier ExclusiveWriteLogNode identifier
+ */
+ public DifferentialWriteLogNode(String identifier) {
+ super(identifier);
+ planWindow = new ArrayDeque<>(WINDOW_LENGTH);
+ }
+
+ @Override
+ void putLog(PhysicalPlan plan) {
+ logBuffer.mark();
+ Pair<PhysicalPlan, Integer> similarPlanIndex = findSimilarPlan(plan);
+ try {
+ serialize(plan, similarPlanIndex);
+ } catch (BufferOverflowException e) {
+ logger.info("WAL BufferOverflow !");
+ logBuffer.reset();
+ sync();
+ serialize(plan, similarPlanIndex);
+ }
+ bufferedLogNum ++;
+ CommonUtils.updatePlanWindow(plan, WINDOW_LENGTH, planWindow);
+ }
+
+ @Override
+ void nextFileWriter() {
+ super.nextFileWriter();
+ planWindow.clear();
+ }
+
+ private void serialize(PhysicalPlan plan, Pair<PhysicalPlan, Integer> similarPlanIndex) {
+ if (similarPlanIndex == null) {
+ plan.serialize(logBuffer);
+ } else {
+ serializeDifferentially(plan, similarPlanIndex);
+ }
+ }
+
+ private void serializeNonDifferentially(PhysicalPlan plan) {
+ logBuffer.putInt(-1);
+ plan.serialize(logBuffer);
+ }
+
+ private void serializeDifferentially(PhysicalPlan plan,
+ Pair<PhysicalPlan, Integer> similarPlanIndex) {
+ if (plan instanceof InsertPlan) {
+ serializeDifferentially(((InsertPlan) plan), ((InsertPlan) similarPlanIndex.left),
+ similarPlanIndex.right);
+ } else {
+ serializeNonDifferentially(plan);
+ }
+ }
+
+ private void serializeDifferentially(InsertPlan plan, InsertPlan base, int index) {
+ plan.serialize(logBuffer, base, index);
+ }
+
+ private Pair<PhysicalPlan, Integer> findSimilarPlan(PhysicalPlan plan) {
+ int index = -1;
+ for (PhysicalPlan next : planWindow) {
+ index++;
+ if (isPlanSimilarEnough(plan, next)) {
+ return new Pair<>(plan, index);
+ }
+ }
+ return null;
+ }
+
+ private boolean isPlanSimilarEnough(PhysicalPlan planA, PhysicalPlan planB) {
+ if (!planA.getClass().equals(planB.getClass())) {
+ return false;
+ }
+ if (planA instanceof InsertTabletPlan && planB instanceof InsertTabletPlan) {
+ return isPlanSimilarEnough(((InsertTabletPlan) planA), ((InsertTabletPlan) planB));
+ }
+ return false;
+ }
+
+ private boolean isPlanSimilarEnough(InsertPlan planA, InsertPlan planB) {
+ return planA.getDeviceId().equals(planB.getDeviceId()) &&
+ Arrays.equals(planA.getMeasurements(), planB.getMeasurements()) &&
+ Arrays.equals(planA.getDataTypes(), planB.getDataTypes());
+ }
+
+ @Override
+ public ILogReader getLogReader() {
+ File[] logFiles = SystemFileFactory.INSTANCE.getFile(logDirectory).listFiles();
+ Arrays.sort(logFiles,
+ Comparator.comparingInt(f -> Integer.parseInt(f.getName().replace(WAL_FILE_NAME, ""))));
+ return new MultiFileLogReader(logFiles, DifferentialSingleFileLogReader::new);
+ }
+}
diff --git a/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java b/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
index f206395..023d23a 100644
--- a/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
+++ b/server/src/main/java/org/apache/iotdb/db/writelog/node/ExclusiveWriteLogNode.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.db.writelog.io.ILogReader;
import org.apache.iotdb.db.writelog.io.ILogWriter;
import org.apache.iotdb.db.writelog.io.LogWriter;
import org.apache.iotdb.db.writelog.io.MultiFileLogReader;
+import org.apache.iotdb.db.writelog.io.SingleFileLogReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,23 +48,23 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
public static final String WAL_FILE_NAME = "wal";
private static final Logger logger = LoggerFactory.getLogger(ExclusiveWriteLogNode.class);
- private String identifier;
+ String identifier;
- private String logDirectory;
+ String logDirectory;
private ILogWriter currentFileWriter;
private IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- private ByteBuffer logBuffer = ByteBuffer
+ ByteBuffer logBuffer = ByteBuffer
.allocate(IoTDBDescriptor.getInstance().getConfig().getWalBufferSize());
- private ReadWriteLock lock = new ReentrantReadWriteLock();
+ ReadWriteLock lock = new ReentrantReadWriteLock();
private long fileId = 0;
private long lastFlushedId = 0;
- private int bufferedLogNum = 0;
+ int bufferedLogNum = 0;
/**
* constructor of ExclusiveWriteLogNode.
@@ -75,7 +76,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
this.logDirectory =
DirectoryManager.getInstance().getWALFolder() + File.separator + this.identifier;
if (SystemFileFactory.INSTANCE.getFile(logDirectory).mkdirs()) {
- logger.info("create the WAL folder {}." + logDirectory);
+ logger.info("create the WAL folder {}.", logDirectory);
}
}
@@ -96,7 +97,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
}
}
- private void putLog(PhysicalPlan plan) {
+ void putLog(PhysicalPlan plan) {
logBuffer.mark();
try {
plan.serialize(logBuffer);
@@ -183,7 +184,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
File[] logFiles = SystemFileFactory.INSTANCE.getFile(logDirectory).listFiles();
Arrays.sort(logFiles,
Comparator.comparingInt(f -> Integer.parseInt(f.getName().replace(WAL_FILE_NAME, ""))));
- return new MultiFileLogReader(logFiles);
+ return new MultiFileLogReader(logFiles, SingleFileLogReader::new);
}
private void discard(File logFile) {
@@ -214,7 +215,7 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
}
}
- private void sync() {
+ void sync() {
lock.writeLock().lock();
try {
if (bufferedLogNum == 0) {
@@ -235,14 +236,14 @@ public class ExclusiveWriteLogNode implements WriteLogNode, Comparable<Exclusive
}
}
- private ILogWriter getCurrentFileWriter() {
+ ILogWriter getCurrentFileWriter() {
if (currentFileWriter == null) {
nextFileWriter();
}
return currentFileWriter;
}
- private void nextFileWriter() {
+ void nextFileWriter() {
fileId++;
File newFile = SystemFileFactory.INSTANCE.getFile(logDirectory, WAL_FILE_NAME + fileId);
if (newFile.getParentFile().mkdirs()) {
diff --git a/server/src/test/java/org/apache/iotdb/db/writelog/io/MultiFileLogReaderTest.java b/server/src/test/java/org/apache/iotdb/db/writelog/io/MultiFileLogReaderTest.java
index f7b5178..ec1a536 100644
--- a/server/src/test/java/org/apache/iotdb/db/writelog/io/MultiFileLogReaderTest.java
+++ b/server/src/test/java/org/apache/iotdb/db/writelog/io/MultiFileLogReaderTest.java
@@ -71,7 +71,7 @@ public class MultiFileLogReaderTest {
@Test
public void test() throws IOException {
- MultiFileLogReader reader = new MultiFileLogReader(logFiles);
+ MultiFileLogReader reader = new MultiFileLogReader(logFiles, SingleFileLogReader::new);
int i = 0;
while (reader.hasNext()) {
PhysicalPlan plan = reader.next();