AVRO-3594: FsInput to use openFile() API #1807

steveloughran · 2022-08-04T16:05:11Z

Boost performance reading from object stores in hadoop 3.3.0+ by using the openFile builder API and passing in the file length as an option (can save a HEAD) and asks for adaptive IO (sequential going to random if the client starts seeking)

saving that HEAD request is a key benefit against s3 as it can save 50-100 mS per file.

Jira

My PR addresses the following Avro Jira issues and references them in the PR title. For example, "AVRO-1234: My Avro PR"
- https://issues.apache.org/jira/browse/AVRO-XXX
- In case you are adding a dependency, check if the license complies with the ASF 3rd Party License Policy.

Tests

My PR does not need testing for this extremely good reason:

All existing local file IO tests act as regression tests.
avro isn't set up for integration tests with abfs/gs/s3a urls.
mocking doesn't really do much here.

Integration tests would be the way to do this, but the foundational set up to do this
is pretty complex. My cloudstream project downstream of spark is set up to do this.

Commits

My commits all reference Jira issues in their subject lines. In addition, my commits follow the guidelines from "How to write a good git commit message":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters (not including Jira issue reference)
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"

Documentation

In case of new functionality, my PR adds documentation that describes how to use it.
- All the public functions and the classes in the PR contain Javadoc that explain what it does

no new docs

steveloughran · 2022-08-04T16:49:33Z

except that hadoop-2 profile still exists, doesn't it? which means that even though hadoop 3 profile is full of features, the 2.x one blocks things from working

clesaec · 2022-08-05T08:14:02Z

lang/java/mapred/src/main/java/org/apache/avro/mapred/FsInput.java

+    // Filesystems which don't recognize the options will ignore them
+
+    this.stream = awaitFuture(fileSystem.openFile(path).opt("fs.option.openfile.read.policy", "adaptive")
+        .opt("fs.option.openfile.length", Long.toString(len)).build());


Nice optimization,
May be better with usage of Options.OpenFileOptions constants
FS_OPTION_OPENFILE_LENGTH
FS_OPTION_OPENFILE_READ_POLICY
FS_OPTION_OPENFILE_READ_POLICY_ADAPTIVE

we only added those options explicitly in branch-3.3; the code wouldn't compile/link with 3.3.0. hence the strings. Unfortunately it also means you don't get that speedup until 3.3.5 ships ....but the release will be ready.
3.3.0 did add the withFileStatus(FileStatus) param which was my original design for passing in all filestatus info, inc etag and maybe version, so you can go straight from listing to opening.

first in s3a, added abfs in 3.3.5. but it is too brittle because the path checking requires status.getPath to equal the path opened. and hive with its wrapper fs doesn't always do that.
Passing in file length is guaranteed to be ignored or actually used...no brittleness. it also suits hive where workers know the length of the file but don't have a status.

One thing i can add with immediate benefit in 3.3.0 is the initial the fs.s3a.experimental.fadvise option, which again can mandate be adaptive, even on hive clusters where they explicitly set read policy to be random (which some do for max orc/parquet performance). The new opt fs.option.openfile.read.policy is an evolution of that (you can now specify a list of policies and the first one recognised is understood. if someone ever implemented explicit "parquet", "orc", and "avro" for example), you could say the read policy is "orc, random, adaptive" and get the first one known.

ok, thanks for explanation.

note that org.apache.hadoop.fs.AvroFSInput does this

clesaec

LGTM

clesaec · 2022-08-05T09:50:09Z

lang/java/mapred/src/main/java/org/apache/avro/mapred/FsInput.java

+    // Filesystems which don't recognize the options will ignore them
+
+    this.stream = awaitFuture(fileSystem.openFile(path).opt("fs.option.openfile.read.policy", "adaptive")
+        .opt("fs.option.openfile.length", Long.toString(len)).build());


ok, thanks for explanation.

steveloughran · 2022-08-05T17:06:51Z

@clesaec this can't go in until hadoop2 is cut as a profile.

I am actually doing a shim library to do reflection invocation on the newer operations, with fallbacks if not found.
https://github.com/steveloughran/fs-api-shim

but even so, the sooner avro goes to recent hadoop 3.x release only the better.

Boost performance reading from object stores in hadoop by using the openFile builder API and passing in the file length as an option (can save a HEAD) and asks for adaptive IO (sequential going to random if the client starts seeking)

Change-Id: Ie7099208b601a08823775e901359a6727f0c6fe6

steveloughran · 2024-12-06T15:29:19Z

@opwvhk thanks for the review, will do a merge rebase to make sure it is good.
FYI, parquet picked this up last week as now they are hadoop 3.3.0+ they can do this without reflection.

steveloughran · 2024-12-06T15:31:46Z

lang/java/mapred/src/main/java/org/apache/avro/mapred/FsInput.java

+    // "random"
+    // will not suffer when reading large avro files.
+    this.stream = awaitFuture(fileSystem.openFile(path)
+        .opt(FS_OPTION_OPENFILE_READ_POLICY, FS_OPTION_OPENFILE_READ_POLICY_ADAPTIVE).withFileStatus(st).build());


need to add the same fallback code as in parquet; s3a connector in 3.3.0-3.3.4 validates the whole path, and overreacts to hive's wrapping of the fs by its own VFS; 3.3.5+ only look at the filename.

Change-Id: I4ad736a35d625db9d67deda1d11a2d6e51a789e6

Change-Id: If4c9726a97d1d550d0256d4798ce1b484a72fe3a

* Update for the specific "avro" read policy in Hadoop 3.4.1. (S3A maps to sequential/forward seeking IO) * Add the file status if the path names all match. Change-Id: If755c047a6fbafff20e9eaacee257be72ca76ddb

