Prefetch Parquet footer to reduce the number of read calls
@@ -37,6 +37,7 @@ import parquet.schema.PrimitiveType.PrimitiveTypeName;
import parquet.schema.Type.Repetition;
import parquet.schema.Types;
+import java.io.ByteArrayInputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
@@ -85,11 +86,11 @@ public final class MetadataReader
validateParquet(fileSize >= MAGIC.length + PARQUET_METADATA_LENGTH + MAGIC.length, "%s is not a valid Parquet File", file);
long metadataLengthIndex = fileSize - PARQUET_METADATA_LENGTH - MAGIC.length;
- inputStream.seek(metadataLengthIndex);
- int metadataLength = readIntLittleEndian(inputStream);
+ InputStream footerStream = readFully(inputStream, metadataLengthIndex, PARQUET_METADATA_LENGTH + MAGIC.length);
+ int metadataLength = readIntLittleEndian(footerStream);
byte[] magic = new byte[MAGIC.length];
- inputStream.readFully(magic);
+ footerStream.read(magic);
validateParquet(Arrays.equals(MAGIC, magic), "Not valid Parquet file: %s expected magic number: %s got: %s", file, Arrays.toString(MAGIC), Arrays.toString(magic));
long metadataIndex = metadataLengthIndex - metadataLength;
@@ -98,8 +99,8 @@ public final class MetadataReader
"Corrupted Parquet file: %s metadata index: %s out of range",
file,
metadataIndex);
- inputStream.seek(metadataIndex);
- FileMetaData fileMetaData = readFileMetaData(inputStream);
+ InputStream metadataStream = readFully(inputStream, metadataIndex, metadataLength);
+ FileMetaData fileMetaData = readFileMetaData(metadataStream);
List<SchemaElement> schema = fileMetaData.getSchema();
validateParquet(!schema.isEmpty(), "Empty Parquet schema in file: %s", file);
@@ -300,4 +301,12 @@ public final class MetadataReader
}
return ((ch4 << 24) + (ch3 << 16) + (ch2 << 8) + (ch1));
}
+
+ private static InputStream readFully(FSDataInputStream from, long position, int length)
+ throws IOException
+ {
+ byte[] buffer = new byte[length];
+ from.readFully(position, buffer);
+ return new ByteArrayInputStream(buffer);
+ }
}
