Add FFI for Arrow C Stream Interface #1384

viirya · 2022-03-02T09:25:45Z

Which issue does this PR close?

Closes #1348.

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

viirya · 2022-03-03T00:32:23Z

Hmm, not sure why

error[E0433]: failed to resolve: could not find `ipc` in the crate root
   --> arrow/src/ffi_stream.rs:388:16
    |
388 |     use crate::ipc::reader::FileReader;
    |                ^^^ could not find `ipc` in the crate root

only in Test Workspace on AMD64 Rust stable?

alamb

I started looking through this, but I don't have time at this moment to complete a detailed review. It looks very cool though and I will try and review it more carefully over the coming days

alamb · 2022-03-03T18:51:08Z

arrow/src/ffi_stream.rs

+    use std::fs::File;
+
+    use crate::datatypes::Schema;
+    use crate::ipc::reader::FileReader;


Since these tests use ipc they probably need to be feature flagged like

#[cfg(feature = "ipc")]

alamb · 2022-03-03T18:52:16Z

arrow/src/ffi_stream.rs

+
+        assert_eq!(batch, expected_batch);
+    }
+}


@jorgecarleitao 's PRs in arrow2 had a nice roundtrip test:

https://github.com/jorgecarleitao/arrow2/pull/857/files#diff-dedf58162923efd1b8f544c8be899fe250c43eff0d7b6c394b5dcb020f87d3a3R7

