Skip to content

[avro] feat: async stream read - #9632

Open
ariel-miculas wants to merge 13 commits into
apache:mainfrom
ariel-miculas:async-stream-reader
Open

[avro] feat: async stream read#9632
ariel-miculas wants to merge 13 commits into
apache:mainfrom
ariel-miculas:async-stream-reader

Conversation

@ariel-miculas

@ariel-miculas ariel-miculas commented Mar 31, 2026

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Fixes #9668

Rationale for this change

Implement an async streamed reader for avro, this is similar to how datafusion handles json and csv scanning.
There are two main advantages:

  • we don't have to read the entire data upfront, which could require a lot of memory
  • data fetching and decoding can happen in parallel, leading to performance improvements

What changes are included in this PR?

Change the avro reader implementation to use async streams.

Are these changes tested?

Yes, added tests.

Are there any user-facing changes?

Users can now implement get_stream as part of the AsyncFileReader trait

@github-actions github-actions Bot added arrow Changes to the arrow crate arrow-avro arrow-avro crate labels Mar 31, 2026

@AndreaBozzo AndreaBozzo left a comment

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.

Hi @ariel-miculas

while i'm very in favor of the idea behind this, there a few concerns about the implementation

mut state: AvroReaderState<R>,
) -> impl Stream<Item = Result<RecordBatch, ArrowError>> + Send + 'static {
async_stream::try_stream! {
if state.meta.range.start >= state.meta.range.end {

@AndreaBozzo AndreaBozzo Mar 31, 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.

try_stream! is sequential? There's no actual prefetching or concurrent work happening, did i miss something?

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.

This implementation is entirely pull-based. The reader API exposes the stream, which can be driven by a dedicated async task (with the granularity of a record batch at a time) if beneficial to the application.

There is a comment about not being able to use the inner spawn helper because of other constraints on the trait design.

(Responding because I had an input on this change.)

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.

In my end-to-end benchmark, prefetching is done by the object-store implementation, specifically in the call to get_opts

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.

There is a comment about not being able to use the inner spawn helper because of other constraints on the trait design.

In newer changes that I added, the AvroObjectStream offloads the byte stream reading to a separate task using the runtime handle, if that was supplied on object construction.

Comment thread arrow-avro/Cargo.toml
@ariel-miculas

Copy link
Copy Markdown
Contributor Author

@jecsand838 can you please take a look at this?

@alamb

alamb commented Apr 6, 2026

Copy link
Copy Markdown
Contributor

Can we also please file a ticket explaining this need / let us discuss this feature a bit?

Note we have also been trying to separate the IO from the decoding in Parquet -- see https://docs.rs/parquet/58.0.0/parquet/arrow/push_decoder/struct.ParquetPushDecoder.html

Perhaps we could move Avro to that model too rather than implementing the async stuff first

ariel-miculas and others added 3 commits April 15, 2026 10:06
Use the runtime handle provided to the AvroObjectReader to spawn
tasks that perform I/O operations on the ObjectStore.
Comment thread arrow-avro/Cargo.toml Outdated
Comment thread arrow-avro/Cargo.toml Outdated
ariel-miculas and others added 2 commits May 11, 2026 14:08
Co-authored-by: Mikhail Zabaluev <mikhail.zabaluev@gmail.com>
Co-authored-by: Mikhail Zabaluev <mikhail.zabaluev@gmail.com>
@alamb alamb changed the title feat: async stream read [avro] feat: async stream read May 21, 2026
@github-actions

Copy link
Copy Markdown

Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.

@github-actions github-actions Bot added the Stale label Jul 21, 2026
@rluvaton

Copy link
Copy Markdown
Member

not stale

@github-actions github-actions Bot removed the Stale label Jul 22, 2026
@mzabaluev

Copy link
Copy Markdown
Contributor

MIRI failure: we didn't touch datatype_parse, did we?

@ariel-miculas

Copy link
Copy Markdown
Contributor Author

No, not sure what's the root cause:

expected: Error converting 4000000000 into i32 for FixedSizeBinary: out of range integral type conversion attempted
    actual: Parser error: Unsupported type 'FixedSizeBinary(4000000000), '. Must be a supported arrow type name such as 'Int32' or 'Timestamp(ns)'. Error converting 4000000000 into i32 for FixedSizeBinary: number too large to fit in target type

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

arrow Changes to the arrow crate arrow-avro arrow-avro crate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Async avro reader OOM with large files

5 participants