You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@iotdb.apache.org by "THUMarkLau (via GitHub)" <gi...@apache.org> on 2024/05/07 08:11:45 UTC
[PR] Support WAL Compression [iotdb]
THUMarkLau opened a new pull request, #12476:
URL: https://github.com/apache/iotdb/pull/12476
The write-ahead logging consumes a significant amount of I/O resources during data writing, sometimes even 2-7 times the amount of TsFile I/O. Currently, the write-ahead logging in the system is uncompressed. Therefore, compressing the write-ahead logging can save I/O resources and enhance system performance. This PR primarily introduces support for compressing write-ahead logging, featuring the following characteristics:
1. The basic unit for compression and decompression is a WAL Buffer, which may contain several WAL Entries, with a default size of 16MB. This design achieves a good compression ratio while minimizing modifications to upper-layer applications.
2. To maintain compatibility with previous versions, we have replaced the original Magic String in the WAL file with a Version String to distinguish between different versions of WAL. If there are subsequent new versions of WAL, we can continue to add new Version Strings.
3. Multiple compression methods are supported. Currently, the choice of compression method is not open to users, but theoretically, our design supports various compression methods, and we may open this option in the future.
Changes in system resource consumption and system performance caused by this feature will be supplemented in subsequent comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2119782229
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
 [B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1622376756
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java:
##########
@@ -75,6 +76,9 @@ protected IWALNode createWALNode(String identifier, String folder) {
} catch (FileNotFoundException e) {
logger.error("Fail to create wal node", e);
return WALFakeNode.getFailureInstance(e);
+ } catch (IOException e) {
+ logger.error("Meet exception when creating wal node", e);
+ return WALFakeNode.getFailureInstance(e);
}
Review Comment:
The initializer of WALWriter will write head magic string to the file, so an IOException will be throw rather than FileNotFoundException.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1622386376
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java:
##########
@@ -345,12 +344,13 @@ public long getFirstValidWALVersionId() {
}
/** Update wal disk cost of active memTables. */
- public void updateCostOfActiveMemTables(Map<Long, Long> memTableId2WalDiskUsage) {
+ public void updateCostOfActiveMemTables(
+ Map<Long, Long> memTableId2WalDiskUsage, double compressionRate) {
for (Map.Entry<Long, Long> memTableWalUsage : memTableId2WalDiskUsage.entrySet()) {
memTableId2Info.computeIfPresent(
memTableWalUsage.getKey(),
(k, v) -> {
- v.addWalDiskUsage(memTableWalUsage.getValue());
+ v.addWalDiskUsage((long) (memTableWalUsage.getValue() * compressionRate));
Review Comment:
It's not about the file size. Multiple memtables will share one wal file, so we cannot get the disk usage of one memtable by getting the size of wal file directly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632196698
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
+ }
+
+ @Override
+ public int read() throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ return dataBuffer.get() & 0xFF;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ if (dataBuffer.remaining() >= len) {
+ dataBuffer.get(b, off, len);
+ return len;
+ }
+ int toBeRead = len;
+ while (toBeRead > 0) {
+ int remaining = dataBuffer.remaining();
+ int bytesRead = Math.min(remaining, toBeRead);
+ dataBuffer.get(b, off, bytesRead);
+ off += bytesRead;
+ toBeRead -= bytesRead;
+ if (toBeRead > 0) {
+ loadNextSegment();
+ }
+ }
+ return len;
+ }
+
+ @Override
+ public void close() throws IOException {
+ channel.close();
+ dataBuffer = null;
+ }
+
+ @Override
+ public int available() throws IOException {
+ long size = (endOffset - channel.position());
+ if (!Objects.isNull(dataBuffer)) {
+ size += dataBuffer.limit() - dataBuffer.position();
+ }
+ return (int) size;
+ }
+
+ private void loadNextSegment() throws IOException {
+ if (channel.position() >= endOffset) {
+ throw new IOException("End of file");
+ }
+ if (version == FileVersion.V2) {
+ loadNextSegmentV2();
+ } else if (version == FileVersion.V1) {
+ loadNextSegmentV1();
+ } else {
+ tryLoadSegment();
+ }
+ }
+
+ private void loadNextSegmentV1() throws IOException {
+ // just read raw data as input
+ if (channel.position() >= fileSize) {
+ throw new IOException("Unexpected end of file");
+ }
+ if (Objects.isNull(dataBuffer)) {
+ // read 128 KB
+ dataBuffer = ByteBuffer.allocate(128 * 1024);
+ }
+ dataBuffer.clear();
+ channel.read(dataBuffer);
+ dataBuffer.flip();
+ }
+
+ private void loadNextSegmentV2() throws IOException {
+ headerBuffer.clear();
+ if (channel.read(headerBuffer) != Integer.BYTES + 1) {
+ throw new IOException("Unexpected end of file");
+ }
+ // compressionType originalSize compressedSize
+ headerBuffer.flip();
+ CompressionType compressionType = CompressionType.deserialize(headerBuffer.get());
+ int dataBufferSize = headerBuffer.getInt();
+ if (compressionType != CompressionType.UNCOMPRESSED) {
+ compressedHeader.clear();
+ if (channel.read(compressedHeader) != Integer.BYTES) {
+ throw new IOException("Unexpected end of file");
+ }
+ compressedHeader.flip();
+ int uncompressedSize = compressedHeader.getInt();
+ dataBuffer = ByteBuffer.allocateDirect(uncompressedSize);
+ ByteBuffer compressedData = ByteBuffer.allocateDirect(dataBufferSize);
Review Comment:
Fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632189743
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -535,12 +535,13 @@ public void run() {
memTableIdsOfWal
.computeIfAbsent(currentWALFileVersion, memTableIds -> new HashSet<>())
.addAll(info.metaData.getMemTablesId());
- checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage);
+ checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage, compressionRate);
boolean forceSuccess = false;
// try to roll log writer
if (info.rollWALFileWriterListener != null
- || (forceFlag && currentWALFileWriter.size() >= config.getWalFileSizeThresholdInByte())) {
+ || (forceFlag
+ && currentWALFileWriter.originalSize() >= config.getWalFileSizeThresholdInByte())) {
Review Comment:
The original WAL file open/close control used the size of the WAL file. This PR only made some simple modifications on top of that to accommodate compression. It is reasonable to control the opening and closing of WAL files using the number of WALFlushListeners, but this goes beyond the scope of this PR and would make the PR's integration cycle longer and the impact more unpredictable. Therefore, it is suggested that controlling the opening and closing of WAL files using WALFlushListeners be planned for future work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1636194330
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java:
##########
@@ -100,11 +101,11 @@ ByteBuffer read() throws IOException {
if (!canRead()) {
throw new IOException("Target file hasn't been specified.");
}
- try (FileChannel channel = openReadFileChannel()) {
+ try (WALInputStream is = openReadFileStream()) {
+ is.skipToGivenPosition(position);
ByteBuffer buffer = ByteBuffer.allocate(size);
- channel.position(position);
- channel.read(buffer);
- buffer.clear();
+ is.read(buffer);
+ buffer.flip();
return buffer;
}
Review Comment:
Just leave a TODO here, then.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java:
##########
@@ -135,6 +136,26 @@ public FileChannel openReadFileChannel() throws IOException {
}
}
+ public WALInputStream openReadFileStream() throws IOException {
+ if (isInSealedFile()) {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } else {
+ try {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } catch (IOException e) {
+ // unsealed file may be renamed after sealed, so we should try again
+ if (isInSealedFile()) {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } else {
+ throw e;
+ }
+ }
+ }
Review Comment:
Just leave a TODO here, then.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1636184029
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -535,12 +535,13 @@ public void run() {
memTableIdsOfWal
.computeIfAbsent(currentWALFileVersion, memTableIds -> new HashSet<>())
.addAll(info.metaData.getMemTablesId());
- checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage);
+ checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage, compressionRate);
boolean forceSuccess = false;
// try to roll log writer
if (info.rollWALFileWriterListener != null
- || (forceFlag && currentWALFileWriter.size() >= config.getWalFileSizeThresholdInByte())) {
+ || (forceFlag
+ && currentWALFileWriter.originalSize() >= config.getWalFileSizeThresholdInByte())) {
Review Comment:
Ok, then, please leave a TODO here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1636194330
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java:
##########
@@ -100,11 +101,11 @@ ByteBuffer read() throws IOException {
if (!canRead()) {
throw new IOException("Target file hasn't been specified.");
}
- try (FileChannel channel = openReadFileChannel()) {
+ try (WALInputStream is = openReadFileStream()) {
+ is.skipToGivenPosition(position);
ByteBuffer buffer = ByteBuffer.allocate(size);
- channel.position(position);
- channel.read(buffer);
- buffer.clear();
+ is.read(buffer);
+ buffer.flip();
return buffer;
}
Review Comment:
Just leave a TODO here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2132088045
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1642261996
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java:
##########
@@ -135,6 +136,26 @@ public FileChannel openReadFileChannel() throws IOException {
}
}
+ public WALInputStream openReadFileStream() throws IOException {
+ if (isInSealedFile()) {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } else {
+ try {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } catch (IOException e) {
+ // unsealed file may be renamed after sealed, so we should try again
+ if (isInSealedFile()) {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } else {
+ throw e;
+ }
+ }
+ }
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2131228057
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2105628782
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
 [B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1636185249
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java:
##########
@@ -64,8 +67,12 @@ public boolean hasNext() {
public ByteBuffer next() throws IOException {
int size = sizeIterator.next();
ByteBuffer buffer = ByteBuffer.allocate(size);
Review Comment:
Just adding a TODO here is enough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2156379992
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1621686309
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java:
##########
@@ -43,23 +45,75 @@ public abstract class LogWriter implements ILogWriter {
protected final File logFile;
protected final FileOutputStream logStream;
protected final FileChannel logChannel;
- protected long size;
+ protected long size = 0;
+ protected long originalSize = 0;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES * 2 + 1);
+ private static final CompressionType compressionAlg =
+ IoTDBDescriptor.getInstance().getConfig().getWALCompressionAlgorithm();
+ private final ICompressor compressor = ICompressor.getCompressor(compressionAlg);
+ private final ByteBuffer compressedByteBuffer;
+ // Minimum size to compress, default is 32 KB
+ private static final long MIN_COMPRESS_SIZE = 0;
Review Comment:
It does not look like 32KB.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java:
##########
@@ -64,8 +67,12 @@ public boolean hasNext() {
public ByteBuffer next() throws IOException {
int size = sizeIterator.next();
ByteBuffer buffer = ByteBuffer.allocate(size);
Review Comment:
Maybe the buffer can be reused (e.g., make it a parameter of the function and use it if its size fits) to reduce memory allocation.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java:
##########
@@ -75,6 +76,9 @@ protected IWALNode createWALNode(String identifier, String folder) {
} catch (FileNotFoundException e) {
logger.error("Fail to create wal node", e);
return WALFakeNode.getFailureInstance(e);
+ } catch (IOException e) {
+ logger.error("Meet exception when creating wal node", e);
+ return WALFakeNode.getFailureInstance(e);
}
Review Comment:
FileNotFoundException is also an IOException, and the two branches do not seem so different to me.
Maybe you can merge them into one.
The same for below.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
Review Comment:
Better to provide string encoding explicitly in all places.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
+ }
+
+ @Override
+ public int read() throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ return dataBuffer.get() & 0xFF;
Review Comment:
What is this for?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java:
##########
@@ -345,12 +344,13 @@ public long getFirstValidWALVersionId() {
}
/** Update wal disk cost of active memTables. */
- public void updateCostOfActiveMemTables(Map<Long, Long> memTableId2WalDiskUsage) {
+ public void updateCostOfActiveMemTables(
+ Map<Long, Long> memTableId2WalDiskUsage, double compressionRate) {
for (Map.Entry<Long, Long> memTableWalUsage : memTableId2WalDiskUsage.entrySet()) {
memTableId2Info.computeIfPresent(
memTableWalUsage.getKey(),
(k, v) -> {
- v.addWalDiskUsage(memTableWalUsage.getValue());
+ v.addWalDiskUsage((long) (memTableWalUsage.getValue() * compressionRate));
Review Comment:
Is it challenging to acquire the actual disk usage?
The compression ratio is already calculated based on that, and I wonder why you calculated it again (and resulted in a not-so-precise value).
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -521,8 +520,9 @@ public void run() {
forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
// flush buffer to os
+ double compressionRate = 1.0;
Review Comment:
Beware of the difference between "rate" and "ratio".
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -535,12 +535,13 @@ public void run() {
memTableIdsOfWal
.computeIfAbsent(currentWALFileVersion, memTableIds -> new HashSet<>())
.addAll(info.metaData.getMemTablesId());
- checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage);
+ checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage, compressionRate);
boolean forceSuccess = false;
// try to roll log writer
if (info.rollWALFileWriterListener != null
- || (forceFlag && currentWALFileWriter.size() >= config.getWalFileSizeThresholdInByte())) {
+ || (forceFlag
+ && currentWALFileWriter.originalSize() >= config.getWalFileSizeThresholdInByte())) {
Review Comment:
Using the originalSize to control force may result in small actual IOs if the compression ratio is high.
I guess this is to control the risk of losing persistence.
It would be better to add a comment explaining why you should choose originalSize over actualSize here.
Ideally, I would expect some experiments to show using originalSize is better than actualSize, or the trade-off.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -521,8 +520,9 @@ public void run() {
forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
// flush buffer to os
+ double compressionRate = 1.0;
try {
- currentWALFileWriter.write(syncingBuffer, info.metaData);
+ compressionRate = currentWALFileWriter.write(syncingBuffer, info.metaData);
Review Comment:
Aside from the compression ratio, the compression throughput is another key metric that would better be recorded.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java:
##########
@@ -43,23 +45,75 @@ public abstract class LogWriter implements ILogWriter {
protected final File logFile;
protected final FileOutputStream logStream;
protected final FileChannel logChannel;
- protected long size;
+ protected long size = 0;
+ protected long originalSize = 0;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES * 2 + 1);
+ private static final CompressionType compressionAlg =
+ IoTDBDescriptor.getInstance().getConfig().getWALCompressionAlgorithm();
+ private final ICompressor compressor = ICompressor.getCompressor(compressionAlg);
+ private final ByteBuffer compressedByteBuffer;
+ // Minimum size to compress, default is 32 KB
+ private static final long MIN_COMPRESS_SIZE = 0;
- protected LogWriter(File logFile) throws FileNotFoundException {
+ protected LogWriter(File logFile) throws IOException {
this.logFile = logFile;
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()));
+ size += logChannel.position();
+ }
+ if (compressionAlg != CompressionType.UNCOMPRESSED) {
+ compressedByteBuffer =
+ ByteBuffer.allocate(
+ compressor.getMaxBytesForCompression(
+ IoTDBDescriptor.getInstance().getConfig().getWalBufferSize()));
Review Comment:
This is a pessimistic assumption. It is okay, but for the target situation where the compression algorithm works fine, this may waste some memory because the compressed buffer should always be much smaller than the original one.
Maybe the utilization of the buffer should be monitored, and we can resize it according to the utilization.
For example, if the buffer utilization is always under 30% for a long time, we can resize it to 40% of the current capacity, saving 60% of the memory.
Just leave a TODO here, if you might.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java:
##########
@@ -153,10 +157,20 @@ public static WALMetaData readFromWALFile(File logFile, FileChannel channel) thr
return metaData;
}
- private static String readTailMagic(FileChannel channel) throws IOException {
+ private static boolean isValidMagicString(FileChannel channel) throws IOException {
ByteBuffer magicStringBytes = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
channel.read(magicStringBytes, channel.size() - WALWriter.MAGIC_STRING_BYTES);
magicStringBytes.flip();
- return new String(magicStringBytes.array());
+ String magicString = new String(magicStringBytes.array());
+ return magicString.equals(WALWriter.MAGIC_STRING)
+ || magicString.startsWith(WALWriter.MAGIC_STRING_V1);
Review Comment:
Why startsWith for V1?
##########
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.storageengine.dataregion.wal.compression;
+
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
+import org.apache.iotdb.db.utils.constant.TestConstant;
+
+import org.junit.After;
+import org.junit.Before;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+
+public class WALCompressionTest {
+ private final File walFile =
+ new File(
+ TestConstant.BASE_OUTPUT_PATH.concat(
+ WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)));
+
+ @Before
+ public void setUp() throws IOException {
+ if (walFile.getParentFile().exists()) {
+ Files.delete(walFile.getParentFile().toPath());
+ }
+ Files.createDirectory(walFile.getParentFile().toPath());
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ if (walFile.getParentFile().exists()) {
+ Files.delete(walFile.getParentFile().toPath());
+ }
+ }
+
+ public void testSkipToGivenPosition() {}
Review Comment:
PLEASE complete the tests.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
+ }
+
+ @Override
+ public int read() throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ return dataBuffer.get() & 0xFF;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ if (dataBuffer.remaining() >= len) {
+ dataBuffer.get(b, off, len);
+ return len;
+ }
+ int toBeRead = len;
+ while (toBeRead > 0) {
+ int remaining = dataBuffer.remaining();
+ int bytesRead = Math.min(remaining, toBeRead);
+ dataBuffer.get(b, off, bytesRead);
+ off += bytesRead;
+ toBeRead -= bytesRead;
+ if (toBeRead > 0) {
+ loadNextSegment();
+ }
+ }
+ return len;
+ }
+
+ @Override
+ public void close() throws IOException {
+ channel.close();
+ dataBuffer = null;
+ }
+
+ @Override
+ public int available() throws IOException {
+ long size = (endOffset - channel.position());
+ if (!Objects.isNull(dataBuffer)) {
+ size += dataBuffer.limit() - dataBuffer.position();
+ }
+ return (int) size;
+ }
+
+ private void loadNextSegment() throws IOException {
+ if (channel.position() >= endOffset) {
+ throw new IOException("End of file");
+ }
+ if (version == FileVersion.V2) {
+ loadNextSegmentV2();
+ } else if (version == FileVersion.V1) {
+ loadNextSegmentV1();
+ } else {
+ tryLoadSegment();
+ }
+ }
+
+ private void loadNextSegmentV1() throws IOException {
+ // just read raw data as input
+ if (channel.position() >= fileSize) {
+ throw new IOException("Unexpected end of file");
+ }
+ if (Objects.isNull(dataBuffer)) {
+ // read 128 KB
+ dataBuffer = ByteBuffer.allocate(128 * 1024);
+ }
+ dataBuffer.clear();
+ channel.read(dataBuffer);
+ dataBuffer.flip();
+ }
+
+ private void loadNextSegmentV2() throws IOException {
+ headerBuffer.clear();
+ if (channel.read(headerBuffer) != Integer.BYTES + 1) {
+ throw new IOException("Unexpected end of file");
+ }
+ // compressionType originalSize compressedSize
+ headerBuffer.flip();
+ CompressionType compressionType = CompressionType.deserialize(headerBuffer.get());
+ int dataBufferSize = headerBuffer.getInt();
+ if (compressionType != CompressionType.UNCOMPRESSED) {
+ compressedHeader.clear();
+ if (channel.read(compressedHeader) != Integer.BYTES) {
+ throw new IOException("Unexpected end of file");
+ }
+ compressedHeader.flip();
+ int uncompressedSize = compressedHeader.getInt();
+ dataBuffer = ByteBuffer.allocateDirect(uncompressedSize);
+ ByteBuffer compressedData = ByteBuffer.allocateDirect(dataBufferSize);
Review Comment:
Make sure the direct buffers are properly cleaned.
If the previous dataBuffer is large enough, use it directly.
The same can be done to compressedData.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
+ }
+
+ @Override
+ public int read() throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ return dataBuffer.get() & 0xFF;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ if (dataBuffer.remaining() >= len) {
+ dataBuffer.get(b, off, len);
+ return len;
+ }
+ int toBeRead = len;
+ while (toBeRead > 0) {
+ int remaining = dataBuffer.remaining();
+ int bytesRead = Math.min(remaining, toBeRead);
+ dataBuffer.get(b, off, bytesRead);
+ off += bytesRead;
+ toBeRead -= bytesRead;
+ if (toBeRead > 0) {
+ loadNextSegment();
+ }
+ }
+ return len;
+ }
+
+ @Override
+ public void close() throws IOException {
+ channel.close();
+ dataBuffer = null;
+ }
+
+ @Override
+ public int available() throws IOException {
+ long size = (endOffset - channel.position());
+ if (!Objects.isNull(dataBuffer)) {
+ size += dataBuffer.limit() - dataBuffer.position();
+ }
+ return (int) size;
+ }
+
+ private void loadNextSegment() throws IOException {
+ if (channel.position() >= endOffset) {
+ throw new IOException("End of file");
+ }
+ if (version == FileVersion.V2) {
+ loadNextSegmentV2();
+ } else if (version == FileVersion.V1) {
+ loadNextSegmentV1();
+ } else {
+ tryLoadSegment();
+ }
+ }
+
+ private void loadNextSegmentV1() throws IOException {
+ // just read raw data as input
+ if (channel.position() >= fileSize) {
+ throw new IOException("Unexpected end of file");
+ }
+ if (Objects.isNull(dataBuffer)) {
+ // read 128 KB
+ dataBuffer = ByteBuffer.allocate(128 * 1024);
+ }
+ dataBuffer.clear();
+ channel.read(dataBuffer);
+ dataBuffer.flip();
+ }
+
+ private void loadNextSegmentV2() throws IOException {
+ headerBuffer.clear();
+ if (channel.read(headerBuffer) != Integer.BYTES + 1) {
+ throw new IOException("Unexpected end of file");
+ }
+ // compressionType originalSize compressedSize
+ headerBuffer.flip();
+ CompressionType compressionType = CompressionType.deserialize(headerBuffer.get());
+ int dataBufferSize = headerBuffer.getInt();
+ if (compressionType != CompressionType.UNCOMPRESSED) {
+ compressedHeader.clear();
+ if (channel.read(compressedHeader) != Integer.BYTES) {
+ throw new IOException("Unexpected end of file");
+ }
+ compressedHeader.flip();
+ int uncompressedSize = compressedHeader.getInt();
+ dataBuffer = ByteBuffer.allocateDirect(uncompressedSize);
+ ByteBuffer compressedData = ByteBuffer.allocateDirect(dataBufferSize);
+ if (channel.read(compressedData) != dataBufferSize) {
+ throw new IOException("Unexpected end of file");
+ }
+ compressedData.flip();
+ IUnCompressor unCompressor = IUnCompressor.getUnCompressor(compressionType);
+ dataBuffer.clear();
+ unCompressor.uncompress(compressedData, dataBuffer);
+ } else {
+ dataBuffer = ByteBuffer.allocate(dataBufferSize);
+ if (channel.read(dataBuffer) != dataBufferSize) {
+ throw new IOException("Unexpected end of file");
+ }
+ }
+ dataBuffer.flip();
+ }
+
+ private void tryLoadSegment() throws IOException {
+ long originPosition = channel.position();
+ try {
+ loadNextSegmentV2();
+ version = FileVersion.V2;
+ } catch (Throwable e) {
+ // failed to load in V2 way, try in V1 way
+ logger.warn("Failed to load WAL segment in V2 way, try in V1 way", e);
+ channel.position(originPosition);
+ }
+
+ if (version == FileVersion.UNKNOWN) {
+ loadNextSegmentV1();
+ version = FileVersion.V1;
+ }
+ }
+
+ public void skipToGivenPosition(long pos) throws IOException {
+ if (version == FileVersion.V2) {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ ByteBuffer buffer = ByteBuffer.allocate(Byte.BYTES + Integer.BYTES);
+ long posRemain = pos;
+ int currSegmentSize = 0;
+ while (posRemain > 0) {
+ buffer.clear();
+ channel.read(buffer);
+ buffer.flip();
+ buffer.get();
+ currSegmentSize = buffer.getInt();
+ if (posRemain >= currSegmentSize) {
+ posRemain -= currSegmentSize;
+ } else {
+ break;
+ }
+ }
Review Comment:
Where is the segment skipped? I only see the header being read.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java:
##########
@@ -79,7 +84,7 @@ public int serializedSize() {
+ (memTablesId.isEmpty() ? 0 : Integer.BYTES + memTablesId.size() * Long.BYTES);
}
- public void serialize(ByteBuffer buffer) {
+ public void serialize(File file, ByteBuffer buffer) {
buffer.putLong(firstSearchIndex);
buffer.putInt(buffersSize.size());
for (int size : buffersSize) {
Review Comment:
Where is the new parameter used?
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java:
##########
@@ -135,6 +136,26 @@ public FileChannel openReadFileChannel() throws IOException {
}
}
+ public WALInputStream openReadFileStream() throws IOException {
+ if (isInSealedFile()) {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } else {
+ try {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } catch (IOException e) {
+ // unsealed file may be renamed after sealed, so we should try again
+ if (isInSealedFile()) {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } else {
+ throw e;
+ }
+ }
+ }
Review Comment:
It seems the "else" branch is just enough, and the branch above is not significantly different from it.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java:
##########
@@ -100,11 +101,11 @@ ByteBuffer read() throws IOException {
if (!canRead()) {
throw new IOException("Target file hasn't been specified.");
}
- try (FileChannel channel = openReadFileChannel()) {
+ try (WALInputStream is = openReadFileStream()) {
+ is.skipToGivenPosition(position);
ByteBuffer buffer = ByteBuffer.allocate(size);
- channel.position(position);
- channel.read(buffer);
- buffer.clear();
+ is.read(buffer);
+ buffer.flip();
return buffer;
}
Review Comment:
I wonder about the efficiency of opening a stream for each WALEntry. Maybe the stream should be reused somehow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1642257479
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java:
##########
@@ -64,8 +67,12 @@ public boolean hasNext() {
public ByteBuffer next() throws IOException {
int size = sizeIterator.next();
ByteBuffer buffer = ByteBuffer.allocate(size);
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2172503692
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2105634119
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
 [B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2142126835
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2141052888
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632188800
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -521,8 +520,9 @@ public void run() {
forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
// flush buffer to os
+ double compressionRate = 1.0;
try {
- currentWALFileWriter.write(syncingBuffer, info.metaData);
+ compressionRate = currentWALFileWriter.write(syncingBuffer, info.metaData);
Review Comment:
<img width="910" alt="image" src="https://github.com/apache/iotdb/assets/37140360/2011b00e-464e-4da6-a9d7-ffd1609e6b85">
Here we get the compression rate to update the WAL disk usage for each MemTable in subsequent updates. As for why we don't use the actual throughput, here are the reasons:
1. The original design only needed to know the approximate size of the WAL, not very precise data volume. I consulted the original maintainer of the WAL for this.
2. A WAL Buffer contains multiple WAL Entries for multiple MemTables, and we cannot know the exact size of each Entry after compression, unless we specifically design a compression algorithm to support this requirement. This is obviously stupid.
The current modification meets the original requirements and the changes are not large. Using precise I/O quantities would only increase the workload. Of course, I can add a TODO, and subsequent maintainers can do this if they want.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2162153346
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632194858
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALByteBufReader.java:
##########
@@ -64,8 +67,12 @@ public boolean hasNext() {
public ByteBuffer next() throws IOException {
int size = sizeIterator.next();
ByteBuffer buffer = ByteBuffer.allocate(size);
Review Comment:
This suggestion is reasonable, but implementing it would require converting the buffer from a temporary variable to a member variable. I'm not certain whether this modification would introduce additional bugs. Given that this is part of the original design and unrelated to the current task, I will not modify this section of the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2131239379
## [Codecov](https://app.codecov.io/gh/apache/iotdb/pull/12476?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `77.86561%` with `56 lines` in your changes are missing coverage. Please review.
> Project coverage is 43.43%. Comparing base [(`7ef1eb7`)](https://app.codecov.io/gh/apache/iotdb/commit/7ef1eb722e92c0287b6bd2ea513f39b00c0c9612?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`e7250f8`)](https://app.codecov.io/gh/apache/iotdb/commit/e7250f8e549361651c27e5923667798f07b5dd8f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 5 commits behind head on master.
| [Files](https://app.codecov.io/gh/apache/iotdb/pull/12476?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...torageengine/dataregion/wal/io/WALInputStream.java](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fstorageengine%2Fdataregion%2Fwal%2Fio%2FWALInputStream.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvc3RvcmFnZWVuZ2luZS9kYXRhcmVnaW9uL3dhbC9pby9XQUxJbnB1dFN0cmVhbS5qYXZh) | 74.69% | [41 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...eengine/dataregion/wal/utils/WALEntryPosition.java](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fstorageengine%2Fdataregion%2Fwal%2Futils%2FWALEntryPosition.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvc3RvcmFnZWVuZ2luZS9kYXRhcmVnaW9uL3dhbC91dGlscy9XQUxFbnRyeVBvc2l0aW9uLmphdmE=) | 50.00% | [7 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...wal/allocation/AbstractNodeAllocationStrategy.java](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fstorageengine%2Fdataregion%2Fwal%2Fallocation%2FAbstractNodeAllocationStrategy.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvc3RvcmFnZWVuZ2luZS9kYXRhcmVnaW9uL3dhbC9hbGxvY2F0aW9uL0Fic3RyYWN0Tm9kZUFsbG9jYXRpb25TdHJhdGVneS5qYXZh) | 0.00% | [6 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [.../db/storageengine/dataregion/wal/io/LogWriter.java](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fstorageengine%2Fdataregion%2Fwal%2Fio%2FLogWriter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvc3RvcmFnZWVuZ2luZS9kYXRhcmVnaW9uL3dhbC9pby9Mb2dXcml0ZXIuamF2YQ==) | 97.36% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
| [...b/storageengine/dataregion/wal/io/WALMetaData.java](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&filepath=iotdb-core%2Fdatanode%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fiotdb%2Fdb%2Fstorageengine%2Fdataregion%2Fwal%2Fio%2FWALMetaData.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-aW90ZGItY29yZS9kYXRhbm9kZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvaW90ZGIvZGIvc3RvcmFnZWVuZ2luZS9kYXRhcmVnaW9uL3dhbC9pby9XQUxNZXRhRGF0YS5qYXZh) | 80.00% | [1 Missing :warning: ](https://app.codecov.io/gh/apache/iotdb/pull/12476?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #12476 +/- ##
============================================
+ Coverage 43.38% 43.43% +0.05%
Complexity 69 69
============================================
Files 3164 3165 +1
Lines 196685 196931 +246
Branches 23426 23459 +33
============================================
+ Hits 85326 85541 +215
- Misses 111359 111390 +31
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/iotdb/pull/12476?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1625218082
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java:
##########
@@ -75,6 +76,9 @@ protected IWALNode createWALNode(String identifier, String folder) {
} catch (FileNotFoundException e) {
logger.error("Fail to create wal node", e);
return WALFakeNode.getFailureInstance(e);
+ } catch (IOException e) {
+ logger.error("Meet exception when creating wal node", e);
+ return WALFakeNode.getFailureInstance(e);
}
Review Comment:
Not asking why IOException is added.
I mean that the two exception branches are effectively the same, and they would better be merged into one to reduce redundant code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1625221331
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -535,12 +535,13 @@ public void run() {
memTableIdsOfWal
.computeIfAbsent(currentWALFileVersion, memTableIds -> new HashSet<>())
.addAll(info.metaData.getMemTablesId());
- checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage);
+ checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage, compressionRate);
boolean forceSuccess = false;
// try to roll log writer
if (info.rollWALFileWriterListener != null
- || (forceFlag && currentWALFileWriter.size() >= config.getWalFileSizeThresholdInByte())) {
+ || (forceFlag
+ && currentWALFileWriter.originalSize() >= config.getWalFileSizeThresholdInByte())) {
Review Comment:
So, in the end, you want to control the number of WALFlushListeners (and ultimately their corresponding memory).
Then, why not just use the number of WALFlushListeners (or their corresponding memory) as one of the predicates?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1630731281
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java:
##########
@@ -345,12 +344,13 @@ public long getFirstValidWALVersionId() {
}
/** Update wal disk cost of active memTables. */
- public void updateCostOfActiveMemTables(Map<Long, Long> memTableId2WalDiskUsage) {
+ public void updateCostOfActiveMemTables(
+ Map<Long, Long> memTableId2WalDiskUsage, double compressionRate) {
for (Map.Entry<Long, Long> memTableWalUsage : memTableId2WalDiskUsage.entrySet()) {
memTableId2Info.computeIfPresent(
memTableWalUsage.getKey(),
(k, v) -> {
- v.addWalDiskUsage(memTableWalUsage.getValue());
+ v.addWalDiskUsage((long) (memTableWalUsage.getValue() * compressionRate));
Review Comment:
First, in this scene, we **do not** need a very specific value to control it. Both methods will work, but this way is simple(in case of editting code).
Second, it's difficult to calculate the extractly I/O size of each memtable, because the every WALBuffer may contain WAL Entry from multiple memtables. After compression, we could not get the extractly size of each memtable occupied.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1630745431
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
+ }
+
+ @Override
+ public int read() throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ return dataBuffer.get() & 0xFF;
Review Comment:
<img width="671" alt="image" src="https://github.com/apache/iotdb/assets/37140360/3729413f-8d85-4d8f-824c-530add032829">
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1642261573
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java:
##########
@@ -100,11 +101,11 @@ ByteBuffer read() throws IOException {
if (!canRead()) {
throw new IOException("Target file hasn't been specified.");
}
- try (FileChannel channel = openReadFileChannel()) {
+ try (WALInputStream is = openReadFileStream()) {
+ is.skipToGivenPosition(position);
ByteBuffer buffer = ByteBuffer.allocate(size);
- channel.position(position);
- channel.read(buffer);
- buffer.clear();
+ is.read(buffer);
+ buffer.flip();
return buffer;
}
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1636186517
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
Review Comment:
We have already encountered bugs due to inconsistent default String encodings when recovering TsFileResource. The same issue should be avoided here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2160197201
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632197114
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java:
##########
@@ -79,7 +84,7 @@ public int serializedSize() {
+ (memTablesId.isEmpty() ? 0 : Integer.BYTES + memTablesId.size() * Long.BYTES);
}
- public void serialize(ByteBuffer buffer) {
+ public void serialize(File file, ByteBuffer buffer) {
buffer.putLong(firstSearchIndex);
buffer.putInt(buffersSize.size());
for (int size : buffersSize) {
Review Comment:
Removed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632197075
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
+ }
+
+ @Override
+ public int read() throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ return dataBuffer.get() & 0xFF;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) throws IOException {
+ if (Objects.isNull(dataBuffer) || dataBuffer.position() >= dataBuffer.limit()) {
+ loadNextSegment();
+ }
+ if (dataBuffer.remaining() >= len) {
+ dataBuffer.get(b, off, len);
+ return len;
+ }
+ int toBeRead = len;
+ while (toBeRead > 0) {
+ int remaining = dataBuffer.remaining();
+ int bytesRead = Math.min(remaining, toBeRead);
+ dataBuffer.get(b, off, bytesRead);
+ off += bytesRead;
+ toBeRead -= bytesRead;
+ if (toBeRead > 0) {
+ loadNextSegment();
+ }
+ }
+ return len;
+ }
+
+ @Override
+ public void close() throws IOException {
+ channel.close();
+ dataBuffer = null;
+ }
+
+ @Override
+ public int available() throws IOException {
+ long size = (endOffset - channel.position());
+ if (!Objects.isNull(dataBuffer)) {
+ size += dataBuffer.limit() - dataBuffer.position();
+ }
+ return (int) size;
+ }
+
+ private void loadNextSegment() throws IOException {
+ if (channel.position() >= endOffset) {
+ throw new IOException("End of file");
+ }
+ if (version == FileVersion.V2) {
+ loadNextSegmentV2();
+ } else if (version == FileVersion.V1) {
+ loadNextSegmentV1();
+ } else {
+ tryLoadSegment();
+ }
+ }
+
+ private void loadNextSegmentV1() throws IOException {
+ // just read raw data as input
+ if (channel.position() >= fileSize) {
+ throw new IOException("Unexpected end of file");
+ }
+ if (Objects.isNull(dataBuffer)) {
+ // read 128 KB
+ dataBuffer = ByteBuffer.allocate(128 * 1024);
+ }
+ dataBuffer.clear();
+ channel.read(dataBuffer);
+ dataBuffer.flip();
+ }
+
+ private void loadNextSegmentV2() throws IOException {
+ headerBuffer.clear();
+ if (channel.read(headerBuffer) != Integer.BYTES + 1) {
+ throw new IOException("Unexpected end of file");
+ }
+ // compressionType originalSize compressedSize
+ headerBuffer.flip();
+ CompressionType compressionType = CompressionType.deserialize(headerBuffer.get());
+ int dataBufferSize = headerBuffer.getInt();
+ if (compressionType != CompressionType.UNCOMPRESSED) {
+ compressedHeader.clear();
+ if (channel.read(compressedHeader) != Integer.BYTES) {
+ throw new IOException("Unexpected end of file");
+ }
+ compressedHeader.flip();
+ int uncompressedSize = compressedHeader.getInt();
+ dataBuffer = ByteBuffer.allocateDirect(uncompressedSize);
+ ByteBuffer compressedData = ByteBuffer.allocateDirect(dataBufferSize);
+ if (channel.read(compressedData) != dataBufferSize) {
+ throw new IOException("Unexpected end of file");
+ }
+ compressedData.flip();
+ IUnCompressor unCompressor = IUnCompressor.getUnCompressor(compressionType);
+ dataBuffer.clear();
+ unCompressor.uncompress(compressedData, dataBuffer);
+ } else {
+ dataBuffer = ByteBuffer.allocate(dataBufferSize);
+ if (channel.read(dataBuffer) != dataBufferSize) {
+ throw new IOException("Unexpected end of file");
+ }
+ }
+ dataBuffer.flip();
+ }
+
+ private void tryLoadSegment() throws IOException {
+ long originPosition = channel.position();
+ try {
+ loadNextSegmentV2();
+ version = FileVersion.V2;
+ } catch (Throwable e) {
+ // failed to load in V2 way, try in V1 way
+ logger.warn("Failed to load WAL segment in V2 way, try in V1 way", e);
+ channel.position(originPosition);
+ }
+
+ if (version == FileVersion.UNKNOWN) {
+ loadNextSegmentV1();
+ version = FileVersion.V1;
+ }
+ }
+
+ public void skipToGivenPosition(long pos) throws IOException {
+ if (version == FileVersion.V2) {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ ByteBuffer buffer = ByteBuffer.allocate(Byte.BYTES + Integer.BYTES);
+ long posRemain = pos;
+ int currSegmentSize = 0;
+ while (posRemain > 0) {
+ buffer.clear();
+ channel.read(buffer);
+ buffer.flip();
+ buffer.get();
+ currSegmentSize = buffer.getInt();
+ if (posRemain >= currSegmentSize) {
+ posRemain -= currSegmentSize;
+ } else {
+ break;
+ }
+ }
Review Comment:
Fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632197502
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
Review Comment:
What's the benefit?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632191314
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java:
##########
@@ -43,23 +45,75 @@ public abstract class LogWriter implements ILogWriter {
protected final File logFile;
protected final FileOutputStream logStream;
protected final FileChannel logChannel;
- protected long size;
+ protected long size = 0;
+ protected long originalSize = 0;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES * 2 + 1);
+ private static final CompressionType compressionAlg =
+ IoTDBDescriptor.getInstance().getConfig().getWALCompressionAlgorithm();
+ private final ICompressor compressor = ICompressor.getCompressor(compressionAlg);
+ private final ByteBuffer compressedByteBuffer;
+ // Minimum size to compress, default is 32 KB
+ private static final long MIN_COMPRESS_SIZE = 0;
Review Comment:
Fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632199051
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java:
##########
@@ -153,10 +157,20 @@ public static WALMetaData readFromWALFile(File logFile, FileChannel channel) thr
return metaData;
}
- private static String readTailMagic(FileChannel channel) throws IOException {
+ private static boolean isValidMagicString(FileChannel channel) throws IOException {
ByteBuffer magicStringBytes = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
channel.read(magicStringBytes, channel.size() - WALWriter.MAGIC_STRING_BYTES);
magicStringBytes.flip();
- return new String(magicStringBytes.array());
+ String magicString = new String(magicStringBytes.array());
+ return magicString.equals(WALWriter.MAGIC_STRING)
+ || magicString.startsWith(WALWriter.MAGIC_STRING_V1);
Review Comment:
I replace it with `contains`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632188800
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -521,8 +520,9 @@ public void run() {
forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
// flush buffer to os
+ double compressionRate = 1.0;
try {
- currentWALFileWriter.write(syncingBuffer, info.metaData);
+ compressionRate = currentWALFileWriter.write(syncingBuffer, info.metaData);
Review Comment:
<img width="910" alt="image" src="https://github.com/apache/iotdb/assets/37140360/2011b00e-464e-4da6-a9d7-ffd1609e6b85">
Here we get the compression rate to update the WAL disk usage for each MemTable in subsequent updates. As for why we don't use the actual throughput, here are the reasons:
1. The original design only needed to know the approximate size of the WAL, not very precise data volume. I consulted the original maintainer(@HeimingZ ) of the WAL for this.
2. A WAL Buffer contains multiple WAL Entries for multiple MemTables, and we cannot know the exact size of each Entry after compression, unless we specifically design a compression algorithm to support this requirement. This is obviously stupid.
The current modification meets the original requirements and the changes are not large. Using precise I/O quantities would only increase the workload. Of course, I can add a TODO, and subsequent maintainers can do this if they want.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2119128640
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
 [B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1642261060
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.storageengine.dataregion.wal.io;
+
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Objects;
+
+public class WALInputStream extends InputStream implements AutoCloseable {
+
+ private static final Logger logger = LoggerFactory.getLogger(WALInputStream.class);
+ private final FileChannel channel;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES + 1);
+ private final ByteBuffer compressedHeader = ByteBuffer.allocate(Integer.BYTES);
+ private ByteBuffer dataBuffer = null;
+ private long fileSize;
+ File logFile;
+ private long endOffset = -1;
+
+ enum FileVersion {
+ V1,
+ V2,
+ UNKNOWN
+ };
+
+ FileVersion version;
+
+ public WALInputStream(File logFile) throws IOException {
+ channel = FileChannel.open(logFile.toPath());
+ fileSize = channel.size();
+ analyzeFileVersion();
+ getEndOffset();
+ this.logFile = logFile;
+ }
+
+ private void getEndOffset() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES + Integer.BYTES) {
+ endOffset = channel.size();
+ return;
+ }
+ ByteBuffer metadataSizeBuf = ByteBuffer.allocate(Integer.BYTES);
+ long position;
+ try {
+ if (version == FileVersion.V2) {
+ 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)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES;
+ } else {
+ ByteBuffer magicStringBuffer =
+ ByteBuffer.allocate(WALWriter.MAGIC_STRING_V1.getBytes().length);
+ channel.read(
+ magicStringBuffer, channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length);
+ magicStringBuffer.flip();
+ if (!new String(magicStringBuffer.array()).equals(WALWriter.MAGIC_STRING_V1)) {
+ // this is a broken wal file
+ endOffset = channel.size();
+ return;
+ }
+ position = channel.size() - WALWriter.MAGIC_STRING_V1.getBytes().length - Integer.BYTES;
+ }
+ channel.read(metadataSizeBuf, position);
+ metadataSizeBuf.flip();
+ int metadataSize = metadataSizeBuf.getInt();
+ endOffset = channel.size() - WALWriter.MAGIC_STRING_BYTES - Integer.BYTES - metadataSize - 1;
+ } finally {
+ channel.position(WALWriter.MAGIC_STRING_BYTES);
+ }
+ }
+
+ private void analyzeFileVersion() throws IOException {
+ if (channel.size() < WALWriter.MAGIC_STRING_BYTES) {
+ version = FileVersion.UNKNOWN;
+ return;
+ }
+ if (isCurrentVersion()) {
+ this.version = FileVersion.V2;
+ return;
+ }
+ this.version = FileVersion.V1;
+ }
+
+ private boolean isCurrentVersion() throws IOException {
+ channel.position(0);
+ ByteBuffer buffer = ByteBuffer.allocate(WALWriter.MAGIC_STRING_BYTES);
+ channel.read(buffer);
+ return new String(buffer.array()).equals(WALWriter.MAGIC_STRING);
Review Comment:
Ok, I get what you mean now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1642263576
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -250,10 +250,14 @@ public void skipToGivenPosition(long pos) throws IOException {
buffer.clear();
channel.read(buffer);
buffer.flip();
- buffer.get();
+ CompressionType type = CompressionType.deserialize(buffer.get());
currSegmentSize = buffer.getInt();
if (posRemain >= currSegmentSize) {
posRemain -= currSegmentSize;
+ channel.position(
+ channel.position()
+ + currSegmentSize
+ + (type == CompressionType.UNCOMPRESSED ? 0 : Integer.BYTES));
Review Comment:
I add a comment for it, and rename the function.
##########
iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties:
##########
@@ -1215,6 +1215,10 @@ data_replication_factor=1
# Datatype: long
# iot_consensus_cache_window_time_in_ms=-1
+# Enable Write Ahead Log compression.
+# Option: true, false
+# enable_wal_compression=false
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1642273450
##########
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java:
##########
@@ -18,37 +18,392 @@
*/
package org.apache.iotdb.db.storageengine.dataregion.wal.compression;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
+import org.apache.iotdb.db.storageengine.dataregion.wal.WALTestUtils;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALBuffer;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALSignalEntry;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.LogWriter;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALByteBufReader;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALInputStream;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALReader;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
import org.apache.iotdb.db.utils.constant.TestConstant;
+import org.apache.commons.io.FileUtils;
+import org.apache.tsfile.compress.ICompressor;
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.utils.PublicBAOS;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
+import org.junit.Test;
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
public class WALCompressionTest {
private final File walFile =
new File(
TestConstant.BASE_OUTPUT_PATH.concat(
WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)));
+ private final String compressionDir =
+ TestConstant.OUTPUT_DATA_DIR.concat(File.separator + "wal-compression");
+
+ private final String devicePath = "root.sg.d1";
+ long originalMinCompressionSize;
+ CompressionType originCompressionType =
+ IoTDBDescriptor.getInstance().getConfig().getWALCompressionAlgorithm();
+
@Before
- public void setUp() throws IOException {
- if (walFile.getParentFile().exists()) {
- Files.delete(walFile.getParentFile().toPath());
+ public void setUp()
+ throws IOException, NoSuchFieldException, ClassNotFoundException, IllegalAccessException {
+ if (walFile.exists()) {
+ FileUtils.delete(walFile);
+ }
+ originalMinCompressionSize = WALTestUtils.getMinCompressionSize();
+ if (new File(compressionDir).exists()) {
+ FileUtils.forceDelete(new File(compressionDir));
}
- Files.createDirectory(walFile.getParentFile().toPath());
}
@After
- public void tearDown() throws IOException {
- if (walFile.getParentFile().exists()) {
- Files.delete(walFile.getParentFile().toPath());
+ public void tearDown()
+ throws IOException, NoSuchFieldException, ClassNotFoundException, IllegalAccessException {
+ if (walFile.exists()) {
+ FileUtils.delete(walFile);
+ }
+ if (new File(compressionDir).exists()) {
+ FileUtils.forceDelete(new File(compressionDir));
+ }
+ WALTestUtils.setMinCompressionSize(originalMinCompressionSize);
+ IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(originCompressionType);
+ }
+
+ @Test
+ public void testSkipToGivenPositionWithCompression()
+ throws NoSuchFieldException,
+ ClassNotFoundException,
+ IllegalAccessException,
+ QueryProcessException,
+ IllegalPathException,
+ IOException {
+ WALTestUtils.setMinCompressionSize(0L);
+ IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(CompressionType.LZ4);
+ testSkipToGivenPosition();
+ }
+
+ @Test
+ public void testSkipToGivenPositionWithoutCompression()
+ throws NoSuchFieldException,
+ ClassNotFoundException,
+ IllegalAccessException,
+ QueryProcessException,
+ IllegalPathException,
+ IOException {
+ WALTestUtils.setMinCompressionSize(1024 * 32);
+ testSkipToGivenPosition();
+ }
+
+ public void testSkipToGivenPosition()
+ throws QueryProcessException, IllegalPathException, IOException {
+ LogWriter writer = new WALWriter(walFile);
+ ByteBuffer buffer = ByteBuffer.allocate(1024 * 4);
+ List<Pair<Long, InsertRowNode>> positionAndEntryPairList = new ArrayList<>();
+ int memTableId = 0;
+ long fileOffset = 0;
+ for (int i = 0; i < 100; ) {
+ InsertRowNode insertRowNode = WALTestUtils.getInsertRowNode(devicePath + memTableId, i);
+ long serializedSize = insertRowNode.serializedSize();
+ if (buffer.remaining() >= serializedSize) {
+ int pos = buffer.position();
+ insertRowNode.serialize(buffer);
+ positionAndEntryPairList.add(new Pair<>(fileOffset, insertRowNode));
+ fileOffset += buffer.position() - pos;
+ i++;
+ } else {
+ writer.write(buffer);
+ buffer.clear();
+ }
+ }
+ if (buffer.position() != 0) {
+ writer.write(buffer);
+ }
+ writer.close();
+ try (WALInputStream stream = new WALInputStream(walFile)) {
+ for (int i = 0; i < 100; ++i) {
+ Pair<Long, InsertRowNode> positionAndNodePair = positionAndEntryPairList.get(i);
+ stream.skipToGivenPosition(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.
+ I don't know if this is a bug or not.
+ */
+ ByteBuffer nodeBuffer1 =
+ ByteBuffer.allocate(positionAndNodePair.right.serializedSize() + 2);
+ stream.read(nodeBuffer1);
+ ByteBuffer nodeBuffer2 =
+ ByteBuffer.allocate(positionAndNodePair.right.serializedSize() + 2);
+ positionAndNodePair.right.serialize(nodeBuffer2);
+ nodeBuffer2.flip();
+ Assert.assertArrayEquals(nodeBuffer1.array(), nodeBuffer2.array());
+ }
+ }
+ }
+
+ @Test
+ public void testUncompressedWALStructure()
+ throws QueryProcessException, IllegalPathException, IOException {
+ PublicBAOS baos = new PublicBAOS();
+ DataOutputStream dataOutputStream = new DataOutputStream(baos);
+ List<InsertRowNode> insertRowNodes = new ArrayList<>();
+ for (int i = 0; i < 100; ++i) {
+ InsertRowNode node = WALTestUtils.getInsertRowNode(devicePath, i);
+ insertRowNodes.add(node);
+ node.serialize(dataOutputStream);
+ }
+ dataOutputStream.close();
+ ByteBuffer buf = ByteBuffer.wrap(baos.toByteArray());
+ // Do not compress it
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setWALCompressionAlgorithm(CompressionType.UNCOMPRESSED);
+ try (WALWriter writer = new WALWriter(walFile)) {
+ buf.position(buf.limit());
+ writer.write(buf);
+ }
+
+ try (DataInputStream dataInputStream =
+ new DataInputStream(new BufferedInputStream(Files.newInputStream(walFile.toPath())))) {
+ byte[] magicStringBytes = new byte[WALWriter.MAGIC_STRING_BYTES];
+ // head magic string
+ dataInputStream.readFully(magicStringBytes);
+ Assert.assertEquals(WALWriter.MAGIC_STRING, new String(magicStringBytes));
+ Assert.assertEquals(
+ CompressionType.UNCOMPRESSED, CompressionType.deserialize(dataInputStream.readByte()));
+ Assert.assertEquals(buf.array().length, dataInputStream.readInt());
+ ByteBuffer dataBuf = ByteBuffer.allocate(buf.array().length);
+ dataInputStream.readFully(dataBuf.array());
+ Assert.assertArrayEquals(buf.array(), dataBuf.array());
+ Assert.assertEquals(
+ new WALSignalEntry(WALEntryType.WAL_FILE_INFO_END_MARKER),
+ WALEntry.deserialize(dataInputStream));
+ ByteBuffer metadataBuf = ByteBuffer.allocate(12 + Integer.BYTES);
+ dataInputStream.readFully(metadataBuf.array());
+ // Tail magic string
+ dataInputStream.readFully(magicStringBytes);
+ Assert.assertEquals(WALWriter.MAGIC_STRING, new String(magicStringBytes));
+ }
+ }
+
+ @Test
+ public void testCompressedWALStructure()
+ throws IOException,
+ QueryProcessException,
+ IllegalPathException,
+ NoSuchFieldException,
+ ClassNotFoundException,
+ IllegalAccessException {
+ PublicBAOS baos = new PublicBAOS();
+ DataOutputStream dataOutputStream = new DataOutputStream(baos);
+ List<InsertRowNode> insertRowNodes = new ArrayList<>();
+ for (int i = 0; i < 100; ++i) {
+ InsertRowNode node = WALTestUtils.getInsertRowNode(devicePath, i);
+ insertRowNodes.add(node);
+ node.serialize(dataOutputStream);
+ }
+ dataOutputStream.close();
+ ByteBuffer buf = ByteBuffer.wrap(baos.toByteArray());
+ // Do not compress it
+ IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(CompressionType.LZ4);
Review Comment:
Fix it. The comment is wrong since it is copied from previous test case.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2129693351
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2133210377
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 merged PR #12476:
URL: https://github.com/apache/iotdb/pull/12476
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1622383815
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -535,12 +535,13 @@ public void run() {
memTableIdsOfWal
.computeIfAbsent(currentWALFileVersion, memTableIds -> new HashSet<>())
.addAll(info.metaData.getMemTablesId());
- checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage);
+ checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage, compressionRate);
boolean forceSuccess = false;
// try to roll log writer
if (info.rollWALFileWriterListener != null
- || (forceFlag && currentWALFileWriter.size() >= config.getWalFileSizeThresholdInByte())) {
+ || (forceFlag
+ && currentWALFileWriter.originalSize() >= config.getWalFileSizeThresholdInByte())) {
Review Comment:
It's @OneSizeFitsQuorum 's suggestion.
The main reason to control the writer with originalSize is that when flushing a WALEntry into a buffer, an instance of WALFlushListener will be created. These objects are released when the file writer is closed. If the compression ratio is high, and we are using the actual size to control the writer, a lot of WALFlushListener instance will be kept in memory and cause an OOM error. This happened in the unit-test, so we fix it in this way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1622367489
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java:
##########
@@ -221,7 +222,7 @@ public class IoTDBConfig {
private int walBufferQueueCapacity = 500;
/** Size threshold of each wal file. Unit: byte */
- private volatile long walFileSizeThresholdInByte = 30 * 1024 * 1024L;
+ private volatile long walFileSizeThresholdInByte = 30 * 1024L * 1024L;
Review Comment:
Recover.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java:
##########
@@ -79,6 +132,10 @@ public long size() {
return size;
}
+ public long oiginalSize() {
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2132163616
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1617517976
##########
iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties:
##########
@@ -763,6 +763,10 @@ data_replication_factor=1
# Datatype: long
# iot_consensus_cache_window_time_in_ms=-1
+# WAL compression algorithm
+# options: UNCOMPRESSED, SNAPPY, LZ4, GZIP, ZSTD
+# wal_compress_algorithm=UNCOMPRESSED
Review Comment:
May tune to decide the default algorithm.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1630720719
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -521,8 +520,9 @@ public void run() {
forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
// flush buffer to os
+ double compressionRate = 1.0;
Review Comment:
fix it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1625229114
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -250,10 +250,14 @@ public void skipToGivenPosition(long pos) throws IOException {
buffer.clear();
channel.read(buffer);
buffer.flip();
- buffer.get();
+ CompressionType type = CompressionType.deserialize(buffer.get());
currSegmentSize = buffer.getInt();
if (posRemain >= currSegmentSize) {
posRemain -= currSegmentSize;
+ channel.position(
+ channel.position()
+ + currSegmentSize
+ + (type == CompressionType.UNCOMPRESSED ? 0 : Integer.BYTES));
Review Comment:
You may skip with the channel by `currSegmentSize + Integer.BYTES` if the WAL is compressed, but `posRemain` is only reduced by `currSegmentSize`.
Is this inconsistency problematic?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2160316097
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1636200265
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -250,10 +250,14 @@ public void skipToGivenPosition(long pos) throws IOException {
buffer.clear();
channel.read(buffer);
buffer.flip();
- buffer.get();
+ CompressionType type = CompressionType.deserialize(buffer.get());
currSegmentSize = buffer.getInt();
if (posRemain >= currSegmentSize) {
posRemain -= currSegmentSize;
+ channel.position(
+ channel.position()
+ + currSegmentSize
+ + (type == CompressionType.UNCOMPRESSED ? 0 : Integer.BYTES));
Review Comment:
This should be clarified by comments or better variable names; try it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1642256536
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -535,12 +535,13 @@ public void run() {
memTableIdsOfWal
.computeIfAbsent(currentWALFileVersion, memTableIds -> new HashSet<>())
.addAll(info.metaData.getMemTablesId());
- checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage);
+ checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage, compressionRate);
boolean forceSuccess = false;
// try to roll log writer
if (info.rollWALFileWriterListener != null
- || (forceFlag && currentWALFileWriter.size() >= config.getWalFileSizeThresholdInByte())) {
+ || (forceFlag
+ && currentWALFileWriter.originalSize() >= config.getWalFileSizeThresholdInByte())) {
Review Comment:
Done
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java:
##########
@@ -43,23 +45,75 @@ public abstract class LogWriter implements ILogWriter {
protected final File logFile;
protected final FileOutputStream logStream;
protected final FileChannel logChannel;
- protected long size;
+ protected long size = 0;
+ protected long originalSize = 0;
+ private final ByteBuffer headerBuffer = ByteBuffer.allocate(Integer.BYTES * 2 + 1);
+ private static final CompressionType compressionAlg =
+ IoTDBDescriptor.getInstance().getConfig().getWALCompressionAlgorithm();
+ private final ICompressor compressor = ICompressor.getCompressor(compressionAlg);
+ private final ByteBuffer compressedByteBuffer;
+ // Minimum size to compress, default is 32 KB
+ private static final long MIN_COMPRESS_SIZE = 0;
- protected LogWriter(File logFile) throws FileNotFoundException {
+ protected LogWriter(File logFile) throws IOException {
this.logFile = logFile;
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()));
+ size += logChannel.position();
+ }
+ if (compressionAlg != CompressionType.UNCOMPRESSED) {
+ compressedByteBuffer =
+ ByteBuffer.allocate(
+ compressor.getMaxBytesForCompression(
+ IoTDBDescriptor.getInstance().getConfig().getWalBufferSize()));
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2097762996
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "Caideyipi (via GitHub)" <gi...@apache.org>.
Caideyipi commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1617517347
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java:
##########
@@ -221,7 +222,7 @@ public class IoTDBConfig {
private int walBufferQueueCapacity = 500;
/** Size threshold of each wal file. Unit: byte */
- private volatile long walFileSizeThresholdInByte = 30 * 1024 * 1024L;
+ private volatile long walFileSizeThresholdInByte = 30 * 1024L * 1024L;
Review Comment:
Why change this here?
##########
iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties:
##########
@@ -763,6 +763,10 @@ data_replication_factor=1
# Datatype: long
# iot_consensus_cache_window_time_in_ms=-1
+# WAL compression algorithm
+# options: UNCOMPRESSED, SNAPPY, LZ4, GZIP, ZSTD
+# wal_compress_algorithm=UNCOMPRESSED
Review Comment:
May tune to decide the default value.
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/LogWriter.java:
##########
@@ -79,6 +132,10 @@ public long size() {
return size;
}
+ public long oiginalSize() {
Review Comment:
May be replaced with originalSize
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2097720127
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2119125729
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1625219915
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -521,8 +520,9 @@ public void run() {
forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
// flush buffer to os
+ double compressionRate = 1.0;
try {
- currentWALFileWriter.write(syncingBuffer, info.metaData);
+ compressionRate = currentWALFileWriter.write(syncingBuffer, info.metaData);
Review Comment:
"metric" here does not refer to IoTDB metrics for monitoring, but exactly the "effective information of wal file".
The idea is that the compression throughout is as important as the compression ratio.
If you decide to only use one, then you should give a good reason.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2158435061
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "OneSizeFitsQuorum (via GitHub)" <gi...@apache.org>.
OneSizeFitsQuorum commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1634121432
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java:
##########
@@ -85,6 +86,9 @@ protected IWALNode createWALNode(
} catch (FileNotFoundException e) {
logger.error("Fail to create wal node", e);
return WALFakeNode.getFailureInstance(e);
+ } catch (IOException e) {
Review Comment:
Exception handling should be the same as above
##########
iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties:
##########
@@ -1215,6 +1215,10 @@ data_replication_factor=1
# Datatype: long
# iot_consensus_cache_window_time_in_ms=-1
+# Enable Write Ahead Log compression.
+# Option: true, false
+# enable_wal_compression=false
Review Comment:
enable hot reload?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1625225151
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/checkpoint/CheckpointManager.java:
##########
@@ -345,12 +344,13 @@ public long getFirstValidWALVersionId() {
}
/** Update wal disk cost of active memTables. */
- public void updateCostOfActiveMemTables(Map<Long, Long> memTableId2WalDiskUsage) {
+ public void updateCostOfActiveMemTables(
+ Map<Long, Long> memTableId2WalDiskUsage, double compressionRate) {
for (Map.Entry<Long, Long> memTableWalUsage : memTableId2WalDiskUsage.entrySet()) {
memTableId2Info.computeIfPresent(
memTableWalUsage.getKey(),
(k, v) -> {
- v.addWalDiskUsage(memTableWalUsage.getValue());
+ v.addWalDiskUsage((long) (memTableWalUsage.getValue() * compressionRate));
Review Comment:
I never mentioned the file size.
I assume "memTableWalUsage.getValue() * compressionRate" is the compressed size, or the actual IO amount.
However, this calculation may result in some precision loss ("b * (a/b)" is very likely not "b" in computers), and you can directly pass down the compressed size and use it. Why not use the compressed size directly?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632199186
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java:
##########
@@ -135,6 +136,26 @@ public FileChannel openReadFileChannel() throws IOException {
}
}
+ public WALInputStream openReadFileStream() throws IOException {
+ if (isInSealedFile()) {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } else {
+ try {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } catch (IOException e) {
+ // unsealed file may be renamed after sealed, so we should try again
+ if (isInSealedFile()) {
+ walFile = walNode.getWALFile(walFileVersionId);
+ return new WALInputStream(walFile);
+ } else {
+ throw e;
+ }
+ }
+ }
Review Comment:
Same as above, this is the original design.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1636636621
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java:
##########
@@ -85,6 +86,9 @@ protected IWALNode createWALNode(
} catch (FileNotFoundException e) {
logger.error("Fail to create wal node", e);
return WALFakeNode.getFailureInstance(e);
+ } catch (IOException e) {
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1630743958
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -250,10 +250,14 @@ public void skipToGivenPosition(long pos) throws IOException {
buffer.clear();
channel.read(buffer);
buffer.flip();
- buffer.get();
+ CompressionType type = CompressionType.deserialize(buffer.get());
currSegmentSize = buffer.getInt();
if (posRemain >= currSegmentSize) {
posRemain -= currSegmentSize;
+ channel.position(
+ channel.position()
+ + currSegmentSize
+ + (type == CompressionType.UNCOMPRESSED ? 0 : Integer.BYTES));
Review Comment:
No. posRemain is a logical size, not a physical size. It represents the offset of the WAL before compression, so you only need to subtract the originalSegmentSize.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2128622499
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
 [B Reliability Rating on New Code](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) (required ≥ A)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
##
 Catch issues before they fail your Quality Gate with our IDE extension  [SonarLint](https://www.sonarsource.com/products/sonarlint/features/connected-mode/?referrer=pull-request)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1622378330
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -521,8 +520,9 @@ public void run() {
forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
// flush buffer to os
+ double compressionRate = 1.0;
try {
- currentWALFileWriter.write(syncingBuffer, info.metaData);
+ compressionRate = currentWALFileWriter.write(syncingBuffer, info.metaData);
Review Comment:
The return value is not used for recording or metrics, it's used to calculate the effective information of wal file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1625221331
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -535,12 +535,13 @@ public void run() {
memTableIdsOfWal
.computeIfAbsent(currentWALFileVersion, memTableIds -> new HashSet<>())
.addAll(info.metaData.getMemTablesId());
- checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage);
+ checkpointManager.updateCostOfActiveMemTables(info.memTableId2WalDiskUsage, compressionRate);
boolean forceSuccess = false;
// try to roll log writer
if (info.rollWALFileWriterListener != null
- || (forceFlag && currentWALFileWriter.size() >= config.getWalFileSizeThresholdInByte())) {
+ || (forceFlag
+ && currentWALFileWriter.originalSize() >= config.getWalFileSizeThresholdInByte())) {
Review Comment:
So, in the end, you want to control the number of WALFlushListeners (and ultimately its corresponding memory).
Then, why not just use the number of WALFlushListeners (or their corresponding memory) as one of the predicates?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632188800
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -521,8 +520,9 @@ public void run() {
forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
// flush buffer to os
+ double compressionRate = 1.0;
try {
- currentWALFileWriter.write(syncingBuffer, info.metaData);
+ compressionRate = currentWALFileWriter.write(syncingBuffer, info.metaData);
Review Comment:
<img width="910" alt="image" src="https://github.com/apache/iotdb/assets/37140360/2011b00e-464e-4da6-a9d7-ffd1609e6b85">
Here we get the compression rate to update the WAL disk usage for each MemTable in subsequent updates. As for why we don't use the actual throughput, here are the reasons:
1. The original design only needed to know the approximate size of the WAL, not very precise data volume. I consulted the original maintainer(@HeimingZ ) of the WAL for this.
2. A WAL Buffer contains multiple WAL Entries for multiple MemTables, and we cannot know the exact size of each Entry after compression, unless we specifically design a compression algorithm to support this requirement. This is obviously stupid.
The current modification meets the original requirements and the changes are not large. Using precise I/O quantities would only increase the work. Of course, I can add a TODO, and subsequent maintainers can do this if they want.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632188800
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/buffer/WALBuffer.java:
##########
@@ -521,8 +520,9 @@ public void run() {
forceFlag, syncingBuffer.position(), syncingBuffer.capacity(), usedRatio * 100);
// flush buffer to os
+ double compressionRate = 1.0;
try {
- currentWALFileWriter.write(syncingBuffer, info.metaData);
+ compressionRate = currentWALFileWriter.write(syncingBuffer, info.metaData);
Review Comment:
<img width="910" alt="image" src="https://github.com/apache/iotdb/assets/37140360/2011b00e-464e-4da6-a9d7-ffd1609e6b85">
Here we get the compression rate to update the WAL disk usage for each MemTable in subsequent updates. As for why we don't use the actual throughput, here are the reasons:
1. The original design only needed to know the approximate size of the WAL, not very precise data volume. I consulted the original maintainer of the WAL(@HeimingZ ) for this.
2. A WAL Buffer contains multiple WAL Entries for multiple MemTables, and we cannot know the exact size of each Entry after compression, unless we specifically design a compression algorithm to support this requirement. This is obviously stupid.
The current modification meets the original requirements and the changes are not large. Using precise I/O quantities would only increase the work. Of course, I can add a TODO, and subsequent maintainers can do this if they want.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1632198623
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/utils/WALEntryPosition.java:
##########
@@ -100,11 +101,11 @@ ByteBuffer read() throws IOException {
if (!canRead()) {
throw new IOException("Target file hasn't been specified.");
}
- try (FileChannel channel = openReadFileChannel()) {
+ try (WALInputStream is = openReadFileStream()) {
+ is.skipToGivenPosition(position);
ByteBuffer buffer = ByteBuffer.allocate(size);
- channel.position(position);
- channel.read(buffer);
- buffer.clear();
+ is.read(buffer);
+ buffer.flip();
return buffer;
}
Review Comment:
My response is the same as in the previous comments: this is how the original design was implemented. I just made slight modifications to make it suitable for compressed cases. Theoretically, the performance is similar to the original since the original design involved opening a Channel, and a WALInputStream essentially operates as a Channel. Sharing the Stream is a reasonable suggestion, but it is unrelated to the current task (Write Ahead Log compression) and may introduce other bugs and performance issues, so I will not be modifying this part of the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#issuecomment-2160172454
## [](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476) **Quality Gate failed**
Failed conditions
 [0.0% Coverage on New Code](https://sonarcloud.io/component_measures?id=apache_iotdb&pullRequest=12476&metric=new_coverage&view=list) (required ≥ 80%)
[See analysis details on SonarCloud](https://sonarcloud.io/dashboard?id=apache_iotdb&pullRequest=12476)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "jt2594838 (via GitHub)" <gi...@apache.org>.
jt2594838 commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1636207336
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java:
##########
@@ -190,33 +204,43 @@ private void loadNextSegmentV1() throws IOException {
}
private void loadNextSegmentV2() throws IOException {
- headerBuffer.clear();
- if (channel.read(headerBuffer) != Integer.BYTES + 1) {
- throw new IOException("Unexpected end of file");
- }
- // compressionType originalSize compressedSize
- headerBuffer.flip();
- CompressionType compressionType = CompressionType.deserialize(headerBuffer.get());
- int dataBufferSize = headerBuffer.getInt();
- if (compressionType != CompressionType.UNCOMPRESSED) {
- compressedHeader.clear();
- if (channel.read(compressedHeader) != Integer.BYTES) {
- throw new IOException("Unexpected end of file");
+ SegmentInfo segmentInfo = getNextSegmentInfo();
+ if (segmentInfo.compressionType != CompressionType.UNCOMPRESSED) {
+ // A compressed segment
+ if (Objects.isNull(dataBuffer)
+ || dataBuffer.capacity() < segmentInfo.uncompressedSize
+ || dataBuffer.capacity() > segmentInfo.uncompressedSize * 2) {
+ dataBuffer = ByteBuffer.allocateDirect(segmentInfo.uncompressedSize);
Review Comment:
If the previous buffer is not null, it should be manually released through `MmapUtil.clean`—the same for the below.
##########
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/wal/compression/WALCompressionTest.java:
##########
@@ -18,37 +18,392 @@
*/
package org.apache.iotdb.db.storageengine.dataregion.wal.compression;
+import org.apache.iotdb.commons.exception.IllegalPathException;
+import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.exception.query.QueryProcessException;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
+import org.apache.iotdb.db.storageengine.dataregion.wal.WALTestUtils;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALBuffer;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALSignalEntry;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.LogWriter;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALByteBufReader;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALInputStream;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALReader;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus;
import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
import org.apache.iotdb.db.utils.constant.TestConstant;
+import org.apache.commons.io.FileUtils;
+import org.apache.tsfile.compress.ICompressor;
+import org.apache.tsfile.compress.IUnCompressor;
+import org.apache.tsfile.file.metadata.enums.CompressionType;
+import org.apache.tsfile.utils.Pair;
+import org.apache.tsfile.utils.PublicBAOS;
import org.junit.After;
+import org.junit.Assert;
import org.junit.Before;
+import org.junit.Test;
+import java.io.BufferedInputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
public class WALCompressionTest {
private final File walFile =
new File(
TestConstant.BASE_OUTPUT_PATH.concat(
WALFileUtils.getLogFileName(0, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)));
+ private final String compressionDir =
+ TestConstant.OUTPUT_DATA_DIR.concat(File.separator + "wal-compression");
+
+ private final String devicePath = "root.sg.d1";
+ long originalMinCompressionSize;
+ CompressionType originCompressionType =
+ IoTDBDescriptor.getInstance().getConfig().getWALCompressionAlgorithm();
+
@Before
- public void setUp() throws IOException {
- if (walFile.getParentFile().exists()) {
- Files.delete(walFile.getParentFile().toPath());
+ public void setUp()
+ throws IOException, NoSuchFieldException, ClassNotFoundException, IllegalAccessException {
+ if (walFile.exists()) {
+ FileUtils.delete(walFile);
+ }
+ originalMinCompressionSize = WALTestUtils.getMinCompressionSize();
+ if (new File(compressionDir).exists()) {
+ FileUtils.forceDelete(new File(compressionDir));
}
- Files.createDirectory(walFile.getParentFile().toPath());
}
@After
- public void tearDown() throws IOException {
- if (walFile.getParentFile().exists()) {
- Files.delete(walFile.getParentFile().toPath());
+ public void tearDown()
+ throws IOException, NoSuchFieldException, ClassNotFoundException, IllegalAccessException {
+ if (walFile.exists()) {
+ FileUtils.delete(walFile);
+ }
+ if (new File(compressionDir).exists()) {
+ FileUtils.forceDelete(new File(compressionDir));
+ }
+ WALTestUtils.setMinCompressionSize(originalMinCompressionSize);
+ IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(originCompressionType);
+ }
+
+ @Test
+ public void testSkipToGivenPositionWithCompression()
+ throws NoSuchFieldException,
+ ClassNotFoundException,
+ IllegalAccessException,
+ QueryProcessException,
+ IllegalPathException,
+ IOException {
+ WALTestUtils.setMinCompressionSize(0L);
+ IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(CompressionType.LZ4);
+ testSkipToGivenPosition();
+ }
+
+ @Test
+ public void testSkipToGivenPositionWithoutCompression()
+ throws NoSuchFieldException,
+ ClassNotFoundException,
+ IllegalAccessException,
+ QueryProcessException,
+ IllegalPathException,
+ IOException {
+ WALTestUtils.setMinCompressionSize(1024 * 32);
+ testSkipToGivenPosition();
+ }
+
+ public void testSkipToGivenPosition()
+ throws QueryProcessException, IllegalPathException, IOException {
+ LogWriter writer = new WALWriter(walFile);
+ ByteBuffer buffer = ByteBuffer.allocate(1024 * 4);
+ List<Pair<Long, InsertRowNode>> positionAndEntryPairList = new ArrayList<>();
+ int memTableId = 0;
+ long fileOffset = 0;
+ for (int i = 0; i < 100; ) {
+ InsertRowNode insertRowNode = WALTestUtils.getInsertRowNode(devicePath + memTableId, i);
+ long serializedSize = insertRowNode.serializedSize();
+ if (buffer.remaining() >= serializedSize) {
+ int pos = buffer.position();
+ insertRowNode.serialize(buffer);
+ positionAndEntryPairList.add(new Pair<>(fileOffset, insertRowNode));
+ fileOffset += buffer.position() - pos;
+ i++;
+ } else {
+ writer.write(buffer);
+ buffer.clear();
+ }
+ }
+ if (buffer.position() != 0) {
+ writer.write(buffer);
+ }
+ writer.close();
+ try (WALInputStream stream = new WALInputStream(walFile)) {
+ for (int i = 0; i < 100; ++i) {
+ Pair<Long, InsertRowNode> positionAndNodePair = positionAndEntryPairList.get(i);
+ stream.skipToGivenPosition(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.
+ I don't know if this is a bug or not.
+ */
+ ByteBuffer nodeBuffer1 =
+ ByteBuffer.allocate(positionAndNodePair.right.serializedSize() + 2);
+ stream.read(nodeBuffer1);
+ ByteBuffer nodeBuffer2 =
+ ByteBuffer.allocate(positionAndNodePair.right.serializedSize() + 2);
+ positionAndNodePair.right.serialize(nodeBuffer2);
+ nodeBuffer2.flip();
+ Assert.assertArrayEquals(nodeBuffer1.array(), nodeBuffer2.array());
+ }
+ }
+ }
+
+ @Test
+ public void testUncompressedWALStructure()
+ throws QueryProcessException, IllegalPathException, IOException {
+ PublicBAOS baos = new PublicBAOS();
+ DataOutputStream dataOutputStream = new DataOutputStream(baos);
+ List<InsertRowNode> insertRowNodes = new ArrayList<>();
+ for (int i = 0; i < 100; ++i) {
+ InsertRowNode node = WALTestUtils.getInsertRowNode(devicePath, i);
+ insertRowNodes.add(node);
+ node.serialize(dataOutputStream);
+ }
+ dataOutputStream.close();
+ ByteBuffer buf = ByteBuffer.wrap(baos.toByteArray());
+ // Do not compress it
+ IoTDBDescriptor.getInstance()
+ .getConfig()
+ .setWALCompressionAlgorithm(CompressionType.UNCOMPRESSED);
+ try (WALWriter writer = new WALWriter(walFile)) {
+ buf.position(buf.limit());
+ writer.write(buf);
+ }
+
+ try (DataInputStream dataInputStream =
+ new DataInputStream(new BufferedInputStream(Files.newInputStream(walFile.toPath())))) {
+ byte[] magicStringBytes = new byte[WALWriter.MAGIC_STRING_BYTES];
+ // head magic string
+ dataInputStream.readFully(magicStringBytes);
+ Assert.assertEquals(WALWriter.MAGIC_STRING, new String(magicStringBytes));
+ Assert.assertEquals(
+ CompressionType.UNCOMPRESSED, CompressionType.deserialize(dataInputStream.readByte()));
+ Assert.assertEquals(buf.array().length, dataInputStream.readInt());
+ ByteBuffer dataBuf = ByteBuffer.allocate(buf.array().length);
+ dataInputStream.readFully(dataBuf.array());
+ Assert.assertArrayEquals(buf.array(), dataBuf.array());
+ Assert.assertEquals(
+ new WALSignalEntry(WALEntryType.WAL_FILE_INFO_END_MARKER),
+ WALEntry.deserialize(dataInputStream));
+ ByteBuffer metadataBuf = ByteBuffer.allocate(12 + Integer.BYTES);
+ dataInputStream.readFully(metadataBuf.array());
+ // Tail magic string
+ dataInputStream.readFully(magicStringBytes);
+ Assert.assertEquals(WALWriter.MAGIC_STRING, new String(magicStringBytes));
+ }
+ }
+
+ @Test
+ public void testCompressedWALStructure()
+ throws IOException,
+ QueryProcessException,
+ IllegalPathException,
+ NoSuchFieldException,
+ ClassNotFoundException,
+ IllegalAccessException {
+ PublicBAOS baos = new PublicBAOS();
+ DataOutputStream dataOutputStream = new DataOutputStream(baos);
+ List<InsertRowNode> insertRowNodes = new ArrayList<>();
+ for (int i = 0; i < 100; ++i) {
+ InsertRowNode node = WALTestUtils.getInsertRowNode(devicePath, i);
+ insertRowNodes.add(node);
+ node.serialize(dataOutputStream);
+ }
+ dataOutputStream.close();
+ ByteBuffer buf = ByteBuffer.wrap(baos.toByteArray());
+ // Do not compress it
+ IoTDBDescriptor.getInstance().getConfig().setWALCompressionAlgorithm(CompressionType.LZ4);
Review Comment:
Do not compress it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1630720277
##########
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/AbstractNodeAllocationStrategy.java:
##########
@@ -75,6 +76,9 @@ protected IWALNode createWALNode(String identifier, String folder) {
} catch (FileNotFoundException e) {
logger.error("Fail to create wal node", e);
return WALFakeNode.getFailureInstance(e);
+ } catch (IOException e) {
+ logger.error("Meet exception when creating wal node", e);
+ return WALFakeNode.getFailureInstance(e);
}
Review Comment:
Fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support WAL Compression [iotdb]
Posted by "THUMarkLau (via GitHub)" <gi...@apache.org>.
THUMarkLau commented on code in PR #12476:
URL: https://github.com/apache/iotdb/pull/12476#discussion_r1630719316
##########
iotdb-core/node-commons/src/assembly/resources/conf/iotdb-common.properties:
##########
@@ -763,6 +763,10 @@ data_replication_factor=1
# Datatype: long
# iot_consensus_cache_window_time_in_ms=-1
+# WAL compression algorithm
+# options: UNCOMPRESSED, SNAPPY, LZ4, GZIP, ZSTD
+# wal_compress_algorithm=UNCOMPRESSED
Review Comment:
Change the configuration item to enable or disable wal compression by true or false.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@iotdb.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org