You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ro...@apache.org on 2023/07/17 05:03:45 UTC
[iotdb] branch rel/1.1 updated: [IOTDB-6068] Pipe: Failed to load tsfile when the first page is empty in a chunk (#10556) (#10559) (#10560)
This is an automated email from the ASF dual-hosted git repository.
rong pushed a commit to branch rel/1.1
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/rel/1.1 by this push:
new 8b2290d2fd0 [IOTDB-6068] Pipe: Failed to load tsfile when the first page is empty in a chunk (#10556) (#10559) (#10560)
8b2290d2fd0 is described below
commit 8b2290d2fd0fdbdd6b14d6685a6fa21f37e8e5fe
Author: Itami Sho <42...@users.noreply.github.com>
AuthorDate: Mon Jul 17 13:03:39 2023 +0800
[IOTDB-6068] Pipe: Failed to load tsfile when the first page is empty in a chunk (#10556) (#10559) (#10560)
In previous pr: Github#10487, We only considered the case where the empty page is at the end of an aligned chunk, and it's not well thought out. Empty page is skipped, causing problems with rewriting tsfile on datanode.
In this pr, we consider all the position of empty pages in an aligned chunk, and handle the empty page through write an empty page into the chunk.
(cherry picked from commit f6a3cb2)
---
.../iotdb/db/engine/load/AlignedChunkData.java | 5 +++-
.../iotdb/db/engine/load/TsFileSplitter.java | 25 +++----------------
.../iotdb/tsfile/file/header/PageHeader.java | 3 +++
.../iotdb/tsfile/write/chunk/ValueChunkWriter.java | 29 ++++++++++++++--------
4 files changed, 30 insertions(+), 32 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/load/AlignedChunkData.java b/server/src/main/java/org/apache/iotdb/db/engine/load/AlignedChunkData.java
index e8cb2f20c1b..cdd40664655 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/load/AlignedChunkData.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/load/AlignedChunkData.java
@@ -179,8 +179,11 @@ public class AlignedChunkData implements ChunkData {
@Override
public void writeEntirePage(PageHeader pageHeader, ByteBuffer pageData) throws IOException {
pageNumbers.set(pageNumbers.size() - 1, pageNumbers.get(pageNumbers.size() - 1) + 1);
+ // serialize needDecode==false
dataSize += ReadWriteIOUtils.write(false, stream);
+ // serialize pageHeader
dataSize += pageHeader.serializeTo(stream);
+ // serialize pageData
dataSize += ReadWriteIOUtils.write(pageData, stream);
}
@@ -221,7 +224,7 @@ public class AlignedChunkData implements ChunkData {
break;
}
- if (values[i] == null) {
+ if (values.length == 0 || values[i] == null) {
dataSize += ReadWriteIOUtils.write(true, stream);
continue;
}
diff --git a/server/src/main/java/org/apache/iotdb/db/engine/load/TsFileSplitter.java b/server/src/main/java/org/apache/iotdb/db/engine/load/TsFileSplitter.java
index 8e5e0259cf2..c2b9dc19bc8 100644
--- a/server/src/main/java/org/apache/iotdb/db/engine/load/TsFileSplitter.java
+++ b/server/src/main/java/org/apache/iotdb/db/engine/load/TsFileSplitter.java
@@ -262,19 +262,15 @@ public class TsFileSplitter {
}
if (alignedChunkDataList.size() == 1) { // write entire page
// write the entire page if it's not an empty page.
- if (!isEmptyPage(pageHeader)) {
- alignedChunkDataList
- .get(0)
- .writeEntirePage(pageHeader, reader.readCompressedPage(pageHeader));
- }
+ alignedChunkDataList
+ .get(0)
+ .writeEntirePage(pageHeader, reader.readCompressedPage(pageHeader));
} else { // decode page
long[] times = pageIndex2Times.get(pageIndex);
TsPrimitiveType[] values =
decodeValuePage(reader, header, pageHeader, times, valueDecoder);
for (AlignedChunkData alignedChunkData : alignedChunkDataList) {
- if (!isEmptyPage(pageHeader)) {
- alignedChunkData.writeDecodeValuePage(times, values, header.getDataType());
- }
+ alignedChunkData.writeDecodeValuePage(times, values, header.getDataType());
}
}
@@ -446,19 +442,6 @@ public class TsFileSplitter {
}
}
- /**
- * handle empty page in aligned chunk, if uncompressedSize and compressedSize are both 0, and the
- * statistics is null, then the page is empty.
- *
- * @param pageHeader page header
- * @return true if the page is empty
- */
- private boolean isEmptyPage(PageHeader pageHeader) {
- return pageHeader.getUncompressedSize() == 0
- && pageHeader.getCompressedSize() == 0
- && pageHeader.getStatistics() == null;
- }
-
private TsPrimitiveType[] decodeValuePage(
TsFileSequenceReader reader,
ChunkHeader chunkHeader,
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/PageHeader.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/PageHeader.java
index 9ab5ea6f997..b17edea5245 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/PageHeader.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/file/header/PageHeader.java
@@ -120,6 +120,9 @@ public class PageHeader {
public int serializeTo(OutputStream outputStream) throws IOException {
int length = 0;
length += ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, outputStream);
+ if (uncompressedSize == 0) { // Empty Page
+ return length;
+ }
length += ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, outputStream);
length += statistics.serialize(outputStream);
return length;
diff --git a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ValueChunkWriter.java b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ValueChunkWriter.java
index fc2ee898003..248dd3bc3a6 100644
--- a/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ValueChunkWriter.java
+++ b/tsfile/src/main/java/org/apache/iotdb/tsfile/write/chunk/ValueChunkWriter.java
@@ -209,13 +209,16 @@ public class ValueChunkWriter {
"start to flush a page header into buffer, buffer position {} ", pageBuffer.size());
// serialize pageHeader see writePageToPageBuffer method
if (numOfPages == 0) { // record the firstPageStatistics
- if (header.getStatistics() != null) {
- this.firstPageStatistics = header.getStatistics();
- }
this.sizeWithoutStatistic +=
ReadWriteForEncodingUtils.writeUnsignedVarInt(header.getUncompressedSize(), pageBuffer);
- this.sizeWithoutStatistic +=
- ReadWriteForEncodingUtils.writeUnsignedVarInt(header.getCompressedSize(), pageBuffer);
+
+ if (header.getStatistics() == null) {
+ this.firstPageStatistics = null;
+ } else {
+ this.firstPageStatistics = header.getStatistics();
+ this.sizeWithoutStatistic +=
+ ReadWriteForEncodingUtils.writeUnsignedVarInt(header.getCompressedSize(), pageBuffer);
+ }
} else if (numOfPages == 1) { // put the firstPageStatistics into pageBuffer
if (firstPageStatistics != null) {
byte[] b = pageBuffer.toByteArray();
@@ -225,20 +228,26 @@ public class ValueChunkWriter {
pageBuffer.write(b, this.sizeWithoutStatistic, b.length - this.sizeWithoutStatistic);
}
ReadWriteForEncodingUtils.writeUnsignedVarInt(header.getUncompressedSize(), pageBuffer);
- ReadWriteForEncodingUtils.writeUnsignedVarInt(header.getCompressedSize(), pageBuffer);
- header.getStatistics().serialize(pageBuffer);
+ if (header.getUncompressedSize() != 0) {
+ ReadWriteForEncodingUtils.writeUnsignedVarInt(header.getCompressedSize(), pageBuffer);
+ header.getStatistics().serialize(pageBuffer);
+ }
firstPageStatistics = null;
} else {
ReadWriteForEncodingUtils.writeUnsignedVarInt(header.getUncompressedSize(), pageBuffer);
- ReadWriteForEncodingUtils.writeUnsignedVarInt(header.getCompressedSize(), pageBuffer);
- header.getStatistics().serialize(pageBuffer);
+ if (header.getUncompressedSize() != 0) {
+ ReadWriteForEncodingUtils.writeUnsignedVarInt(header.getCompressedSize(), pageBuffer);
+ header.getStatistics().serialize(pageBuffer);
+ }
}
logger.debug(
"finish to flush a page header {} of time page into buffer, buffer position {} ",
header,
pageBuffer.size());
- statistics.mergeStatistics(header.getStatistics());
+ if (header.getStatistics() != null) {
+ statistics.mergeStatistics(header.getStatistics());
+ }
} catch (IOException e) {
throw new PageException("IO Exception in writeDataPageHeader,ignore this page", e);