This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-11422-85a95757297b392f4ce9607656eb3d4a67a106e6 in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
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 =