* spotlessness Change-Id: Ia7b0336853c5b5f24bb015e1d27d3401cfedd39a

Change-Id: I3f3ac8c579d0b4e93789e9fc952a503943e4b659

Key one: what happens when you try to open a file that isn't there. The other two just check semantics of out of bound seeks/reads. Change-Id: I2417506354ae92e27de3b18c357e97ed73369baa

steveloughran · 2025-08-04T10:27:16Z

new test failure

Error:  Tests run: 10, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 0.518 s <<< FAILURE! -- in org.apache.avro.mapreduce.TestFsInput
Error:  org.apache.avro.mapreduce.TestFsInput.seekPastEOF -- Time elapsed: 0.015 s <<< ERROR!
java.io.EOFException: Cannot seek after EOF
	at org.apache.hadoop.fs.ChecksumFileSystem$FSDataBoundedInputStream.seek(ChecksumFileSystem.java:553)
	at org.apache.avro.mapred.FsInput.seek(FsInput.java:75)
	at org.apache.avro.mapreduce.TestFsInput.seekPastEOF(TestFsInput.java:145)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)

Build flagged that two existing tests weren't using the try with resources input stream `in`. This means they weren't actually testing that the constructors produced valid streams. Change-Id: Iee35ef8931c71d940640ad67dc06d963ef42cd63

anujmodi2021 · 2025-08-07T04:58:38Z

lang/java/mapred/src/main/java/org/apache/avro/mapred/FsInput.java

+    // optimize read performance and save on a HEAD request when opening
+    // a file.
+    final FutureDataInputStreamBuilder builder = fileSystem.openFile(path).opt(FS_OPTION_OPENFILE_READ_POLICY,
+        "avro, sequential, adaptive");


Since these read policies are going to be used by file sysem to understand the user intended pattern. Do we have a way to standardise these values via means of constants or an enum class that can be used across projects?

I might be missing a similar thing already existing in hadoop. Please let me know if its already there.

we do have them in hadoop, but that hard codes to specific versions (the "avro" one is fairly recent. Better for me to put them into an avro class

github-actions bot added the Java Pull Requests for Java binding label Aug 4, 2022

steveloughran marked this pull request as draft August 4, 2022 16:47

clesaec reviewed Aug 5, 2022

View reviewed changes

clesaec approved these changes Aug 5, 2022

View reviewed changes

RyanSkraba changed the title ~~AVRO-3594. FsInput to use openFile() API~~ AVRO-3594: FsInput to use openFile() API Aug 25, 2022

AVRO-3594: FsInput to use openFile() API

088c346

Boost performance reading from object stores in hadoop by using the openFile builder API and passing in the file length as an option (can save a HEAD) and asks for adaptive IO (sequential going to random if the client starts seeking)

steveloughran force-pushed the AVRO-3594-openfile branch from 19c0695 to d70a1ba Compare April 21, 2023 15:37

AVRO-3594. openFile() API through shim

a20c7fb

Change-Id: Ie7099208b601a08823775e901359a6727f0c6fe6

steveloughran force-pushed the AVRO-3594-openfile branch from d70a1ba to a20c7fb Compare April 21, 2023 15:50

opwvhk approved these changes Jul 1, 2024

View reviewed changes

steveloughran commented Dec 6, 2024

View reviewed changes

steveloughran added 4 commits December 6, 2024 15:33

Merge branch 'main' into AVRO-3594-openfile

401125b

Change-Id: I4ad736a35d625db9d67deda1d11a2d6e51a789e6

Merge branch 'main' into AVRO-3594-openfile

b07687a

Change-Id: If4c9726a97d1d550d0256d4798ce1b484a72fe3a

AVRO-3594. openFile()

35457e9

* Update for the specific "avro" read policy in Hadoop 3.4.1. (S3A maps to sequential/forward seeking IO) * Add the file status if the path names all match. Change-Id: If755c047a6fbafff20e9eaacee257be72ca76ddb

AVRO-3594. openFile()

7a4ab3e

* spotlessness Change-Id: Ia7b0336853c5b5f24bb015e1d27d3401cfedd39a

steveloughran marked this pull request as ready for review July 31, 2025 20:27

steveloughran added 2 commits July 31, 2025 21:29

Merge branch 'main' into AVRO-3594-openfile

a0c193e

Change-Id: I3f3ac8c579d0b4e93789e9fc952a503943e4b659

AVRO-3594. Added some tests around stream reads.

0c310ff

Key one: what happens when you try to open a file that isn't there. The other two just check semantics of out of bound seeks/reads. Change-Id: I2417506354ae92e27de3b18c357e97ed73369baa

AVRO-3594. Fixed all the test asserts

b47316a

Build flagged that two existing tests weren't using the try with resources input stream `in`. This means they weren't actually testing that the constructors produced valid streams. Change-Id: Iee35ef8931c71d940640ad67dc06d963ef42cd63

anujmodi2021 reviewed Aug 7, 2025

View reviewed changes

AVRO-3594: FsInput to use openFile() API #1807

Are you sure you want to change the base?

AVRO-3594: FsInput to use openFile() API #1807

Uh oh!

Conversation

steveloughran commented Aug 4, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Jira

Tests

Commits

Documentation

Uh oh!

steveloughran commented Aug 4, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

clesaec left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

steveloughran commented Aug 5, 2022

Uh oh!

steveloughran commented Dec 6, 2024

Uh oh!

Choose a reason for hiding this comment

Uh oh!

steveloughran commented Aug 4, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

steveloughran commented Aug 4, 2022 •

edited

Loading