Unify stream deserialization
- 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
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