Fix short reads in fixed-length deserialization (apache#17870)

* Fix short reads in fixed-length deserialization

* Fix tag log append EOF handling

* spotless

---------

Co-authored-by: Jiang Tian <jt2594838@163.com>

Author: 2 people

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/transformation/datastructure/SerializableList.java

@@ -22,6 +22,7 @@
 import org.apache.iotdb.calc.service.AbstractTemporaryQueryDataFileService;
 import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.commons.file.SystemFileFactory;
+import org.apache.iotdb.commons.utils.IOUtils;
 
 import org.apache.tsfile.utils.PublicBAOS;
 
@@ -60,7 +61,7 @@ default void deserialize() throws IOException {
     }
     init();
     ByteBuffer byteBuffer = ByteBuffer.allocate(recorder.getSerializedByteLength());
-    recorder.getFileChannel().read(byteBuffer);
+    IOUtils.readFully(recorder.getFileChannel(), byteBuffer);
     byteBuffer.flip();
     deserialize(byteBuffer);
     recorder.closeFile();

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/utils/sort/FileSpillerReader.java

@@ -21,6 +21,7 @@
 
 import org.apache.iotdb.calc.utils.datastructure.MergeSortKey;
 import org.apache.iotdb.commons.exception.IoTDBException;
+import org.apache.iotdb.commons.utils.IOUtils;
 import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.tsfile.common.conf.TSFileDescriptor;
@@ -93,10 +94,11 @@ private long read() throws IoTDBException {
       if (readLen == -1) {
         return -1;
       }
+      IOUtils.readFully(fileChannel, bytes);
       bytes.flip();
       int capacity = bytes.getInt();
       ByteBuffer tsBlockBytes = ByteBuffer.allocate(capacity);
-      fileChannel.read(tsBlockBytes);
+      IOUtils.readFully(fileChannel, tsBlockBytes);
       tsBlockBytes.flip();
       TsBlock cachedTsBlock = serde.deserialize(tsBlockBytes);
       cacheBlocks.add(cachedTsBlock);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/externalservice/ExternalServiceInfo.java

@@ -38,6 +38,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.File;
 import java.io.FileInputStream;
@@ -239,7 +240,7 @@ private ServiceInfo deserializeServiceInfoConsiderCRC(InputStream inputStream)
       throws IOException {
     int length = ReadWriteIOUtils.readInt(inputStream);
     byte[] bytes = new byte[length];
-    inputStream.read(bytes);
+    new DataInputStream(inputStream).readFully(bytes);
 
     crc32.reset();
     crc32.update(bytes, 0, length);

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

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.conf.CommonDescriptor;
 import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
 import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.commons.utils.IOUtils;
 import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.confignode.consensus.request.write.procedure.DeleteProcedurePlan;
 import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan;
@@ -165,8 +166,16 @@ private static Optional<Procedure> loadProcedure(Path procedureFilePath) {
     try (FileInputStream fis = new FileInputStream(procedureFilePath.toFile())) {
       Procedure procedure = null;
       try (FileChannel channel = fis.getChannel()) {
-        ByteBuffer byteBuffer = ByteBuffer.allocate(PROCEDURE_LOAD_BUFFER_SIZE);
-        if (channel.read(byteBuffer) > 0) {
+        final long fileSize = channel.size();
+        if (fileSize > PROCEDURE_LOAD_BUFFER_SIZE) {
+          throw new IOException(
+              String.format(
+                  "Procedure file %s exceeds the load buffer limit %s, actual size %s",
+                  procedureFilePath, PROCEDURE_LOAD_BUFFER_SIZE, fileSize));
+        }
+        ByteBuffer byteBuffer = ByteBuffer.allocate((int) fileSize);
+        if (fileSize > 0) {
+          IOUtils.readFully(channel, byteBuffer);
           byteBuffer.flip();
           procedure = ProcedureFactory.getInstance().create(byteBuffer);
           byteBuffer.clear();

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

@@ -76,10 +76,7 @@ public boolean hasNext() {
       }
       buffer = new byte[logSize];
 
-      int readLen = logStream.read(buffer, 0, logSize);
-      if (readLen < logSize) {
-        throw new IOException(ConfigNodeMessages.REACH_EOF);
-      }
+      logStream.readFully(buffer, 0, logSize);
 
       final long checkSum = logStream.readLong();
       checkSummer.reset();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/consensus/deletion/recover/DeletionReader.java

@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.db.pipe.consensus.deletion.recover;
 
+import org.apache.iotdb.commons.utils.IOUtils;
 import org.apache.iotdb.db.i18n.DataNodePipeMessages;
 import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResource;
 import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResourceManager;
@@ -60,7 +61,7 @@ public List<DeletionResource> readAllDeletions() throws IOException {
     try {
       // Read magic string
       ByteBuffer magicStringBuffer = ByteBuffer.allocate(MAGIC_STRING_BYTES_SIZE);
-      fileChannel.read(magicStringBuffer);
+      IOUtils.readFully(fileChannel, magicStringBuffer);
       magicStringBuffer.flip();
       String magicVersion = new String(magicStringBuffer.array(), StandardCharsets.UTF_8);
       if (LOGGER.isDebugEnabled()) {
@@ -70,7 +71,7 @@ public List<DeletionResource> readAllDeletions() throws IOException {
       // Read deletions
       long remainingBytes = fileChannel.size() - fileChannel.position();
       ByteBuffer byteBuffer = ByteBuffer.allocate((int) remainingBytes);
-      fileChannel.read(byteBuffer);
+      IOUtils.readFully(fileChannel, byteBuffer);
       byteBuffer.flip();
 
       List<DeletionResource> deletions = new ArrayList<>();

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/planner/plan/node/write/ObjectNode.java

@@ -158,7 +158,7 @@ public static ObjectNode deserializeFromWAL(ByteBuffer buffer) {
     if (objectFile.isPresent()) {
       try (RandomAccessFile raf = new RandomAccessFile(objectFile.get(), "r")) {
         raf.seek(offset);
-        raf.read(contents);
+        raf.readFully(contents);
       } catch (IOException e) {
         throw new RuntimeException(e);
       }
@@ -308,7 +308,7 @@ public ByteBuffer serialize() {
   private void readContentFromFile(File file, byte[] contents) throws IOException {
     try (RandomAccessFile raf = new RandomAccessFile(file, "r")) {
       raf.seek(offset);
-      raf.read(contents);
+      raf.readFully(contents);
     }
   }
 

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

@@ -30,7 +30,6 @@
 
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
-import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.ByteBuffer;
@@ -70,9 +69,7 @@ public T deserialize(InputStream inputStream) throws IOException {
     }
 
     byte[] logBuffer = new byte[logLength];
-    if (logLength < inputStream.read(logBuffer, 0, logLength)) {
-      throw new EOFException();
-    }
+    dataInputStream.readFully(logBuffer, 0, logLength);
 
     T result = deserializer.deserialize(ByteBuffer.wrap(logBuffer));
 

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

@@ -23,6 +23,7 @@
 import org.apache.iotdb.commons.schema.SchemaConstant;
 import org.apache.iotdb.commons.schema.node.role.IDatabaseMNode;
 import org.apache.iotdb.commons.schema.node.utils.IMNodeFactory;
+import org.apache.iotdb.commons.utils.IOUtils;
 import org.apache.iotdb.commons.utils.PathUtils;
 import org.apache.iotdb.commons.utils.TestOnly;
 import org.apache.iotdb.consensus.ConsensusFactory;
@@ -360,7 +361,7 @@ private void initFileHeader() throws IOException, MetadataException {
       lastSGAddr = 0L;
       pageManager = new BTreePageManager(channel, pmtFile, -1, logPath);
     } else {
-      channel.read(headerContent);
+      IOUtils.readFully(channel, headerContent);
       headerContent.clear();
       lastPageIndex = ReadWriteIOUtils.readInt(headerContent);
       dataTTL = ReadWriteIOUtils.readLong(headerContent);

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

@@ -27,6 +27,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.DataInputStream;
+import java.io.EOFException;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
@@ -91,8 +93,9 @@ public List<byte[]> collectUpdatedEntries() throws IOException, SchemaFileLogCor
         }
       }
 
-      // corrupted within one entry
-      if (inputStream.read(tempBytes, 1, tempBytes.length - 1) < tempBytes.length - 2) {
+      try {
+        new DataInputStream(inputStream).readFully(tempBytes, 1, tempBytes.length - 1);
+      } catch (EOFException e) {
         throw new SchemaFileLogCorruptedException(
             logFile.getAbsolutePath(), DataNodeSchemaMessages.SCHEMA_FILE_LOG_INCOMPLETE_ENTRY);
       }