feat(parquet): add async support for wasm targets - #8460
Conversation
a05e85c to
04290ef
Compare
Signed-off-by: Ho Kim <ho.kim@ulagbulag.io>
04290ef to
39ded9b
Compare
Thank you for this PR @HoKim98 One thing I have been pushing on is a "SansIO" type of API for the parquet readers, described here: The idea is that with those APIs, you could use your own IO routines (aka not the AsyncRead traits). We recently released a metadata parsing version here: This lets you decode ParquetMetadata with any IO (e.g this example): use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeek, AsyncSeekExt};
// This function decodes Parquet Metadata from anything that implements
// [`AsyncRead`] and [`AsyncSeek`] such as a tokio::fs::File
async fn decode_metadata(
file_len: u64,
mut async_source: impl AsyncRead + AsyncSeek + Unpin
) -> Result<ParquetMetaData, ParquetError> {
// We need a ParquetMetaDataPushDecoder to decode the metadata.
let mut decoder = ParquetMetaDataPushDecoder::try_new(file_len).unwrap();
loop {
match decoder.try_decode() {
Ok(DecodeResult::Data(metadata)) => { return Ok(metadata); } // decode successful
Ok(DecodeResult::NeedsData(ranges)) => {
// The decoder needs more data
//
// In this example we use the AsyncRead and AsyncSeek traits to read the
// required ranges from the async source.
let mut data = Vec::with_capacity(ranges.len());
for range in &ranges {
let mut buffer = vec![0; (range.end - range.start) as usize];
async_source.seek(std::io::SeekFrom::Start(range.start)).await?;
async_source.read_exact(&mut buffer).await?;
data.push(Bytes::from(buffer));
}
// Push the data into the decoder and try to decode again on the next iteration.
decoder.push_ranges(ranges, data).unwrap();
}
Ok(DecodeResult::Finished) => { unreachable!("returned metadata in previous match arm") }
Err(e) => return Err(e),
}
}
}I also have code to do the same for the actual parquet decoder here: If you are interested in that feature I will try and prioritize it more |
|
Hello @alamb, thank you for great work!
I'll try and take some feedbacks about building on WASM targets. |
|
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. |
Which issue does this PR close?
AsyncFileWriterdoesn't need to beSend#7612.Rationale for this change
This PR introduces
BoxedFuturetype alias to replacefutures::future::BoxFuture.It helps building on WebAssembly targets (e.g.
wasm32-unknown-unknown,wasm32-wasip2).The original codes are driven from: https://github.com/apache/opendal/blob/37efe24235388788b892f46e4101d59ecd37918c/core/src/raw/futures_util.rs#L43-L58
I think it's not an ultimate solution, because the API rework is in progress.
But I think it's useful for now.
What changes are included in this PR?
futures::future::BoxFutureintoBoxedFuturefor WASM targets, which is thread-local.FutureExt::boxedcalls intoBox::pinto catchSendautomaticallyAre these changes tested?
All build steps below are passed on my local machine:
cargo build --package parquet --target wasm32-unknown-unknown -p parquet --features asynccargo build --package parquet --target wasm32-wasip1 -p parquet --features asynccargo build --package parquet --target wasm32-wasip2 -p parquet --features asyncThe tests are not passed because:
wasm32-unknown-unknown:getrandomis not supportedwasm32-wasip1,wasm32-wasip2:tokio/fsis not supported yetAre there any user-facing changes?
NONE