diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/HadoopInputFile.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/HadoopInputFile.java index fa6a7bdf42..adae5383e4 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/HadoopInputFile.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/util/HadoopInputFile.java @@ -28,6 +28,7 @@ import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.FutureDataInputStreamBuilder; import org.apache.hadoop.fs.Path; import org.apache.parquet.io.InputFile; import org.apache.parquet.io.SeekableInputStream; @@ -39,6 +40,14 @@ public class HadoopInputFile implements InputFile { */ private static final String OPENFILE_READ_POLICY_KEY = "fs.option.openfile.read.policy"; + /** + * openFile() option name for passing a known file length: {@value}. + * + *
This is an optional filesystem hint, so retain the string form to support Hadoop versions before the + * constant was introduced. + */ + private static final String OPENFILE_LENGTH_KEY = "fs.option.openfile.length"; + /** * Read policy when opening parquet files: {@value}. *
Policy-aware stores pick the first policy they recognize in the list. @@ -53,6 +62,8 @@ public class HadoopInputFile implements InputFile { private static final String PARQUET_READ_POLICY = "parquet, vector, random, adaptive"; private final FileSystem fs; + private final Path path; + private final long length; private final FileStatus stat; private final Configuration conf; @@ -61,6 +72,32 @@ public static HadoopInputFile fromPath(Path path, Configuration conf) throws IOE return new HadoopInputFile(fs, fs.getFileStatus(path), conf); } + /** + * Creates an input file using a caller-supplied file length. + * + *
The length is trusted and no file status lookup is performed. Callers must provide the exact + * length of the file that will be opened. The file system may defer checking whether the file + * exists until the first read. If the supplied length is incorrect, reads may fail. + * + *
When a {@link FileStatus} is available, prefer {@link #fromStatus(FileStatus, Configuration)}. + * A length alone does not preserve file-system-specific identity metadata used when opening a file + * (for example, S3A ETags and version IDs), so initial-open change detection may be weaker. + * + * @param path file path + * @param length exact file length in bytes + * @param conf configuration used to resolve the file system + * @return an input file that uses the supplied length + * @throws IllegalArgumentException if {@code length} is negative + * @throws IOException if the file system cannot be resolved + */ + public static HadoopInputFile fromPath(Path path, long length, Configuration conf) throws IOException { + if (length < 0) { + throw new IllegalArgumentException("Invalid file length: " + length); + } + FileSystem fs = path.getFileSystem(conf); + return new HadoopInputFile(fs, path, length, conf); + } + public static HadoopInputFile fromPathUnchecked(Path path, Configuration conf) { try { return fromPath(path, conf); @@ -76,28 +113,38 @@ public static HadoopInputFile fromStatus(FileStatus stat, Configuration conf) th private HadoopInputFile(FileSystem fs, FileStatus stat, Configuration conf) { this.fs = fs; + this.path = stat.getPath(); + this.length = stat.getLen(); this.stat = stat; this.conf = conf; } + private HadoopInputFile(FileSystem fs, Path path, long length, Configuration conf) { + this.fs = fs; + this.path = path; + this.length = length; + this.stat = null; + this.conf = conf; + } + public Configuration getConfiguration() { return conf; } public Path getPath() { - return stat.getPath(); + return path; } @Override public long getLength() { - return stat.getLen(); + return length; } /** * Open the file. - *
Uses {@code FileSystem.openFile()} so that - * the existing FileStatus can be passed down: saves a HEAD request on cloud storage. - * and ignored everywhere else. + *
Uses {@code FileSystem.openFile()} so that the existing FileStatus, when available, or the
+ * known file length can be passed down. File systems may use either to avoid a metadata request
+ * when opening the file.
*
* @return the input stream.
*
@@ -109,20 +156,24 @@ public long getLength() {
public SeekableInputStream newStream() throws IOException {
FSDataInputStream stream;
try {
- // this method is async so that implementations may do async HEAD head
+ // this method is async so that implementations may do async HEAD
// requests, such as S3A/ABFS when a file status is passed down.
- final CompletableFuture