Optionally require alignment when reading IPC, respect alignment when writing

arrow-buffer/src/buffer/scalar.rs

@@ -63,9 +63,15 @@ impl<T: ArrowNativeType> ScalarBuffer<T> {
     /// This method will panic if
     ///
     /// * `offset` or `len` would result in overflow
-    /// * `buffer` is not aligned to a multiple of `std::mem::size_of::<T>`
+    /// * `buffer` is not aligned to a multiple of `std::mem::align_of::<T>`
     /// * `bytes` is not large enough for the requested slice
     pub fn new(buffer: Buffer, offset: usize, len: usize) -> Self {
+        assert_eq!(
+            buffer.as_ptr().align_offset(std::mem::align_of::<T>()),
+            0,
+            "buffer is not aligned"
+        );
+
         let size = std::mem::size_of::<T>();
         let byte_offset = offset.checked_mul(size).expect("offset overflow");
         let byte_len = len.checked_mul(size).expect("length overflow");

arrow-flight/src/decode.rs

@@ -308,6 +308,7 @@ impl FlightDataDecoder {
                     &state.schema,
                     &mut state.dictionaries_by_field,
                     &message.version(),
+                    false,
                 )
                 .map_err(|e| {
                     FlightError::DecodeError(format!("Error decoding ipc dictionary: {e}"))

arrow-flight/src/sql/client.rs

@@ -590,6 +590,7 @@ pub fn arrow_data_from_flight_data(
                 &dictionaries_by_field,
                 None,
                 &ipc_message.version(),
+                false,
             )?;
             Ok(ArrowFlightData::RecordBatch(record_batch))
         }

arrow-flight/src/utils.rs

@@ -99,6 +99,7 @@ pub fn flight_data_to_arrow_batch(
                 dictionaries_by_id,
                 None,
                 &message.version(),
+                false,
             )
         })?
 }

arrow-integration-testing/src/flight_client_scenarios/integration_test.rs

@@ -269,6 +269,7 @@ async fn receive_batch_flight_data(
             &schema,
             dictionaries_by_id,
             &message.version(),
+            false,
         )
         .expect("Error reading dictionary");
 

arrow-integration-testing/src/flight_server_scenarios/integration_test.rs

@@ -308,6 +308,7 @@ async fn record_batch_from_message(
         dictionaries_by_id,
         None,
         &message.version(),
+        false,
     );
 
     arrow_batch_result
@@ -330,6 +331,7 @@ async fn dictionary_from_message(
         &schema_ref,
         dictionaries_by_id,
         &message.version(),
+        false,
     );
     dictionary_batch_result
         .map_err(|e| Status::internal(format!("Could not convert to Dictionary: {e:?}")))