feat: deterministic metadata encoding (#7437)

* Apply ordering to metadata keys before encoding

* Add unit test to ensure consistent metadata encoding

* Clippy warning

Author: timsaucer

arrow-ipc/src/convert.rs

@@ -138,9 +138,12 @@ pub fn metadata_to_fb<'a>(
     fbb: &mut FlatBufferBuilder<'a>,
     metadata: &HashMap<String, String>,
 ) -> WIPOffset<Vector<'a, ForwardsUOffset<KeyValue<'a>>>> {
-    let custom_metadata = metadata
-        .iter()
-        .map(|(k, v)| {
+    let mut ordered_keys = metadata.keys().collect::<Vec<_>>();
+    ordered_keys.sort();
+    let custom_metadata = ordered_keys
+        .into_iter()
+        .map(|k| {
+            let v = metadata.get(k).unwrap();
             let fb_key_name = fbb.create_string(k);
             let fb_val_name = fbb.create_string(v);
 

arrow-ipc/src/writer.rs

@@ -1854,6 +1854,7 @@ fn pad_to_alignment(alignment: u8, len: usize) -> usize {
 
 #[cfg(test)]
 mod tests {
+    use std::hash::Hasher;
     use std::io::Cursor;
     use std::io::Seek;
 
@@ -3306,4 +3307,48 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_metadata_encoding_ordering() {
+        fn create_hash() -> u64 {
+            let metadata: HashMap<String, String> = [
+                ("a", "1"), //
+                ("b", "2"), //
+                ("c", "3"), //
+                ("d", "4"), //
+                ("e", "5"), //
+            ]
+            .into_iter()
+            .map(|(k, v)| (k.to_owned(), v.to_owned()))
+            .collect();
+
+            // Set metadata on both the schema and a field within it.
+            let schema = Arc::new(
+                Schema::new(vec![
+                    Field::new("a", DataType::Int64, true).with_metadata(metadata.clone())
+                ])
+                .with_metadata(metadata)
+                .clone(),
+            );
+            let batch = RecordBatch::new_empty(schema.clone());
+
+            let mut bytes = Vec::new();
+            let mut w = StreamWriter::try_new(&mut bytes, batch.schema_ref()).unwrap();
+            w.write(&batch).unwrap();
+            w.finish().unwrap();
+
+            let mut h = std::hash::DefaultHasher::new();
+            h.write(&bytes);
+            h.finish()
+        }
+
+        let expected = create_hash();
+
+        // Since there is randomness in the HashMap and we cannot specify our
+        // own Hasher for the implementation used for metadata, run the above
+        // code 20x and verify it does not change. This is not perfect but it
+        // should be good enough.
+        let all_passed = (0..20).all(|_| create_hash() == expected);
+        assert!(all_passed);
+    }
 }