bugfix: preserve schema metadata for record batch in FFI (#19293)

## Which issue does this PR close?

- Closes #19292.

## Rationale for this change

Metadata on a schema for a record batch is lost during FFI conversion.
This is not always obvious because our other integrations like
`ExecutionPlan` and `TableProvider` have their own ways to provide the
schema.

This is currently an issue because we have FFI table providers who say
they have a specific schema but the schema of the actual record batches
does not match. The metadata is dropped.

## What changes are included in this PR?

We already have the schema in the FFI object, so use it both when
converting to and from FFI.

## Are these changes tested?

Unit test added.

## Are there any user-facing changes?

No

Author: timsaucer

datafusion/ffi/src/record_batch_stream.rs

@@ -105,12 +105,11 @@ unsafe extern "C" fn release_fn_wrapper(provider: &mut FFI_RecordBatchStream) {
 pub(crate) fn record_batch_to_wrapped_array(
     record_batch: RecordBatch,
 ) -> FFIResult<WrappedArray> {
+    let schema = WrappedSchema::from(record_batch.schema());
     let struct_array = StructArray::from(record_batch);
     rresult!(
-        to_ffi(&struct_array.to_data()).map(|(array, schema)| WrappedArray {
-            array,
-            schema: WrappedSchema(schema)
-        })
+        to_ffi(&struct_array.to_data())
+            .map(|(array, _schema)| WrappedArray { array, schema })
     )
 }
 
@@ -157,6 +156,7 @@ impl RecordBatchStream for FFI_RecordBatchStream {
 pub(crate) fn wrapped_array_to_record_batch(array: WrappedArray) -> Result<RecordBatch> {
     let array_data =
         unsafe { from_ffi(array.array, &array.schema.0).map_err(DataFusionError::from)? };
+    let schema: arrow::datatypes::SchemaRef = array.schema.into();
     let array = make_array(array_data);
     let struct_array = array
         .as_any()
@@ -165,7 +165,9 @@ pub(crate) fn wrapped_array_to_record_batch(array: WrappedArray) -> Result<Recor
         "Unexpected array type during record batch collection in FFI_RecordBatchStream - expected StructArray"
     ))?;
 
-    Ok(struct_array.into())
+    let rb: RecordBatch = struct_array.into();
+
+    rb.with_schema(schema).map_err(Into::into)
 }
 
 fn maybe_wrapped_array_to_record_batch(
@@ -219,7 +221,11 @@ mod tests {
     use datafusion::test_util::bounded_stream;
     use futures::StreamExt;
 
-    use super::FFI_RecordBatchStream;
+    use super::{
+        FFI_RecordBatchStream, record_batch_to_wrapped_array,
+        wrapped_array_to_record_batch,
+    };
+    use crate::df_result;
 
     #[tokio::test]
     async fn test_round_trip_record_batch_stream() -> Result<()> {
@@ -252,4 +258,28 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn round_trip_record_batch_with_metadata() -> Result<()> {
+        let rb = record_batch!(
+            ("a", Int32, vec![1, 2, 3]),
+            ("b", Float64, vec![Some(4.0), None, Some(5.0)])
+        )?;
+
+        let schema = rb
+            .schema()
+            .as_ref()
+            .clone()
+            .with_metadata([("some_key".to_owned(), "some_value".to_owned())].into())
+            .into();
+
+        let rb = rb.with_schema(schema)?;
+
+        let ffi_rb = df_result!(record_batch_to_wrapped_array(rb.clone()))?;
+
+        let round_trip_rb = wrapped_array_to_record_batch(ffi_rb)?;
+
+        assert_eq!(rb, round_trip_rb);
+        Ok(())
+    }
 }