Skip to content

Core: Add Eager fetch to parquet read path #17284

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
RussellSpitzer merged 11 commits into from
Sep 1, 2026
Merged

Core: Add Eager fetch to parquet read path #17284

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
75e0048
add eager support in parquet.java
varun-lakhyani Aug 14, 2026
c6b6861
checkstyle
varun-lakhyani Aug 14, 2026
4d33a4e
make HadoopInputFile implementing HadoopConfigurable
varun-lakhyani Aug 20, 2026
812d748
add EagerInputFileConfigurable support + use length instead of fileSi…
varun-lakhyani Aug 20, 2026
bcfe1d5
integrate Eager in Parquet read path
varun-lakhyani Aug 20, 2026
2cce30c
address comments - nits
varun-lakhyani Aug 24, 2026
b700403
add EOF check in EagerInputFile to catch file length > reported
varun-lakhyani Aug 29, 2026
3257b1d
Checkstyle - hasMessageContaining is req
varun-lakhyani Aug 29, 2026
8bd13b6
rename testStaleLengthThrows to testIncorrectLengthThrows to show bro…
varun-lakhyani Aug 29, 2026
b17b679
nit
varun-lakhyani Aug 29, 2026
5eb2cbc
nit make exception clear
varun-lakhyani Aug 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion core/src/main/java/org/apache/iceberg/hadoop/HadoopInputFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.function.Function;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
Expand All @@ -35,13 +36,14 @@
import org.apache.iceberg.io.SeekableInputStream;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.util.SerializableSupplier;

