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 <[email protected]> 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();
