Streams implementation #296

H-Plus-Time · 2023-08-22T17:31:43Z

Considerably cleaned up, the api for this is wholly functional thanks to wasm-streams (and the stream! macro from async-stream):

const stream = readFFIStream(targetUrl);
const batches = []
for await (const ffiTable of stream) {
    const batch = parseRecordBatch(memory.buffer, ffiTable.arrayAddr(), ffiTable.schemaAddr(), true);
    batches.push(batch);
}

parseRecordBatch now works for both arrow1 and arrow2, as far as I can tell given the ffi tests, it should be fully compatible.

Two things came up wrt the typescript interface:

The bindings to ReadableStream are opaque, so their TS equivalents are just ReadableStream<R = any>. There's a combination of wasm_bindgen arguments that override a function's return type by fiat.
The much bigger problem - apparently all versions of Typescript (yes, even nightlies) still interpret WHATWG ReadableStreams as both a. lacking a Symbol.asyncIterator and b. being non-iterable. Even once I forced the types to solve for 1, I ended up having to do as unknown as FFIArrowRecordBatch[].

NB: I've opted to leave the write half of this for a future PR (post-integration with the arrow-wasm repo), and of course IPC stream outputs aren't particularly relevant anymore.

…up_async, ParquetWasmError parity for arrow1

…ronous arrow1 ffi read

…t), explicitly pin yarn.

kylebarron · 2023-08-23T15:10:35Z

Awesome, thanks! I'll do a final review and merge tonight or tomorrow!

kylebarron · 2023-08-23T21:58:51Z

.github/workflows/test.yml

@@ -66,7 +66,7 @@ jobs:

      - uses: actions/setup-node@v2
        with:
-          node-version: "16"
+          node-version: "18"


Does this require a Node 18+ for users as well? Or is this just for our tests? Is there a shim/polyfill that users can use for pre-18 versions?

Yep, turns out the ReadableStream constructor only showed up in Node 18. There is indeed a poly/ponyfill, written by a familar face: MattiasBuelens/web-streams-polyfill.

That does bring up a good point - overall README level documentation, and doc-strings for the new exports. On that note, the codeblocks embedded in rust doc comments, do those show up with syntax highlighting for embedded code fragments (a bit like jsdocs) in editors for you?

Yeah they show up in editors and on the docs site https://kylebarron.dev/parquet-wasm/modules/bundler_arrow1.html#readParquet

kylebarron · 2023-08-23T21:59:36Z

scripts/build.sh

-  $FLAGS
+  --features async \
+  $FLAGS &
+[ -n "$CI" ] && wait;


This is awesome; it parallelizes each build? Should we do this for arrow2 as well?