/**
* {@link InputFile} implementation using the Hadoop {@link FileSystem} API.
*
* <p>This class is based on Parquet's HadoopInputFile.
*/
public class HadoopInputFile implements InputFile, NativelyEncryptedFile {
public class HadoopInputFile implements InputFile, NativelyEncryptedFile, HadoopConfigurable {
public static final String[] NO_LOCATION_PREFERENCE = new String[0];

private final String location;
Expand Down Expand Up @@ -188,10 +190,17 @@ public SeekableInputStream newStream() {
}
}

@Override
public Configuration getConf() {
return conf;
}

@Override
public void serializeConfWith(
Function<Configuration, SerializableSupplier<Configuration>> confSerializer) {
throw new UnsupportedOperationException("Cannot serialize a Hadoop input file: " + location());
}

public FileSystem getFileSystem() {
return fs;
}
Expand Down
67 changes: 58 additions & 9 deletions core/src/main/java/org/apache/iceberg/io/EagerInputFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,12 @@
package org.apache.iceberg.io;

import java.io.IOException;
import java.util.function.Function;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.hadoop.HadoopConfigurable;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.util.SerializableSupplier;

/**
* A decorator that collapses multiple object-store requests into one by fetching the entire file
Expand All @@ -29,23 +33,31 @@
public class EagerInputFile implements InputFile {

private final InputFile delegate;
private final long fileSize;
private final long length;

public EagerInputFile(InputFile delegate, long fileSize) {
private EagerInputFile(InputFile delegate, long length) {
Preconditions.checkNotNull(delegate, "delegate is null");
Preconditions.checkArgument(fileSize >= 0, "fileSize is negative: %s", fileSize);
Preconditions.checkArgument(length >= 0, "length is negative: %s", length);
Preconditions.checkArgument(
fileSize <= Integer.MAX_VALUE,
"Cannot eagerly load file because fileSize exceeds eager loading capacity, consider reducing eager fetch threshold below %s bytes: %s",
length <= Integer.MAX_VALUE,
"Cannot eagerly load file because length exceeds eager loading capacity, consider reducing eager fetch threshold below %s bytes: %s",
Integer.MAX_VALUE,
fileSize);
length);
this.delegate = delegate;
this.fileSize = fileSize;
this.length = length;
}

/** Returns an InputFile that reads the entire file in a single request. */
public static InputFile of(InputFile delegate, long length) {
if (delegate instanceof HadoopConfigurable) {
return new EagerInputFileConfigurable(delegate, length);
}
return new EagerInputFile(delegate, length);
}

@Override
public long getLength() {
return fileSize;
return length;
}

@Override
Expand All @@ -60,12 +72,49 @@ public boolean exists() {

@Override
public SeekableInputStream newStream() {
byte[] bytes = new byte[(int) fileSize];
byte[] bytes = new byte[(int) length];
try (SeekableInputStream src = delegate.newStream()) {
IOUtil.readFully(src, bytes, 0, bytes.length);
Comment thread
danielcweeks marked this conversation as resolved.
// reads from the already open stream; no additional request
if (src.read() != -1) {
throw new IOException(
"Incorrect length provided for file "
+ delegate.location()
+ ", given a length of "
+ length
+ " and did not reach the end of stream");
}
} catch (IOException e) {
throw new RuntimeIOException(e, "Failed to fetch file: %s", delegate.location());
}
return new EagerInputStream(bytes);
}

/** An {@link EagerInputFile} that preserves the delegate's Hadoop configuration. */
private static class EagerInputFileConfigurable extends EagerInputFile
implements HadoopConfigurable {

private final HadoopConfigurable delegate;

EagerInputFileConfigurable(InputFile delegate, long length) {
super(delegate, length);
Preconditions.checkArgument(
delegate instanceof HadoopConfigurable,
"Cannot create Hadoop Configurable Eager Input File because %s does not implement HadoopConfigurable",
delegate.getClass().getName());
this.delegate = (HadoopConfigurable) delegate;
}

@Override
public Configuration getConf() {
return delegate.getConf();
}

@Override
public void serializeConfWith(
Function<Configuration, SerializableSupplier<Configuration>> confSerializer) {
throw new UnsupportedOperationException(
"Cannot serialize an eager input file: " + location());
}
}
}
58 changes: 46 additions & 12 deletions core/src/test/java/org/apache/iceberg/io/TestEagerInputFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,10 @@

import java.io.EOFException;
import java.util.Arrays;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.hadoop.HadoopConfigurable;
import org.apache.iceberg.hadoop.HadoopInputFile;
import org.apache.iceberg.inmemory.InMemoryInputFile;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
Expand All @@ -39,14 +43,14 @@ private static byte[] makeBytes() {
return data;
}

private static EagerInputFile eagerFile(byte[] bytes) {
return new EagerInputFile(new InMemoryInputFile(bytes), bytes.length);
private static InputFile eagerFile(byte[] bytes) {
return EagerInputFile.of(new InMemoryInputFile(bytes), bytes.length);
}

@Test
public void testNewStream() throws Exception {
InputFile delegate = Mockito.spy(new InMemoryInputFile(makeBytes()));
try (SeekableInputStream stream = new EagerInputFile(delegate, SIZE).newStream()) {
try (SeekableInputStream stream = EagerInputFile.of(delegate, SIZE).newStream()) {
assertThat(stream)
.as("newStream() should return an EagerInputStream")
.isInstanceOf(EagerInputStream.class);
Expand Down Expand Up @@ -169,14 +173,14 @@ public void testAvailable() throws Exception {
@Test
public void testGetLength() {
assertThat(eagerFile(makeBytes()).getLength())
.as("getLength() should return fileSize")
.as("getLength() should return the file length")
.isEqualTo(SIZE);
}

@Test
public void testLocation() {
InMemoryInputFile inner = new InMemoryInputFile(makeBytes());
assertThat(new EagerInputFile(inner, SIZE).location())
assertThat(EagerInputFile.of(inner, SIZE).location())
.as("location should pass through unchanged")
.isEqualTo(inner.location());
}
Expand All @@ -188,21 +192,20 @@ public void testExists() {

@Test
public void testNewFileThrows() {
assertThatThrownBy(() -> new EagerInputFile(null, SIZE))
assertThatThrownBy(() -> EagerInputFile.of(null, SIZE))
.as("Null delegate is not allowed")
.isInstanceOf(NullPointerException.class)
.hasMessageContaining("delegate is null");

assertThatThrownBy(() -> new EagerInputFile(new InMemoryInputFile(makeBytes()), -1))
.as("Negative fileSize is not allowed")
assertThatThrownBy(() -> EagerInputFile.of(new InMemoryInputFile(makeBytes()), -1))
.as("Negative length is not allowed")
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("fileSize is negative");
.hasMessageContaining("length is negative");

assertThatThrownBy(
() ->
new EagerInputFile(
new InMemoryInputFile(makeBytes()), (long) Integer.MAX_VALUE + 1))
.as("fileSize exceeds eager loading capacity")
EagerInputFile.of(new InMemoryInputFile(makeBytes()), (long) Integer.MAX_VALUE + 1))
.as("length exceeds eager loading capacity")
.isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("exceeds eager loading capacity");
}
Expand Down Expand Up @@ -231,4 +234,35 @@ public void testReadFullyThrows() throws Exception {
.hasMessageContaining("exceeds stream length");
}
}

@Test
public void testIncorrectLengthThrows() {
byte[] bytes = makeBytes();

assertThatThrownBy(() -> EagerInputFile.of(new InMemoryInputFile(bytes), SIZE - 1).newStream())
.as("File is longer than the recorded length")
.isInstanceOf(RuntimeIOException.class)
.hasMessageContaining("Failed to fetch file")
.cause()
.hasMessageContaining("Incorrect length provided for file");

assertThatThrownBy(() -> EagerInputFile.of(new InMemoryInputFile(bytes), SIZE + 1).newStream())
.as("File is shorter than the recorded length")
.isInstanceOf(RuntimeIOException.class)
.hasMessageContaining("Failed to fetch file")
.cause()
.hasMessageContaining("Reached the end of stream with 1 bytes left to read");
}

@Test
public void testOfPreservesHadoopConfiguration() {
Configuration conf = new Configuration(false);
conf.set("test.key", "test-value");
InputFile delegate = HadoopInputFile.fromLocation("file:/tmp/eager-configurable", SIZE, conf);
InputFile eager = EagerInputFile.of(delegate, SIZE);

assertThat(((HadoopConfigurable) eager).getConf().get("test.key"))
.as("getConf() should return the delegate's configuration")
.isEqualTo("test-value");
}
}
17 changes: 13 additions & 4 deletions parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,11 +92,12 @@
import org.apache.iceberg.encryption.NativeEncryptionOutputFile;
import org.apache.iceberg.exceptions.RuntimeIOException;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.hadoop.HadoopInputFile;
import org.apache.iceberg.hadoop.HadoopConfigurable;
import org.apache.iceberg.hadoop.HadoopOutputFile;
import org.apache.iceberg.io.CloseableIterable;
import org.apache.iceberg.io.DataWriter;
import org.apache.iceberg.io.DeleteSchemaUtil;
import org.apache.iceberg.io.EagerInputFile;
import org.apache.iceberg.io.FileAppender;
import org.apache.iceberg.io.InputFile;
import org.apache.iceberg.io.OutputFile;
Expand Down Expand Up @@ -144,6 +145,9 @@ private Parquet() {}
"parquet.read.support.class",
"parquet.crypto.factory.class");

// Size threshold (bytes) at or below which a Parquet file is fetched eagerly on the first read.
private static final long EAGER_FETCH_THRESHOLD_BYTES = 1024 * 1024;

public static WriteBuilder write(OutputFile file) {
if (file instanceof EncryptedOutputFile) {
return write((EncryptedOutputFile) file);
Expand Down Expand Up @@ -1352,7 +1356,12 @@ public ReaderFunction withSchema(Schema expectedSchema) {
}

private ReadBuilder(InputFile file) {
this.file = file;
long fileLength = file.getLength();
this.file = canEagerFetch(fileLength) ? EagerInputFile.of(file, fileLength) : file;

@kevinjqliu kevinjqliu Aug 26, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Codex helped me flag this issue. In summary, S3/GCS/ADLS already accepts and use the (potentially staled) length. Hadoop was using getStat() to get the length directly. So this is a new regression for Hadoop.

Details

For small Hadoop-backed Parquet reads, file.getLength() may come from the manifest-list entry and can be stale. Previously, Hadoop reads obtained the physical size through:

HadoopInputFile.fromStatus(hfile.getStat(), hfile.getConf());

See ParquetIO.file().

This PR wraps files up to 1 MiB using the reported length:

long fileLength = file.getLength();
this.file = canEagerFetch(fileLength) ? EagerInputFile.of(file, fileLength) : file;

The eager reader then reads exactly that many bytes:

byte[] bytes = new byte[(int) length];
IOUtil.readFully(src, bytes, 0, bytes.length);

If the recorded length is smaller than the physical file, the buffer excludes the real Parquet footer and trailing PAR1 magic:

java.lang.RuntimeException: ... is not a Parquet file.
Expected magic number at tail, but found [0, 80, 65, 82]
  at org.apache.parquet.hadoop.ParquetFileReader.readFooter(ParquetFileReader.java:622)
  at org.apache.iceberg.parquet.ReadConf.newReader(ReadConf.java:194)
  at org.apache.iceberg.parquet.ParquetReader.init(ParquetReader.java:74)

If the recorded length is larger, readFully attempts to read past EOF.

#16910 corrects new complete rewrites, but it does not repair existing affected manifest-list entries and still preserves source lengths for manifests carried over by incremental rewrites.

This PR therefore removes Hadoop's existing protection against stale lengths by wrapping HadoopInputFile before ParquetIO can call getStat(). Other input implementations already trusted their reported lengths and are not newly broken.

Please preserve the physical Hadoop FileStatus length before eager wrapping and add a stale-length regression test.

@RussellSpitzer RussellSpitzer Aug 26, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hadoop should not be protecting against this :) Honestly I think this is a misfeature. It exists to cover for bad implementations which store the wrong length, but leads to even weirder situations where HadoopFileIO would work reading a file via S3A that would fail through S3FileIO

@kevinjqliu kevinjqliu Aug 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since all the other FileIO implementations already trust the supplied length, I think its reasonable for HadoopFileIO to do this.

I think its worth calling that out, perhaps in the PR description. And perhaps with a unit test

@RussellSpitzer RussellSpitzer Aug 26, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also to note, this doesn't apply to manfiest-lists written for any currently valid tables since they are all using Avro for manifest list and manifests which do not break on the wrong length regardless

@varun-lakhyani varun-lakhyani Aug 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

updated PR description to include this change for Hadoop and mentioned the trusting length part

}

private static boolean canEagerFetch(long fileLength) {
return fileLength > 0 && fileLength <= EAGER_FETCH_THRESHOLD_BYTES;
}

/**
Expand Down Expand Up @@ -1545,9 +1554,9 @@ public <D> CloseableIterable<D> build() {
|| batchedReaderFuncWithSchema != null
|| readerFunction != null) {
ParquetReadOptions.Builder optionsBuilder;
if (file instanceof HadoopInputFile) {
if (file instanceof HadoopConfigurable) {
// remove read properties already set that may conflict with this read
Configuration conf = new Configuration(((HadoopInputFile) file).getConf());
Configuration conf = new Configuration(((HadoopConfigurable) file).getConf());
for (String property : READ_PROPERTIES_TO_REMOVE) {
conf.unset(property);
}
Expand Down
Loading
Loading