It might be good to add one here as well (which would also mean the test for ffi didn't depend on the ipc feature)

Rewrote the tests to roundtrip tests.

And it doesn't depend on ipc now.

alamb · 2022-03-03T18:55:20Z

arrow/src/error.rs

@@ -40,6 +40,8 @@ pub enum ArrowError {
    ParquetError(String),
    /// Error during import or export to/from the C Data Interface
    CDataInterface(String),
+    /// Error during import or export to/from the C Stream Interface
+    CStreamInterface(String),


Since the Stream interface is built on the CDataInterface, what do you think about just reusing the same CDataInterface error variant instead of adding a new one?

alamb · 2022-03-03T18:56:06Z

arrow/src/ffi_stream.rs

+// under the License.
+
+//! Contains declarations to bind to the [C Stream Interface](https://arrow.apache.org/docs/format/CStreamInterface.html).
+//!


how did you create this file (was it bindgen?)

I followed the C Stream interface definition. Let me go to generate it using bindgen again.

Re-generated the binding by bindgen and verified it looks the same. I added This was created by bindgen in the doc.

alamb · 2022-03-06T11:41:29Z

I plan to review this early next week (BTW the integration test failure has been fixed on master, if you want to merge / rebase to get a clean CI)

viirya · 2022-03-06T18:01:36Z

Thanks @alamb . I'm writing roundtrip test as you suggested. I will do rebase after that.

codecov-commenter · 2022-03-07T00:09:42Z

Codecov Report

Merging #1384 (3716c16) into master (d384836) will increase coverage by 0.02%.
The diff coverage is 79.14%.

@@            Coverage Diff             @@
##           master    #1384      +/-   ##
==========================================
+ Coverage   82.70%   82.72%   +0.02%     
==========================================
  Files         187      189       +2     
  Lines       54208    54561     +353     
==========================================
+ Hits        44832    45138     +306     
- Misses       9376     9423      +47

Impacted Files	Coverage Δ
arrow/src/array/array_binary.rs	`92.93% <ø> (+0.31%)`	⬆️
arrow/src/ffi_stream.rs	`79.03% <79.03%> (ø)`
arrow/src/ffi.rs	`87.50% <100.00%> (ø)`
arrow/src/array/equal/list.rs	`79.38% <0.00%> (-9.85%)`	⬇️
parquet_derive/src/parquet_field.rs	`65.98% <0.00%> (-0.23%)`	⬇️
arrow/src/array/transform/mod.rs	`86.35% <0.00%> (-0.23%)`	⬇️
parquet/src/encodings/encoding.rs	`93.37% <0.00%> (-0.19%)`	⬇️
arrow/src/compute/kernels/filter.rs	`88.00% <0.00%> (ø)`
arrow/src/compute/kernels/length.rs	`100.00% <0.00%> (ø)`
...ion-testing/src/bin/arrow-json-integration-test.rs	`0.00% <0.00%> (ø)`
... and 12 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 1c25580...3716c16. Read the comment docs.

arrow/src/ffi_stream.rs

sunchao · 2022-03-08T17:36:09Z

arrow/src/ffi_stream.rs

+        Arc::into_raw(this)
+    }
+
+    /// Get `FFI_ArrowArrayStream` from raw pointer


nit: maybe we should mention that the input ptr is consumed after the call and the ownership of the input FFI_ArrowArrayStream has been transferred to the returned value.

Hmm, do we change the ownership if the input FFI_ArrowArrayStream? As it is behind a raw pointer, we cannot move it. That's why I need to do clone here.

sunchao · 2022-03-08T17:36:36Z

arrow/src/ffi_stream.rs

+}
+
+impl ExportedArrayStream {
+    fn get_private_data(&self) -> Box<StreamPrivateData> {


nit: can we just return &mut StreamPrivateData

private_data is kept in the stream as a raw pointer. We cannot create and return a reference of a temporary object here.

I was thinking of something like this:

fn get_private_data(&mut self) -> &mut StreamPrivateData { unsafe { &mut *((*self.stream).private_data as *mut StreamPrivateData) } }

Oh, this works. If we just take reference, we don't need move it. I will change it.

sunchao · 2022-03-08T18:11:14Z

arrow/src/ffi_stream.rs

+}
+
+/// Get the last error from `ArrowArrayStreamReader`
+fn get_stream_last_error(stream_reader: &ArrowArrayStreamReader) -> Option<String> {


can this be a method of ArrowArrayStreamReader?

arrow/src/ffi_stream.rs

sunchao · 2022-03-08T19:16:48Z

arrow/src/ffi_stream.rs

+        let empty_schema = Arc::new(FFI_ArrowSchema::empty());
+        let schema_ptr = Arc::into_raw(empty_schema) as *mut FFI_ArrowSchema;
+
+        let ret_code = unsafe { self.stream.get_schema.unwrap()(stream_ptr, schema_ptr) };


maybe we can have a try_new ctor for ArrowArrayStreamReader and initialize the cached schema via get_schema there.

sunchao · 2022-03-08T19:40:12Z

arrow/src/ffi_stream.rs

+
+impl RecordBatchReader for ArrowArrayStreamReader {
+    fn schema(&self) -> SchemaRef {
+        if self.stream.get_schema.is_none() {


I think we can just unwrap here instead of returning an empty schema. It's an implementation error if the callback is not set.

Now schema is cached when constructing ArrowArrayStreamReader. Here we simply return it.

sunchao · 2022-03-08T19:42:33Z

arrow/src/ffi_stream.rs

+}
+
+#[cfg(test)]
+mod tests {


maybe we should have some integration tests between Python and Rust too in arrow-pyarrow-integration-testing.

do we have some for ffi too? if so, I may follow it up. If not, maybe we can have both in later PR.

Not quite sure. I'm assuming PyArrow supports it since Arrow C++ implemented the stream interface.

I just checked PyArrow. It has the stream interface. But at Rust side, the arrow-pyarrow-integration-testing crate doesn't have some basic code for testing it yet. I may need sometime to try it and write some integration tests between Python and Rust.

I'm fine to hold this until I add the integration test, or I can work on in in later PR.

sunchao · 2022-03-08T22:24:58Z

cc @jorgecarleitao too

viirya · 2022-03-13T22:34:31Z

@sunchao Any more comments? Thanks.

sunchao · 2022-03-14T22:21:28Z

Will take another look soon. Sorry for the delay @viirya !

viirya · 2022-03-14T23:32:20Z

No problem! Thank you @sunchao !

alamb · 2022-03-16T07:38:38Z

FWIW I plan to make a 11.0 release candidate Thursday or Friday so if this is merged by then it will be included. Thanks a lot for this @viirya and @sunchao

sunchao · 2022-03-09T01:03:26Z

arrow/src/ffi_stream.rs

+}
+
+#[cfg(test)]
+mod tests {


Not quite sure. I'm assuming PyArrow supports it since Arrow C++ implemented the stream interface.

sunchao · 2022-03-09T01:05:35Z

arrow/src/ffi_stream.rs

+}
+
+impl ExportedArrayStream {
+    fn get_private_data(&self) -> Box<StreamPrivateData> {


I was thinking of something like this:

fn get_private_data(&mut self) -> &mut StreamPrivateData { unsafe { &mut *((*self.stream).private_data as *mut StreamPrivateData) } }

arrow/src/ffi_stream.rs

sunchao · 2022-03-18T05:24:04Z

arrow/src/ffi_stream.rs

+    /// # Safety
+    /// Assumes that the pointer represents valid C Stream Interfaces, both in memory
+    /// representation and lifetime via the `release` mechanism.
+    /// This function copies the content from the raw pointer and cleans it up to prevent


Why we need this when we already have ArrowArrayStreamReader.from_raw?

Not sure if in any case, users need FFI_ArrowArrayStream instead of ArrowArrayStreamReader? I can remove it now, we can add it if it is needed in the future.

sunchao · 2022-03-18T05:34:17Z

arrow/src/ffi_stream.rs

+
+        match schema {
+            Ok(mut schema) => {
+                unsafe {


we can perhaps just do:

Ok(mut schema) => unsafe { std::ptr::copy(&schema as *const FFI_ArrowSchema, out, 1); schema.release = None; 0 },

arrow/src/ffi_stream.rs

sunchao · 2022-03-18T06:00:41Z

arrow/src/ffi_stream.rs

+
+        let stream_data = std::ptr::replace(raw_stream, FFI_ArrowArrayStream::empty());
+
+        let stream = Arc::new(stream_data);


can we call Self::try_new(stream_data) here?

sunchao · 2022-03-18T06:04:57Z

arrow/src/ffi_stream.rs

+    fn get_stream_last_error(&self) -> Option<String> {
+        self.stream.get_last_error?;
+
+        let stream_ptr = Arc::into_raw(self.stream.clone()) as *mut FFI_ArrowArrayStream;


hmm can we use Arc::as_ptr(&self.stream) as *mut FFI_ArrowArrayStream here?

sunchao · 2022-03-18T06:05:50Z

arrow/src/ffi_stream.rs

+    type Item = Result<RecordBatch>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.stream.get_next?;


this seems unnecessary

If the stream is released, get_next is None.

But I think we checked in the constructors to make sure it is not released?

Hmm, okay, I thought it is safer to make sure it is not released here too. I will use unwrap directly then.

sunchao · 2022-03-18T06:19:10Z

arrow/src/ffi_stream.rs

+        ffi_array.release?;
+
+        let schema_ref = self.schema();
+        let schema = FFI_ArrowSchema::try_from(schema_ref.as_ref());


we can just do:

let schema = FFI_ArrowSchema::try_from(schema_ref.as_ref()).ok()?;

sunchao · 2022-03-18T06:19:19Z

arrow/src/ffi_stream.rs

+            }
+            .to_data();
+
+            if data.is_err() {


sunchao · 2022-03-18T06:21:32Z

arrow/src/ffi_stream.rs

+            return Some(Err(schema.err().unwrap()));
+        }
+
+        if ret_code == 0 {


can we skip the above part of creating schema if ret_code is not zero?

alamb · 2022-03-23T20:54:35Z

I wonder if this one is ready to go? It seems like it has stalled (no thanks to me, of course, who hasn't reviewed it 😢 )

viirya · 2022-03-23T22:32:22Z

ah, I think I missed @chao's latest comments. I will address them soon. Thanks @alamb for reminding me. 😄

sunchao · 2022-03-23T22:34:56Z

I think it's almost ready.😀 will take another round of look after the comments are addressed.

viirya · 2022-03-23T23:48:58Z

Thanks @sunchao. Addressed your latest comments.

alamb · 2022-03-28T20:49:55Z

Is this one ready to merge? I am preparing to create an arrow release candidate later this week

viirya · 2022-03-28T21:00:25Z

@sunchao Do you need to look this again?

sunchao · 2022-03-29T21:40:33Z

Sorry I'm on vacation. Will take another look tomorrow.

sunchao · 2022-03-29T21:58:46Z

arrow/src/ffi_stream.rs

+//!
+//! // export it
+//! let stream = Arc::new(FFI_ArrowArrayStream::new(reader));
+//! let stream_ptr = FFI_ArrowArrayStream::to_raw(stream) as *mut FFI_ArrowArrayStream;


This needs to be updated. Also, similar to export_array_into_raw in FFI Array, do we need to have a function to allow an importer to allocate struct memory for the exported stream?

I added export_reader_into_raw and used it in test and doc now.

sunchao · 2022-03-29T22:04:42Z

arrow/src/ffi_stream.rs

+    type Item = Result<RecordBatch>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.stream.get_next?;


But I think we checked in the constructors to make sure it is not released?

sunchao · 2022-03-29T22:06:46Z

arrow/src/ffi_stream.rs

+/// that requires [FFI_ArrowArrayStream].
+#[derive(Debug)]
+pub struct ArrowArrayStreamReader {
+    stream: Arc<FFI_ArrowArrayStream>,


nit: we can use Box here to indicate that stream is an unique reference, not shared.

To make it Box, it will be in trouble when we want to get raw pointer back. Because Box::into_raw(self.stream) will be complained self.stream cannot be moved, and we cannot clone this Box (FFI_ArrowArrayStream is not clonable).

sunchao

Overall looks pretty good! just a few last comments.

arrow/src/ffi_stream.rs

sunchao · 2022-03-30T21:03:18Z

arrow/src/ffi_stream.rs

+    type Item = Result<RecordBatch>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.stream.get_next.unwrap();


this seems unnecessary since by contract get_next should be defined. Plus we also call unwrap at line 368.

yea, this can be removed.

sunchao · 2022-03-30T21:03:44Z

arrow/src/ffi_stream.rs

+    fn next(&mut self) -> Option<Self::Item> {
+        self.stream.get_next.unwrap();
+
+        let stream_ptr = Arc::into_raw(self.stream.clone()) as *mut FFI_ArrowArrayStream;


I think we can just use Arc::as_ptr instead of clone here.

sunchao · 2022-03-30T21:04:34Z

arrow/src/ffi_stream.rs

+            let record_batch = RecordBatch::from(&StructArray::from(data));
+
+            Some(Ok(record_batch))
+        } else {


we need to drop array_ptr in the else branch too?

sunchao

LGTM

viirya · 2022-03-31T00:39:26Z

Thanks @sunchao

sunchao · 2022-03-31T01:07:15Z

Merged, thanks @viirya !

alamb · 2022-03-31T19:14:05Z

This is epic work @viirya and @sunchao -- thank you so much.

Add FFI for Arrow C Stream Interface

13dad80

viirya marked this pull request as draft March 2, 2022 09:25

github-actions bot added the arrow Changes to the arrow crate label Mar 2, 2022

viirya added 4 commits March 2, 2022 12:55

Add ArrowArrayStreamReader

4d4cb7a

Add test

5570fc2

Fix clippy

b256c91

fix format

415146f

viirya marked this pull request as ready for review March 2, 2022 23:03

define error code

62bf805

alamb reviewed Mar 3, 2022

View reviewed changes

Regenerate ffi binding using bindgen

79f5d8a

viirya added 4 commits March 6, 2022 15:01

Rewrite test

2623b33

Remove CStreamInterface

a9bc94c

Merge remote-tracking branch 'upstream/master' into issue_1348

2560e79

Fix clippy error

083e014

Fix more clippy errors

b222a44

sunchao reviewed Mar 8, 2022

View reviewed changes

viirya added 4 commits March 8, 2022 13:19

For review comment.

e029f61

Fix clippy error

5a35a8e

Fix clippy error

4b3c91b

not run example code in comment

99033c1

ignore doctest

78444b1

sunchao reviewed Mar 17, 2022

View reviewed changes

viirya added 2 commits March 17, 2022 12:06

For review

b0a6ac9

Fix clippy

b6efc8d

sunchao reviewed Mar 18, 2022

View reviewed changes

viirya added 2 commits March 23, 2022 16:29

Merge remote-tracking branch 'upstream/master' into issue_1348

4385684

For review comment

a162c84

sunchao reviewed Mar 29, 2022

View reviewed changes

viirya added 2 commits March 30, 2022 01:01

For review

35da4f1

Add export_reader_into_raw

9fad867

sunchao reviewed Mar 30, 2022

View reviewed changes

For review

3716c16

sunchao approved these changes Mar 30, 2022

View reviewed changes

sunchao merged commit 15c87ae into apache:master Mar 31, 2022


		let stream_data = std::ptr::replace(raw_stream, FFI_ArrowArrayStream::empty());

		let stream = Arc::new(stream_data);

Add FFI for Arrow C Stream Interface #1384

Add FFI for Arrow C Stream Interface #1384

Conversation

viirya commented Mar 2, 2022

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are there any user-facing changes?

viirya commented Mar 3, 2022

alamb left a 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

alamb commented Mar 6, 2022

viirya commented Mar 6, 2022

codecov-commenter commented Mar 7, 2022 • edited Loading

Codecov Report

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

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

sunchao commented Mar 8, 2022

viirya commented Mar 13, 2022

sunchao commented Mar 14, 2022

viirya commented Mar 14, 2022

alamb commented Mar 16, 2022

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

alamb commented Mar 23, 2022

viirya commented Mar 23, 2022

sunchao commented Mar 23, 2022

viirya commented Mar 23, 2022

alamb commented Mar 28, 2022

viirya commented Mar 28, 2022

sunchao commented Mar 29, 2022

Choose a reason for hiding this comment

viirya Mar 30, 2022 • edited Loading

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

sunchao left a 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

sunchao left a comment

Choose a reason for hiding this comment

codecov-commenter commented Mar 7, 2022 •

edited

Loading

viirya Mar 30, 2022 •

edited

Loading