Yeah, turning the parallel behaviour off in CI took a bit of bash-fu (conditionally backgrounding a command isn't possible, but nobody said anything about conditionally running wait 😉 ).

Not much point for arrow2 (the wasm-opt step for arrow2 builds is quite short, while the compile step is highly parallelized, so trying to run that in parallel ended up slowing things down).

I wonder why wasm-opt would be so much slower for arrow1. I don't think this used to be the case; I only recently noticed it taking longer

kylebarron · 2023-08-23T22:01:20Z

src/arrow1/ffi.rs

+    #[wasm_bindgen]
+    pub fn free(self) {
+        drop(self.0)
+    }
+
+    #[wasm_bindgen]
+    pub fn drop(self) {
+        drop(self.0)
+    }


General note: I need to check that the memory pointed to by this actually gets freed by wasm-bindgen. I only need one method but I'm not sure if I can/need to add a manual drop call to release the array

kylebarron · 2023-08-23T22:11:16Z

src/arrow1/reader_async.rs

+pub async fn read_record_batch_stream(
+    url: String,
+    content_length: Option<usize>,
+) -> Result<ParquetRecordBatchStream<Compat<RangedAsyncReader>>> {
+    let content_length = match content_length {
+        Some(_content_length) => _content_length,
+        None => get_content_length(url.clone()).await?,
+    };
+    let reader = crate::common::fetch::create_reader(url, content_length, None);
+
+    let builder = ParquetRecordBatchStreamBuilder::new(reader.compat()).await?;
+    let parquet_reader = builder.build()?;
+    Ok(parquet_reader)
+}


This is so clean!

kylebarron · 2023-08-23T22:19:15Z

src/arrow2/reader_async.rs

+    url: String,
+) -> Result<impl futures::Stream<Item = super::ffi::FFIArrowRecordBatch>> {
+    use async_stream::stream;
+    let inner_stream = stream! {


woah! That's cool! I didn't even know yield was a keyword

The cake (yield) is a lie with this one - that stream! macro slightly extends the grammar (apparently it's on it's way to the language - generator syntax seems to be one of the harder ones to add to a language).

Ah I see it's a proc macro so it can make up new keywords!

kylebarron · 2023-08-24T03:39:10Z

Thanks again! As a follow up I'll try to update some docs and integrate with the new arrow-wasm Table object, but it would be great to release this sooner than later

kylebarron and others added 9 commits August 6, 2023 23:20

wip ffi for arrow1

2da4734

add make_array

ab7c571

Merge branch 'kyle/wip-arrow1-ffi'

7ad666b

Fix up From<FFIArrowRecordBatch>, shift create_reader into common/fetch

359922e

Internal-only arrow1 equivalents of read_metadata_async, read_row_gro…

18f9cbb

…up_async, ParquetWasmError parity for arrow1

arrow1 ffi stream functional api + cargo fmt

4690ad9

Match ffi api of arrow2, get parseRecordBatch tests to pass for synch…

7c598f3

…ronous arrow1 ffi read

Satisfy ffi stream tests for arrow1

058f606

functional api for arrow2 ffi stream reads + tests

0dcc0b2

H-Plus-Time force-pushed the streams-implementation branch from 7995d88 to 5ae82dc Compare August 22, 2023 17:37

Formatting!

8051f22

H-Plus-Time force-pushed the streams-implementation branch from 5ae82dc to 8051f22 Compare August 22, 2023 17:39

H-Plus-Time added 3 commits August 23, 2023 21:29

Fix feature flag combinations

35fac83

Avoid build parallelism in CI (still highly useful for arrow1 wasm-op…

2a2ab3d

…t), explicitly pin yarn.

Bump node to v18 (ReadableStream global, purely for testing)

bdf582d

kylebarron self-requested a review August 23, 2023 21:58

kylebarron approved these changes Aug 23, 2023

View reviewed changes

kylebarron merged commit 0c0fc2a into kylebarron:main Aug 24, 2023
5 checks passed

This was referenced Aug 29, 2023

Read/write from/to streams #267

Closed

Return iterator of arrow record batches to JS #95

Closed

wip ffi for arrow1 #278

Closed

try to add node async tests #207

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Streams implementation #296

Streams implementation #296

H-Plus-Time commented Aug 22, 2023 •

edited

Loading

kylebarron commented Aug 23, 2023

kylebarron Aug 23, 2023

H-Plus-Time Aug 24, 2023

kylebarron Aug 24, 2023

kylebarron Aug 23, 2023

H-Plus-Time Aug 23, 2023

kylebarron Aug 24, 2023

kylebarron Aug 23, 2023

kylebarron Aug 23, 2023

kylebarron Aug 23, 2023

H-Plus-Time Aug 23, 2023

kylebarron Aug 24, 2023

kylebarron commented Aug 24, 2023

Streams implementation #296

Streams implementation #296

Conversation

H-Plus-Time commented Aug 22, 2023 • edited Loading

kylebarron commented Aug 23, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kylebarron commented Aug 24, 2023

H-Plus-Time commented Aug 22, 2023 •

edited

Loading