Changed async ipc writer to accept schema by value (#1439)

src/io/ipc/write/file_async.rs

@@ -37,7 +37,7 @@ type WriteOutput<W> = (usize, Option<Block>, Vec<Block>, Option<W>);
 /// let mut buffer = Cursor::new(vec![]);
 /// let mut sink = FileSink::new(
 ///     &mut buffer,
-///     &schema,
+///     schema,
 ///     None,
 ///     Default::default(),
 /// );
@@ -78,13 +78,13 @@ where
     /// Create a new file writer.
     pub fn new(
         writer: W,
-        schema: &Schema,
+        schema: Schema,
         ipc_fields: Option<Vec<IpcField>>,
         options: WriteOptions,
     ) -> Self {
         let fields = ipc_fields.unwrap_or_else(|| default_ipc_fields(&schema.fields));
         let encoded = EncodedData {
-            ipc_message: schema_to_bytes(schema, &fields),
+            ipc_message: schema_to_bytes(&schema, &fields),
             arrow_data: vec![],
         };
         let task = Some(Self::start(writer, encoded).boxed());
@@ -94,7 +94,7 @@ where
             options,
             fields,
             offset: 0,
-            schema: schema.clone(),
+            schema,
             dictionary_tracker: DictionaryTracker {
                 dictionaries: Default::default(),
                 cannot_replace: true,

tests/it/io/ipc/write_file_async.rs

@@ -22,7 +22,12 @@ async fn write_(
     let mut result = AsyncCursor::new(vec![]);
 
     let options = WriteOptions { compression: None };
-    let mut sink = FileSink::new(&mut result, schema, Some(ipc_fields.to_vec()), options);
+    let mut sink = FileSink::new(
+        &mut result,
+        schema.clone(),
+        Some(ipc_fields.to_vec()),
+        options,
+    );
     for batch in batches {
         sink.feed((batch, Some(ipc_fields)).into()).await?;
     }