You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ma...@apache.org on 2024/06/17 07:05:44 UTC
(iotdb) branch wal-compress-formal-branch updated (7201ed68fbe -> ee767637a94)
This is an automated email from the ASF dual-hosted git repository.
marklau99 pushed a change to branch wal-compress-formal-branch
in repository https://gitbox.apache.org/repos/asf/iotdb.git
from 7201ed68fbe reuse the byte buffer if possible
new 0e0236650d4 Indicate the encoding of String
new ee767637a94 Edit according to comment
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../dataregion/wal/buffer/WALBuffer.java | 1 +
.../storageengine/dataregion/wal/io/LogWriter.java | 4 ++-
.../dataregion/wal/io/WALByteBufReader.java | 1 +
.../dataregion/wal/io/WALInputStream.java | 40 ++++++++++++++++++----
.../dataregion/wal/io/WALMetaData.java | 3 +-
.../storageengine/dataregion/wal/io/WALWriter.java | 5 +--
.../dataregion/wal/utils/WALEntryPosition.java | 4 ++-
.../wal/compression/WALCompressionTest.java | 4 +--
8 files changed, 48 insertions(+), 14 deletions(-)
(iotdb) 02/02: Edit according to comment
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
marklau99 pushed a commit to branch wal-compress-formal-branch
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit ee767637a941b2ae91e35e8586b056138843dc3f
Author: Liu Xuxin <li...@outlook.com>
AuthorDate: Mon Jun 17 15:04:57 2024 +0800
Edit according to comment
---
.../dataregion/wal/io/WALInputStream.java | 21 ++++++++++++++++++++-
.../dataregion/wal/utils/WALEntryPosition.java | 4 +++-
.../wal/compression/WALCompressionTest.java | 4 ++--
3 files changed, 25 insertions(+), 4 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
index 7cdcb63095f..844c06436b7 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
@@ -18,6 +18,8 @@
*/
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
+import org.apache.iotdb.db.utils.MmapUtil;
+
import org.apache.tsfile.compress.IUnCompressor;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.slf4j.Logger;
@@ -27,6 +29,7 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
+import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.charset.StandardCharsets;
import java.util.Objects;
@@ -217,6 +220,9 @@ public class WALInputStream extends InputStream implements AutoCloseable {
if (Objects.isNull(dataBuffer)
|| dataBuffer.capacity() < segmentInfo.uncompressedSize
|| dataBuffer.capacity() > segmentInfo.uncompressedSize * 2) {
+ if (!Objects.isNull(dataBuffer)) {
+ MmapUtil.clean((MappedByteBuffer) dataBuffer);
+ }
dataBuffer = ByteBuffer.allocateDirect(segmentInfo.uncompressedSize);
}
dataBuffer.clear();
@@ -224,6 +230,9 @@ public class WALInputStream extends InputStream implements AutoCloseable {
if (Objects.isNull(compressedBuffer)
|| compressedBuffer.capacity() < segmentInfo.dataInDiskSize
|| compressedBuffer.capacity() > segmentInfo.dataInDiskSize * 2) {
+ if (!Objects.isNull(compressedBuffer)) {
+ MmapUtil.clean((MappedByteBuffer) compressedBuffer);
+ }
compressedBuffer = ByteBuffer.allocateDirect(segmentInfo.dataInDiskSize);
}
compressedBuffer.clear();
@@ -241,6 +250,9 @@ public class WALInputStream extends InputStream implements AutoCloseable {
if (Objects.isNull(dataBuffer)
|| dataBuffer.capacity() < segmentInfo.dataInDiskSize
|| dataBuffer.capacity() > segmentInfo.dataInDiskSize * 2) {
+ if (!Objects.isNull(dataBuffer)) {
+ MmapUtil.clean((MappedByteBuffer) dataBuffer);
+ }
dataBuffer = ByteBuffer.allocateDirect(segmentInfo.dataInDiskSize);
}
dataBuffer.clear();
@@ -271,7 +283,14 @@ public class WALInputStream extends InputStream implements AutoCloseable {
}
}
- public void skipToGivenPosition(long pos) throws IOException {
+ /**
+ * Since current WAL file is compressed, but some part of the system need to skip the offset of an
+ * uncompressed wal file, this method is used to skip to the given logical position.
+ *
+ * @param pos The logical offset to skip to
+ * @throws IOException If the file is broken or the given position is invalid
+ */
+ public void skipToGivenLogicalPosition(long pos) throws IOException {
if (version == FileVersion.V2) {
channel.position(WALWriter.MAGIC_STRING_BYTES);
long posRemain = pos;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java
index 1e89d8c546d..c794745c6f5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java
@@ -101,8 +101,9 @@ public class WALEntryPosition {
if (!canRead()) {
throw new IOException("Target file hasn't been specified.");
}
+ // TODO: Reuse the file stream
try (WALInputStream is = openReadFileStream()) {
- is.skipToGivenPosition(position);
+ is.skipToGivenLogicalPosition(position);
ByteBuffer buffer = ByteBuffer.allocate(size);
is.read(buffer);
buffer.flip();
@@ -137,6 +138,7 @@ public class WALEntryPosition {
}
public WALInputStream openReadFileStream() throws IOException {
+ // TODO: Refactor this part of code
if (isInSealedFile()) {
walFile = walNode.getWALFile(walFileVersionId);
return new WALInputStream(walFile);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java
index 7ab33343b5d..d187f6107b6 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java
@@ -150,7 +150,7 @@ public class WALCompressionTest {
try (WALInputStream stream = new WALInputStream(walFile)) {
for (int i = 0; i < 100; ++i) {
Pair<Long, InsertRowNode> positionAndNodePair = positionAndEntryPairList.get(i);
- stream.skipToGivenPosition(positionAndNodePair.left);
+ stream.skipToGivenLogicalPosition(positionAndNodePair.left);
/*
Add the allocated buffer size by 2, because the actual serialized size
of InsertRowNode is larger than the estimated value got by serializedSize.
@@ -231,7 +231,7 @@ public class WALCompressionTest {
}
dataOutputStream.close();
ByteBuffer buf = ByteBuffer.wrap(baos.toByteArray());
- // Do not compress it
+ // Compress it
IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(CompressionType.LZ4);
WALTestUtils.setMinCompressionSize(0);
try (WALWriter writer = new WALWriter(walFile)) {
(iotdb) 01/02: Indicate the encoding of String
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
marklau99 pushed a commit to branch wal-compress-formal-branch
in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0e0236650d4703baa26418adb4f03fed206f9a63
Author: Liu Xuxin <li...@outlook.com>
AuthorDate: Mon Jun 17 14:45:58 2024 +0800
Indicate the encoding of String
---
.../dataregion/wal/buffer/WALBuffer.java | 1 +
.../db/storageengine/dataregion/wal/io/LogWriter.java | 4 +++-
.../dataregion/wal/io/WALByteBufReader.java | 1 +
.../dataregion/wal/io/WALInputStream.java | 19 +++++++++++++------
.../storageengine/dataregion/wal/io/WALMetaData.java | 3 ++-
.../db/storageengine/dataregion/wal/io/WALWriter.java | 5 +++--
6 files changed, 23 insertions(+), 10 deletions(-)
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
index 627771bde38..20970aea34b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java
@@ -540,6 +540,7 @@ public class WALBuffer extends AbstractWALBuffer {
boolean forceSuccess = false;
// try to roll log writer
if (info.rollWALFileWriterListener != null
+ // TODO: Control the wal file by the number of WALEntry
|| (forceFlag
&& currentWALFileWriter.originalSize() >= config.getWalFileSizeThresholdInByte())) {
try {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
index 98a42b75280..29335efde5c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java
@@ -34,6 +34,7 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
import java.util.Objects;
/**
@@ -61,7 +62,8 @@ public abstract class LogWriter implements ILogWriter {
this.logStream = new FileOutputStream(logFile, true);
this.logChannel = this.logStream.getChannel();
if (!logFile.exists() || logFile.length() == 0) {
- this.logChannel.write(ByteBuffer.wrap(WALWriter.MAGIC_STRING.getBytes()));
+ this.logChannel.write(
+ ByteBuffer.wrap(WALWriter.MAGIC_STRING.getBytes(StandardCharsets.UTF_8)));
size += logChannel.position();
}
if (IoTDBDescriptor.getInstance().getConfig().getWALCompressionAlgorithm()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
index 3e0909d6c52..882b5ea468c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java
@@ -66,6 +66,7 @@ public class WALByteBufReader implements Closeable {
*/
public ByteBuffer next() throws IOException {
int size = sizeIterator.next();
+ // TODO: Reuse this buffer
ByteBuffer buffer = ByteBuffer.allocate(size);
/*
Notice, we don't need to flip the buffer after calling
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
index 780be52671c..7cdcb63095f 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java
@@ -28,6 +28,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
import java.util.Objects;
public class WALInputStream extends InputStream implements AutoCloseable {
@@ -78,7 +79,8 @@ public class WALInputStream extends InputStream implements AutoCloseable {
ByteBuffer magicStringBuffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
channel.read(magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_BYTES);
magicStringBuffer.flip();
- if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING)) {
+ if (!new String(magicStringBuffer.array(), StandardCharsets.UTF_8)
+ .equals(WALWriter.MAGIC_STRING)) {
// This is a broken wal file
endOffset = channel.size();
return;
@@ -89,16 +91,21 @@ public class WALInputStream extends InputStream implements AutoCloseable {
} else {
// Old version
ByteBuffer magicStringBuffer =
- ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes(StandardCharsets.UTF_8).length);
channel.read(
- magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer,
+ channel.size() - WALWriter.MAGIC_STRING_V1.getBytes(StandardCharsets.UTF_8).length);
magicStringBuffer.flip();
- if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ if (!new String(magicStringBuffer.array(), StandardCharsets.UTF_8)
+ .equals(WALWriter.MAGIC_STRING_V1)) {
// this is a broken wal file
endOffset = channel.size();
return;
} else {
- position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ position =
+ channel.size()
+ - WALWriter.MAGIC_STRING_V1.getBytes(StandardCharsets.UTF_8).length
+ - Integer.BYTES;
}
}
// Read the meta data size
@@ -127,7 +134,7 @@ public class WALInputStream extends InputStream implements AutoCloseable {
channel.position(0);
ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
channel.read(buffer);
- return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
+ return new String(buffer.array(), StandardCharsets.UTF_8).equals(WALWriter.MAGIC_STRING);
}
@Override
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
index 8824bff3a26..7fa634ffbdc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java
@@ -29,6 +29,7 @@ import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
@@ -161,7 +162,7 @@ public class WALMetaData implements SerializedSize {
ByteBuffer magicStringBytes = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
channel.read(magicStringBytes, channel.size() - WALWriter.MAGIC_STRING_BYTES);
magicStringBytes.flip();
- String magicString = new String(magicStringBytes.array());
+ String magicString = new String(magicStringBytes.array(), StandardCharsets.UTF_8);
return magicString.equals(WALWriter.MAGIC_STRING)
|| magicString.contains(WALWriter.MAGIC_STRING_V1);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java
index c5be73892e7..7017b4be6cb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALWriter.java
@@ -30,13 +30,14 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
/** WALWriter writes the binary {@link WALEntry} into .wal file. */
public class WALWriter extends LogWriter {
private static final Logger logger = LoggerFactory.getLogger(WALWriter.class);
public static final String MAGIC_STRING_V1 = "WAL";
public static final String MAGIC_STRING = "V2-WAL";
- public static final int MAGIC_STRING_BYTES = MAGIC_STRING.getBytes().length;
+ public static final int MAGIC_STRING_BYTES = MAGIC_STRING.getBytes(StandardCharsets.UTF_8).length;
private WALFileStatus walFileStatus = WALFileStatus.CONTAINS_NONE_SEARCH_INDEX;
// wal files' metadata
@@ -74,7 +75,7 @@ public class WALWriter extends LogWriter {
metaData.serialize(logFile, buffer);
buffer.putInt(metaDataSize);
// add magic string
- buffer.put(MAGIC_STRING.getBytes());
+ buffer.put(MAGIC_STRING.getBytes(StandardCharsets.UTF_8));
size += buffer.position();
writeMetadata(buffer);
}