Skip to content

Commit ea93b6f

Browse files
CaideyipiCaideyipi
committed
Fix short reads in fixed-length deserialization (#17870)
* Fix short reads in fixed-length deserialization * Fix tag log append EOF handling * spotless --------- (cherry picked from commit aa2acd6)
1 parent f0dc49e commit ea93b6f

23 files changed

Lines changed: 414 additions & 64 deletions

File tree

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/ProcedureInfo.java‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.iotdb.commons.conf.CommonDescriptor;
2424
import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
2525
import org.apache.iotdb.commons.utils.FileUtils;
26+
import org.apache.iotdb.commons.utils.IOUtils;
2627
import org.apache.iotdb.commons.utils.TestOnly;
2728
import org.apache.iotdb.confignode.consensus.request.write.procedure.DeleteProcedurePlan;
2829
import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan;
@@ -163,8 +164,16 @@ private static Optional<Procedure> loadProcedure(Path procedureFilePath) {
163164
try (FileInputStream fis = new FileInputStream(procedureFilePath.toFile())) {
164165
Procedure procedure = null;
165166
try (FileChannel channel = fis.getChannel()) {
166-
ByteBuffer byteBuffer = ByteBuffer.allocate(PROCEDURE_LOAD_BUFFER_SIZE);
167-
if (channel.read(byteBuffer) > 0) {
167+
final long fileSize = channel.size();
168+
if (fileSize > PROCEDURE_LOAD_BUFFER_SIZE) {
169+
throw new IOException(
170+
String.format(
171+
"Procedure file %s exceeds the load buffer limit %s, actual size %s",
172+
procedureFilePath, PROCEDURE_LOAD_BUFFER_SIZE, fileSize));
173+
}
174+
ByteBuffer byteBuffer = ByteBuffer.allocate((int) fileSize);
175+
if (fileSize > 0) {
176+
IOUtils.readFully(channel, byteBuffer);
168177
byteBuffer.flip();
169178
procedure = ProcedureFactory.getInstance().create(byteBuffer);
170179
byteBuffer.clear();

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/writelog/io/SingleFileLogReader.java‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -75,10 +75,7 @@ public boolean hasNext() {
7575
}
7676
buffer = new byte[logSize];
7777

78-
int readLen = logStream.read(buffer, 0, logSize);
79-
if (readLen < logSize) {
80-
throw new IOException("Reach eof");
81-
}
78+
logStream.readFully(buffer, 0, logSize);
8279

8380
final long checkSum = logStream.readLong();
8481
checkSummer.reset();

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/transformation/datastructure/SerializableList.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
package org.apache.iotdb.db.queryengine.transformation.datastructure;
2121

2222
import org.apache.iotdb.commons.file.SystemFileFactory;
23+
import org.apache.iotdb.commons.utils.IOUtils;
2324
import org.apache.iotdb.db.conf.IoTDBDescriptor;
2425
import org.apache.iotdb.db.service.TemporaryQueryDataFileService;
2526

@@ -60,7 +61,7 @@ default void deserialize() throws IOException {
6061
}
6162
init();
6263
ByteBuffer byteBuffer = ByteBuffer.allocate(recorder.getSerializedByteLength());
63-
recorder.getFileChannel().read(byteBuffer);
64+
IOUtils.readFully(recorder.getFileChannel(), byteBuffer);
6465
byteBuffer.flip();
6566
deserialize(byteBuffer);
6667
recorder.closeFile();

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/logfile/FakeCRC32Deserializer.java‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,6 @@
2727

2828
import java.io.ByteArrayInputStream;
2929
import java.io.DataInputStream;
30-
import java.io.EOFException;
3130
import java.io.IOException;
3231
import java.io.InputStream;
3332
import java.nio.ByteBuffer;
@@ -67,9 +66,7 @@ public T deserialize(InputStream inputStream) throws IOException {
6766
}
6867

6968
byte[] logBuffer = new byte[logLength];
70-
if (logLength < inputStream.read(logBuffer, 0, logLength)) {
71-
throw new EOFException();
72-
}
69+
dataInputStream.readFully(logBuffer, 0, logLength);
7370

7471
T result = deserializer.deserialize(ByteBuffer.wrap(logBuffer));
7572

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/SchemaFile.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.iotdb.commons.schema.SchemaConstant;
2424
import org.apache.iotdb.commons.schema.node.role.IDatabaseMNode;
2525
import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory;
26+
import org.apache.iotdb.commons.utils.IOUtils;
2627
import org.apache.iotdb.commons.utils.PathUtils;
2728
import org.apache.iotdb.commons.utils.TestOnly;
2829
import org.apache.iotdb.consensus.ConsensusFactory;
@@ -360,7 +361,7 @@ private void initFileHeader() throws IOException, MetadataException {
360361
lastSGAddr = 0L;
361362
pageManager = new BTreePageManager(channel, pmtFile, -1, logPath);
362363
} else {
363-
channel.read(headerContent);
364+
IOUtils.readFully(channel, headerContent);
364365
headerContent.clear();
365366
lastPageIndex = ReadWriteIOUtils.readInt(headerContent);
366367
dataTTL = ReadWriteIOUtils.readLong(headerContent);

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/log/SchemaFileLogReader.java‎

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@
2626
import org.slf4j.Logger;
2727
import org.slf4j.LoggerFactory;
2828

29+
import java.io.DataInputStream;
30+
import java.io.EOFException;
2931
import java.io.File;
3032
import java.io.FileInputStream;
3133
import java.io.IOException;
@@ -91,8 +93,9 @@ public List<byte[]> collectUpdatedEntries() throws IOException, SchemaFileLogCor
9193
}
9294
}
9395

94-
// corrupted within one entry
95-
if (inputStream.read(tempBytes, 1, tempBytes.length - 1) < tempBytes.length - 2) {
96+
try {
97+
new DataInputStream(inputStream).readFully(tempBytes, 1, tempBytes.length - 1);
98+
} catch (EOFException e) {
9699
throw new SchemaFileLogCorruptedException(logFile.getAbsolutePath(), "incomplete entry.");
97100
}
98101

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/mtree/impl/pbtree/schemafile/pagemgr/PageIOChannel.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
package org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.pagemgr;
2020

2121
import org.apache.iotdb.commons.exception.MetadataException;
22+
import org.apache.iotdb.commons.utils.IOUtils;
2223
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.ISchemaPage;
2324
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.SchemaFileConfig;
2425
import org.apache.iotdb.db.schemaengine.schemaregion.mtree.impl.pbtree.schemafile.log.SchemaFileLogReader;
@@ -107,7 +108,7 @@ public void loadFromFileToBuffer(ByteBuffer dst, int pageIndex) throws IOExcepti
107108
if (!readChannel.isOpen()) {
108109
readChannel = FileChannel.open(pmtFile.toPath(), StandardOpenOption.READ);
109110
}
110-
readChannel.read(dst, getPageAddress(pageIndex));
111+
IOUtils.readFully(readChannel, dst, getPageAddress(pageIndex));
111112
}
112113

113114
// region Flush Strategy

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/tag/TagLogFile.java‎

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.apache.iotdb.commons.conf.CommonDescriptor;
2323
import org.apache.iotdb.commons.exception.MetadataException;
2424
import org.apache.iotdb.commons.file.SystemFileFactory;
25+
import org.apache.iotdb.commons.utils.IOUtils;
2526
import org.apache.iotdb.db.conf.IoTDBDescriptor;
2627

2728
import org.apache.commons.io.FileUtils;
@@ -114,7 +115,7 @@ public static ByteBuffer parseByteBuffer(FileChannel fileChannel, long position)
114115
throws IOException {
115116
// Read the first block
116117
ByteBuffer byteBuffer = ByteBuffer.allocate(MAX_LENGTH);
117-
fileChannel.read(byteBuffer, position);
118+
IOUtils.readFully(fileChannel, byteBuffer, position);
118119
byteBuffer.flip();
119120
if (byteBuffer.limit() > 0) { // This indicates that there is data at this position
120121
int firstInt = ReadWriteIOUtils.readInt(byteBuffer); // first int
@@ -129,7 +130,7 @@ public static ByteBuffer parseByteBuffer(FileChannel fileChannel, long position)
129130
// read one offset, then use filechannel's read to read it
130131
byteBuffers.position(MAX_LENGTH * i);
131132
byteBuffers.limit(MAX_LENGTH * (i + 1));
132-
fileChannel.read(byteBuffers, nextPosition);
133+
IOUtils.readFully(fileChannel, byteBuffers, nextPosition);
133134
byteBuffers.position(4 + i * Long.BYTES);
134135
}
135136
byteBuffers.limit(byteBuffers.capacity());
@@ -144,7 +145,10 @@ private List<Long> parseOffsetList(long position) throws IOException {
144145
blockOffset.add(position);
145146
// Read the first block
146147
ByteBuffer byteBuffer = ByteBuffer.allocate(MAX_LENGTH);
147-
fileChannel.read(byteBuffer, position);
148+
if (position == fileChannel.size()) {
149+
return blockOffset;
150+
}
151+
IOUtils.readFully(fileChannel, byteBuffer, position);
148152
byteBuffer.flip();
149153
if (byteBuffer.limit() > 0) { // This indicates that there is data at this position
150154
int firstInt = ReadWriteIOUtils.readInt(byteBuffer); // first int
@@ -167,7 +171,7 @@ private List<Long> parseOffsetList(long position) throws IOException {
167171
// read
168172
blockBuffer.position(MAX_LENGTH * i);
169173
blockBuffer.limit(MAX_LENGTH * (i + 1));
170-
fileChannel.read(blockBuffer, blockOffset.get(i));
174+
IOUtils.readFully(fileChannel, blockBuffer, blockOffset.get(i));
171175
blockBuffer.position(4 + i * Long.BYTES);
172176
}
173177
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALFileVersion.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,8 @@
1818
*/
1919
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
2020

21+
import org.apache.iotdb.commons.utils.IOUtils;
22+
2123
import java.io.File;
2224
import java.io.IOException;
2325
import java.nio.ByteBuffer;
@@ -63,7 +65,7 @@ public static WALFileVersion getVersion(FileChannel channel) throws IOException
6365
continue;
6466
}
6567
ByteBuffer buffer = ByteBuffer.allocate(version.versionBytes.length);
66-
channel.read(buffer);
68+
IOUtils.readFully(channel, buffer);
6769
buffer.flip();
6870
String versionString = new String(buffer.array(), StandardCharsets.UTF_8);
6971
if (version.versionString.equals(versionString)) {

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALInputStream.java‎

Lines changed: 19 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
package org.apache.iotdb.db.storageengine.dataregion.wal.io;
2020

2121
import org.apache.iotdb.commons.conf.IoTDBConstant;
22+
import org.apache.iotdb.commons.utils.IOUtils;
2223
import org.apache.iotdb.db.service.metrics.WritingMetrics;
2324
import org.apache.iotdb.db.utils.MmapUtil;
2425

@@ -85,7 +86,8 @@ private void getEndOffset() throws IOException {
8586
if (version == WALFileVersion.V2) {
8687
// New Version
8788
ByteBuffer magicStringBuffer = ByteBuffer.allocate(version.getVersionBytes().length);
88-
channel.read(magicStringBuffer, channel.size() - version.getVersionBytes().length);
89+
IOUtils.readFully(
90+
channel, magicStringBuffer, channel.size() - version.getVersionBytes().length);
8991
magicStringBuffer.flip();
9092
if (logFile.getName().endsWith(IoTDBConstant.WAL_CHECKPOINT_FILE_SUFFIX)
9193
|| !new String(magicStringBuffer.array(), StandardCharsets.UTF_8)
@@ -105,7 +107,8 @@ private void getEndOffset() throws IOException {
105107
}
106108
// Old version
107109
ByteBuffer magicStringBuffer = ByteBuffer.allocate(version.getVersionBytes().length);
108-
channel.read(magicStringBuffer, channel.size() - version.getVersionBytes().length);
110+
IOUtils.readFully(
111+
channel, magicStringBuffer, channel.size() - version.getVersionBytes().length);
109112
magicStringBuffer.flip();
110113
if (!new String(magicStringBuffer.array(), StandardCharsets.UTF_8)
111114
.equals(version.getVersionString())) {
@@ -117,7 +120,7 @@ private void getEndOffset() throws IOException {
117120
}
118121
}
119122
// Read the metadata size
120-
channel.read(metadataSizeBuf, position);
123+
IOUtils.readFully(channel, metadataSizeBuf, position);
121124
metadataSizeBuf.flip();
122125
int metadataSize = metadataSizeBuf.getInt();
123126
endOffset = channel.size() - version.getVersionBytes().length - Integer.BYTES - metadataSize;
@@ -237,9 +240,7 @@ private void loadNextSegmentV2() throws IOException {
237240
compressedBuffer.clear();
238241
// limit the buffer to prevent it from reading too much byte than expected
239242
compressedBuffer.limit(segmentInfo.dataInDiskSize);
240-
if (readWALBufferFromChannel(compressedBuffer) != segmentInfo.dataInDiskSize) {
241-
throw new IOException("Unexpected end of file");
242-
}
243+
readWALBufferFullyFromChannel(compressedBuffer);
243244
compressedBuffer.flip();
244245
IUnCompressor unCompressor = IUnCompressor.getUnCompressor(segmentInfo.compressionType);
245246
uncompressWALBuffer(compressedBuffer, dataBuffer, unCompressor);
@@ -255,9 +256,7 @@ private void loadNextSegmentV2() throws IOException {
255256
// limit the buffer to prevent it from reading too much byte than expected
256257
dataBuffer.limit(segmentInfo.dataInDiskSize);
257258

258-
if (readWALBufferFromChannel(dataBuffer) != segmentInfo.dataInDiskSize) {
259-
throw new IOException("Unexpected end of file");
260-
}
259+
readWALBufferFullyFromChannel(dataBuffer);
261260
}
262261
dataBuffer.flip();
263262
}
@@ -301,7 +300,7 @@ public void skipToGivenLogicalPosition(long pos) throws IOException {
301300

302301
if (segmentInfo.compressionType != CompressionType.UNCOMPRESSED) {
303302
compressedBuffer = ByteBuffer.allocateDirect(segmentInfo.dataInDiskSize);
304-
readWALBufferFromChannel(compressedBuffer);
303+
readWALBufferFullyFromChannel(compressedBuffer);
305304
compressedBuffer.flip();
306305
IUnCompressor unCompressor = IUnCompressor.getUnCompressor(segmentInfo.compressionType);
307306
dataBuffer = ByteBuffer.allocateDirect(segmentInfo.uncompressedSize);
@@ -310,7 +309,7 @@ public void skipToGivenLogicalPosition(long pos) throws IOException {
310309
compressedBuffer = null;
311310
} else {
312311
dataBuffer = ByteBuffer.allocateDirect(segmentInfo.dataInDiskSize);
313-
readWALBufferFromChannel(dataBuffer);
312+
readWALBufferFullyFromChannel(dataBuffer);
314313
dataBuffer.flip();
315314
}
316315

@@ -349,15 +348,15 @@ public WALMetaData getWALMetaData() throws IOException {
349348

350349
private SegmentInfo getNextSegmentInfo() throws IOException {
351350
segmentHeaderWithoutCompressedSizeBuffer.clear();
352-
channel.read(segmentHeaderWithoutCompressedSizeBuffer);
351+
readWALBufferFullyFromChannel(segmentHeaderWithoutCompressedSizeBuffer);
353352
segmentHeaderWithoutCompressedSizeBuffer.flip();
354353
SegmentInfo info = new SegmentInfo();
355354
info.compressionType =
356355
CompressionType.deserialize(segmentHeaderWithoutCompressedSizeBuffer.get());
357356
info.dataInDiskSize = segmentHeaderWithoutCompressedSizeBuffer.getInt();
358357
if (info.compressionType != CompressionType.UNCOMPRESSED) {
359358
compressedSizeBuffer.clear();
360-
readWALBufferFromChannel(compressedSizeBuffer);
359+
readWALBufferFullyFromChannel(compressedSizeBuffer);
361360
compressedSizeBuffer.flip();
362361
info.uncompressedSize = compressedSizeBuffer.getInt();
363362
} else {
@@ -373,6 +372,13 @@ private int readWALBufferFromChannel(ByteBuffer buffer) throws IOException {
373372
return size;
374373
}
375374

375+
private void readWALBufferFullyFromChannel(ByteBuffer buffer) throws IOException {
376+
long startTime = System.nanoTime();
377+
int size = buffer.remaining();
378+
IOUtils.readFully(channel, buffer);
379+
WritingMetrics.getInstance().recordWALRead(size, System.nanoTime() - startTime);
380+
}
381+
376382
private void uncompressWALBuffer(
377383
ByteBuffer compressed, ByteBuffer uncompressed, IUnCompressor unCompressor)
378384
throws IOException {

0 commit comments

Comments
 (0)