This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 6c31f419a2 fix(ffi): report missing C stream callbacks (#11422)
6c31f419a2 is described below
commit 6c31f419a211531fa29e9920374cc02a20b9c2a0
Author: Dhruv Dagar <[email protected]>
AuthorDate: Thu Oct 8 03:54:35 2026 +0000
fix(ffi): report missing C stream callbacks (#11422)
# Which issue does this PR close?
- Closes #11388
# Rationale for this change
Malformed or incomplete Arrow C Stream Interface inputs can cause
`ArrowArrayStreamReader` to panic when required callbacks are missing.
In particular, the absence of the `get_schema` or `get_next` callback
currently results in an `Option::unwrap()` panic instead of being
reported as an error.
The reader should handle these invalid inputs gracefully by returning an
appropriate `CDataInterface` error.
# What changes are included in this PR?
- Handle a missing `get_schema` callback in `get_stream_schema` by
returning a `CDataInterface` error.
- Handle a missing `get_next` callback in `ArrowArrayStreamReader::next`
by returning a `CDataInterface` error.
- Add regression tests covering streams with missing `get_schema` and
`get_next` callbacks.
# Are these changes tested?
Regression tests were added for both missing callback cases. The
existing FFI stream tests continue to pass.
# Are there any significant user-facing changes?
No significant user-facing changes. This change replaces a panic caused
by malformed C Stream Interface inputs with a proper `CDataInterface`
error.
---
arrow-array/src/ffi_stream.rs | 57 ++++++++++++++++++++++++++++++++++++++++---
1 file changed, 54 insertions(+), 3 deletions(-)
diff --git a/arrow-array/src/ffi_stream.rs b/arrow-array/src/ffi_stream.rs
index 32ed6526ec..3e06e489ce 100644
--- a/arrow-array/src/ffi_stream.rs
+++ b/arrow-array/src/ffi_stream.rs
@@ -440,7 +440,13 @@ unsafe fn producer_error(stream_ptr: *mut
FFI_ArrowArrayStream) -> Option<String
fn get_stream_schema(stream_ptr: *mut FFI_ArrowArrayStream) ->
Result<SchemaRef> {
let mut schema = FFI_ArrowSchema::empty();
- let ret_code = unsafe { (*stream_ptr).get_schema.unwrap()(stream_ptr, &raw
mut schema) };
+ let Some(get_schema) = (unsafe { (*stream_ptr).get_schema }) else {
+ return Err(ArrowError::CDataInterface(
+ "input stream does not provide the required get_schema
callback".to_string(),
+ ));
+ };
+
+ let ret_code = unsafe { get_schema(stream_ptr, &raw mut schema) };
if ret_code == 0 {
let schema = Schema::try_from(&schema)?;
@@ -493,8 +499,13 @@ impl Iterator for ArrowArrayStreamReader {
fn next(&mut self) -> Option<Self::Item> {
let mut array = FFI_ArrowArray::empty();
- let ret_code =
- unsafe { self.stream.get_next.unwrap()(&raw mut self.stream, &raw
mut array) };
+ let Some(get_next) = self.stream.get_next else {
+ return Some(Err(ArrowError::CDataInterface(
+ "input stream does not provide the required get_next
callback".to_string(),
+ )));
+ };
+
+ let ret_code = unsafe { get_next(&raw mut self.stream, &raw mut array)
};
if ret_code == 0 {
// The end of stream has been reached
@@ -755,6 +766,46 @@ mod tests {
stream
}
+ #[test]
+ fn test_import_stream_without_get_schema_callback() {
+ // The C Stream Interface requires get_schema for an unreleased stream.
+ let stream = unsafe {
+ FFI_ArrowArrayStream::new_unchecked(
+ None,
+ Some(failing_get_next),
+ None,
+ Some(mark_released),
+ std::ptr::null_mut(),
+ )
+ };
+
+ let err = ArrowArrayStreamReader::try_new(stream).unwrap_err();
+ assert!(matches!(err, ArrowError::CDataInterface(_)));
+ assert!(err.to_string().contains("get_schema"));
+ }
+
+ #[test]
+ fn test_import_stream_without_get_next_callback() {
+ // The C Stream Interface requires get_next for an unreleased stream.
+ let stream = unsafe {
+ FFI_ArrowArrayStream::new_unchecked(
+ Some(working_get_schema),
+ None,
+ None,
+ Some(mark_released),
+ std::ptr::null_mut(),
+ )
+ };
+
+ let err = ArrowArrayStreamReader::try_new(stream)
+ .unwrap()
+ .next()
+ .unwrap()
+ .unwrap_err();
+ assert!(matches!(err, ArrowError::CDataInterface(_)));
+ assert!(err.to_string().contains("get_next"));
+ }
+
#[test]
fn test_import_schema_error_reports_producer_message() {
let err =
