apache / apache/datafusion

Unify stream deserialization

Open
#13,411 1 comment 0 reactions 0 assignees View on GitHub
enhancement
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### Is your feature request related to a problem or challenge?

Implementations for deserialization of streams are disjointed. Unifying stream deserialization would help with logic deduplication and performance. Other formats that support streaming could be added more easily in the future. It would also become easier to create custom data sources that do not use `FileOpener`

For example, when reading the payload from ObjectStore and deserializing `GetResultPayload::Stream`
- `CsvOpener` and `JsonOpener` use `Decoder`s from `arrow-csv` and `arrow-json`, respectively. Additionally, the logic is duplicated.
- `ArrowOpener` and `AvroOpener` read the entire stream before deserializing. Deserializing as the stream is read would parallelize the IO and computation, improving performance.
- `ParquetOpener` has a completely different implementation from the others.

`arrow-ipc` does have a `StreamDecoder` but I believe it only works for the [IPC Streaming Format](https://arrow.apache.org/docs/format/Columnar.html#ipc-streaming-format), not the [IPC File Format](https://arrow.apache.org/docs/format/Columnar.html#ipc-file-format) that ObjectStore is likely to return.

This issue is likely to take several PRs to resolve.

### Describe the solution you'd like

- Add a `Decoder` trait similar to `Decoder` structs of `arrow-csv` and `arrow-json` and implement it for every format.
- Add a `BatchDeserializer` trait (similar to `BatchSerializer`) with the following API
- `digest`: consume bytes
- `next`: try to deserialize a batch, inform if more data is needed or if the stream is exhausted
- `finish`: notify the end of stream
- Add a `DecoderDeserializer` struct that implements `BatchDeserializer` and uses the provided `Decoder` implementation to deserialize the stream.
- Formats that cannot implement a `Decoder` for some reason can instead implement `BatchDeserializer` directly.

### Describe alternatives you've considered

All types can implement `BatchDeserializer` directly instead of relying on `Decoder`, which would remove a layer of abstraction. However this can lead to some duplication, as seen in CSV and JSON deserialization implementations.

### Additional context

- Relevant parts of `CsvOpener` and `JsonOpener` (duplicated logic)
https://github.com/apache/datafusion/blob/a5d0563f53d05f5589df83d163c91910f51020ba/datafusion/core/src/datasource/physical_plan/csv.rs#L653-L683
https://github.com/apache/datafusion/blob/a5d0563f53d05f5589df83d163c91910f51020ba/datafusion/core/src/datasource/physical_plan/json.rs#L307-L340
- Relevant parts of `ArrowOpener` and `AvroOpener` (`.bytes().await` reads the entire stream)
https://github.com/apache/datafusion/blob/a5d0563f53d05f5589df83d163c91910f51020ba/datafusion/core/src/datasource/physical_plan/arrow_file.rs#L239-L245
https://github.com/apache/datafusion/blob/a5d0563f53d05f5589df83d163c91910f51020ba/datafusion/core/src/datasource/physical_plan/avro.rs#L229-L233

Contributor guide

Open the contributing guide

Research direction

Start with the linked CsvOpener and JsonOpener sections in datafusion/core/src/datasource/physical_plan/csv.rs and json.rs, then compare ArrowOpener and AvroOpener in arrow_file.rs and avro.rs. Review the existing arrow-ipc StreamDecoder and the proposed Decoder and BatchDeserializer APIs. Done means stream deserialization is unified across the formats, while formats that cannot use Decoder implement BatchDeserializer directly.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust
Domain
backend, data-engineering
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Mostly clear
Newbie friendliness
35/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.