Add support for IPC Streaming Read/Write

[joshuataylor]

This is waiting for jorgecarleitao/arrow2#1095 to be reviewed and merged.

I've pretty much copied IPC, and made changes to use StreamWriter instead. I also think it might be worth moving IPC into its own directory, similar to parquet/csv_core/ndjson_core?

Closes #3778

[ritchie46]

Thanks for the PR.

I've pretty much copied IPC, and made changes to use StreamWriter instead. I also think it might be worth moving IPC into its own directory, similar to parquet/csv_core/ndjson_core?

Yes, could you do that? And would you then feature gate this under ipc_streaming?

[joshuataylor]

That makes sense, I've added it. Pretty straight forward process.

[joshuataylor]

I've moved it to the ipc, with ipc_streaming under a different feature flag. They both rely on io_ipc in arrow2.

We just need to bump arrow2 to support this -- #3793

Sorry as well for the multiple commits, wasn't 100% sure how to do it for feature flags (I'm new to Rust, so apologies as I'm still learning).

[ritchie46]

Sorry as well for the multiple commits, wasn't 100% sure how to do it for feature flags (I'm new to Rust, so apologies as I'm still learning).

No worries, I have got a squash button! :)

[joshuataylor]

Yeah, was just about some people review commits or get notifications on pushes.

As this is my first PR to this repository, and I'm new to Rust, please give any feedback you find helpful 🙌

[codecov-commenter]

Codecov Report

Merging #3783 (d206421) into master (c57df45) will decrease coverage by 0.15%.
The diff coverage is 0.00%.

@@            Coverage Diff             @@
##           master    #3783      +/-   ##
==========================================
- Coverage   78.12%   77.96%   -0.16%     
==========================================
  Files         445      446       +1     
  Lines       73691    73838     +147     
==========================================
+ Hits        57569    57570       +1     
- Misses      16122    16268     +146     

Impacted Files	Coverage Δ
polars/polars-io/src/aggregations.rs	7.27% <ø> (ø)
polars/polars-io/src/ipc/ipc_file.rs	94.32% <ø> (ø)
polars/polars-io/src/ipc/ipc_stream.rs	0.00% <0.00%> (ø)
polars/polars-io/src/lib.rs	60.93% <ø> (ø)
polars/polars-io/src/predicates.rs	83.33% <ø> (ø)
polars/polars-io/src/prelude.rs	100.00% <ø> (ø)
polars/polars-io/src/utils.rs	72.97% <ø> (ø)
...polars-core/src/chunked_array/iterator/par/list.rs	95.23% <0.00%> (-4.77%)	⬇️
...lars/polars-core/src/chunked_array/ops/take/mod.rs	62.45% <0.00%> (-0.34%)	⬇️
polars/polars-io/src/csv_core/buffer.rs	79.49% <0.00%> (-0.26%)	⬇️
... and 4 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update c57df45...d206421. Read the comment docs.

[ritchie46]

Thanks for the PR. I have a few comments, nothing big.

[joshuataylor]

I'm running this through my streaming files from SF and will let you know if I have any dramas with them.

[joshuataylor]

I have ran this across lots of Snowflake files using their sample dataset + some test ones from myself and it seems to be working well.

The two main changes I've made are:

1. Move metadata to self, as the schema can only be read once. Otherwise it throws an error, if you do something like this:

    let f = File::open("file.arrow-stream").unwrap();
    let mut reader = BufReader::new(f);
    let mut stream_reader = IpcStreamReader::new(reader);
    let mut schema = stream_reader.arrow_schema().unwrap();
    let mut df = stream_reader.finish().unwrap(); // can't access schema again.

2. Match on StreamState, I was sometimes getting errors that there were no more batches to read. 186f0dc