diff --git a/datafusion/physical-plan/src/aggregates/topk/heap.rs b/datafusion/physical-plan/src/aggregates/topk/heap.rs
index b202f812c6e1..8b4b07d211a0 100644
--- a/datafusion/physical-plan/src/aggregates/topk/heap.rs
+++ b/datafusion/physical-plan/src/aggregates/topk/heap.rs
@@ -485,6 +485,8 @@ pub fn new_heap(
 
 #[cfg(test)]
 mod tests {
+    use insta::assert_snapshot;
+
     use super::*;
 
     #[test]
@@ -494,10 +496,9 @@ mod tests {
         heap.append_or_replace(1, 1, &mut map);
 
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=1 idx=0, bucket=1
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
 
         Ok(())
     }
@@ -514,11 +515,10 @@ val=1 idx=0, bucket=1
         assert_eq!(map, vec![(2, 0), (1, 1)]);
 
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=2 idx=0, bucket=2
 └── val=1 idx=1, bucket=1
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
 
         Ok(())
     }
@@ -532,22 +532,20 @@ val=2 idx=0, bucket=2
         heap.append_or_replace(2, 2, &mut map);
         heap.append_or_replace(3, 3, &mut map);
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=3 idx=0, bucket=3
 ├── val=1 idx=1, bucket=1
 └── val=2 idx=2, bucket=2
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
 
         let mut map = vec![];
         heap.append_or_replace(0, 0, &mut map);
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=2 idx=0, bucket=2
 ├── val=1 idx=1, bucket=1
 └── val=0 idx=2, bucket=0
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
         assert_eq!(map, vec![(2, 0), (0, 2)]);
 
         Ok(())
@@ -563,24 +561,22 @@ val=2 idx=0, bucket=2
         heap.append_or_replace(3, 3, &mut map);
         heap.append_or_replace(4, 4, &mut map);
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=4 idx=0, bucket=4
 ├── val=3 idx=1, bucket=3
 │   └── val=1 idx=3, bucket=1
 └── val=2 idx=2, bucket=2
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
 
         let mut map = vec![];
         heap.replace_if_better(1, 0, &mut map);
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=4 idx=0, bucket=4
 ├── val=1 idx=1, bucket=1
 │   └── val=0 idx=3, bucket=3
 └── val=2 idx=2, bucket=2
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
         assert_eq!(map, vec![(1, 1), (3, 3)]);
 
         Ok(())
@@ -595,11 +591,10 @@ val=4 idx=0, bucket=4
         heap.append_or_replace(2, 2, &mut map);
 
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=2 idx=0, bucket=2
 └── val=1 idx=1, bucket=1
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
 
         assert_eq!(heap.worst_val(), Some(&2));
         assert_eq!(heap.worst_map_idx(), 2);
@@ -616,11 +611,10 @@ val=2 idx=0, bucket=2
         heap.append_or_replace(2, 2, &mut map);
 
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=2 idx=0, bucket=2
 └── val=1 idx=1, bucket=1
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
 
         let (vals, map_idxs) = heap.drain();
         assert_eq!(vals, vec![1, 2]);
@@ -639,20 +633,18 @@ val=2 idx=0, bucket=2
         heap.append_or_replace(2, 2, &mut map);
 
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=2 idx=0, bucket=2
 └── val=1 idx=1, bucket=1
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
 
         let numbers = vec![(0, 1), (1, 2)];
         heap.renumber(numbers.as_slice());
         let actual = heap.to_string();
-        let expected = r#"
+        assert_snapshot!(actual, @r#"
 val=2 idx=0, bucket=1
 └── val=1 idx=1, bucket=2
-        "#;
-        assert_eq!(actual.trim(), expected.trim());
+            "#);
 
         Ok(())
     }
diff --git a/datafusion/physical-plan/src/joins/nested_loop_join.rs b/datafusion/physical-plan/src/joins/nested_loop_join.rs
index 88d3ea9e7e1e..cdd2eaeca899 100644
--- a/datafusion/physical-plan/src/joins/nested_loop_join.rs
+++ b/datafusion/physical-plan/src/joins/nested_loop_join.rs
@@ -1047,13 +1047,15 @@ pub(crate) mod tests {
     use arrow::array::Int32Array;
     use arrow::compute::SortOptions;
     use arrow::datatypes::{DataType, Field};
-    use datafusion_common::{assert_batches_sorted_eq, assert_contains, ScalarValue};
+    use datafusion_common::test_util::batches_to_sort_string;
+    use datafusion_common::{assert_contains, ScalarValue};
     use datafusion_execution::runtime_env::RuntimeEnvBuilder;
     use datafusion_expr::Operator;
     use datafusion_physical_expr::expressions::{BinaryExpr, Literal};
     use datafusion_physical_expr::{Partitioning, PhysicalExpr};
     use datafusion_physical_expr_common::sort_expr::{LexOrdering, PhysicalSortExpr};
 
+    use insta::assert_snapshot;
     use rstest::rstest;
 
     fn build_table(
@@ -1216,15 +1218,13 @@ pub(crate) mod tests {
         )
         .await?;
         assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 5  | 5  | 50 | 2  | 2  | 80 |",
-            "+----+----+----+----+----+----+",
-        ];
-
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b2 | c2 |
+            +----+----+----+----+----+----+
+            | 5  | 5  | 50 | 2  | 2  | 80 |
+            +----+----+----+----+----+----+
+            "#);
 
         Ok(())
     }
@@ -1245,17 +1245,15 @@ pub(crate) mod tests {
         )
         .await?;
         assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);
-        let expected = [
-            "+----+----+-----+----+----+----+",
-            "| a1 | b1 | c1  | a2 | b2 | c2 |",
-            "+----+----+-----+----+----+----+",
-            "| 11 | 8  | 110 |    |    |    |",
-            "| 5  | 5  | 50  | 2  | 2  | 80 |",
-            "| 9  | 8  | 90  |    |    |    |",
-            "+----+----+-----+----+----+----+",
-        ];
-
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+-----+----+----+----+
+            | a1 | b1 | c1  | a2 | b2 | c2 |
+            +----+----+-----+----+----+----+
+            | 11 | 8  | 110 |    |    |    |
+            | 5  | 5  | 50  | 2  | 2  | 80 |
+            | 9  | 8  | 90  |    |    |    |
+            +----+----+-----+----+----+----+
+            "#);
 
         Ok(())
     }
@@ -1276,17 +1274,15 @@ pub(crate) mod tests {
         )
         .await?;
         assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);
-        let expected = [
-            "+----+----+----+----+----+-----+",
-            "| a1 | b1 | c1 | a2 | b2 | c2  |",
-            "+----+----+----+----+----+-----+",
-            "|    |    |    | 10 | 10 | 100 |",
-            "|    |    |    | 12 | 10 | 40  |",
-            "| 5  | 5  | 50 | 2  | 2  | 80  |",
-            "+----+----+----+----+----+-----+",
-        ];
-
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+----+----+----+-----+
+            | a1 | b1 | c1 | a2 | b2 | c2  |
+            +----+----+----+----+----+-----+
+            |    |    |    | 10 | 10 | 100 |
+            |    |    |    | 12 | 10 | 40  |
+            | 5  | 5  | 50 | 2  | 2  | 80  |
+            +----+----+----+----+----+-----+
+            "#);
 
         Ok(())
     }
@@ -1307,19 +1303,17 @@ pub(crate) mod tests {
         )
         .await?;
         assert_eq!(columns, vec!["a1", "b1", "c1", "a2", "b2", "c2"]);
-        let expected = [
-            "+----+----+-----+----+----+-----+",
-            "| a1 | b1 | c1  | a2 | b2 | c2  |",
-            "+----+----+-----+----+----+-----+",
-            "|    |    |     | 10 | 10 | 100 |",
-            "|    |    |     | 12 | 10 | 40  |",
-            "| 11 | 8  | 110 |    |    |     |",
-            "| 5  | 5  | 50  | 2  | 2  | 80  |",
-            "| 9  | 8  | 90  |    |    |     |",
-            "+----+----+-----+----+----+-----+",
-        ];
-
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+-----+----+----+-----+
+            | a1 | b1 | c1  | a2 | b2 | c2  |
+            +----+----+-----+----+----+-----+
+            |    |    |     | 10 | 10 | 100 |
+            |    |    |     | 12 | 10 | 40  |
+            | 11 | 8  | 110 |    |    |     |
+            | 5  | 5  | 50  | 2  | 2  | 80  |
+            | 9  | 8  | 90  |    |    |     |
+            +----+----+-----+----+----+-----+
+            "#);
 
         Ok(())
     }
@@ -1340,15 +1334,13 @@ pub(crate) mod tests {
         )
         .await?;
         assert_eq!(columns, vec!["a1", "b1", "c1"]);
-        let expected = [
-            "+----+----+----+",
-            "| a1 | b1 | c1 |",
-            "+----+----+----+",
-            "| 5  | 5  | 50 |",
-            "+----+----+----+",
-        ];
-
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+----+
+            | a1 | b1 | c1 |
+            +----+----+----+
+            | 5  | 5  | 50 |
+            +----+----+----+
+            "#);
 
         Ok(())
     }
@@ -1369,16 +1361,14 @@ pub(crate) mod tests {
         )
         .await?;
         assert_eq!(columns, vec!["a1", "b1", "c1"]);
-        let expected = [
-            "+----+----+-----+",
-            "| a1 | b1 | c1  |",
-            "+----+----+-----+",
-            "| 11 | 8  | 110 |",
-            "| 9  | 8  | 90  |",
-            "+----+----+-----+",
-        ];
-
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+-----+
+            | a1 | b1 | c1  |
+            +----+----+-----+
+            | 11 | 8  | 110 |
+            | 9  | 8  | 90  |
+            +----+----+-----+
+            "#);
 
         Ok(())
     }
@@ -1399,15 +1389,13 @@ pub(crate) mod tests {
         )
         .await?;
         assert_eq!(columns, vec!["a2", "b2", "c2"]);
-        let expected = [
-            "+----+----+----+",
-            "| a2 | b2 | c2 |",
-            "+----+----+----+",
-            "| 2  | 2  | 80 |",
-            "+----+----+----+",
-        ];
-
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+----+
+            | a2 | b2 | c2 |
+            +----+----+----+
+            | 2  | 2  | 80 |
+            +----+----+----+
+            "#);
 
         Ok(())
     }
@@ -1428,16 +1416,14 @@ pub(crate) mod tests {
         )
         .await?;
         assert_eq!(columns, vec!["a2", "b2", "c2"]);
-        let expected = [
-            "+----+----+-----+",
-            "| a2 | b2 | c2  |",
-            "+----+----+-----+",
-            "| 10 | 10 | 100 |",
-            "| 12 | 10 | 40  |",
-            "+----+----+-----+",
-        ];
-
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+-----+
+            | a2 | b2 | c2  |
+            +----+----+-----+
+            | 10 | 10 | 100 |
+            | 12 | 10 | 40  |
+            +----+----+-----+
+            "#);
 
         Ok(())
     }
@@ -1458,17 +1444,15 @@ pub(crate) mod tests {
         )
         .await?;
         assert_eq!(columns, vec!["a1", "b1", "c1", "mark"]);
-        let expected = [
-            "+----+----+-----+-------+",
-            "| a1 | b1 | c1  | mark  |",
-            "+----+----+-----+-------+",
-            "| 11 | 8  | 110 | false |",
-            "| 5  | 5  | 50  | true  |",
-            "| 9  | 8  | 90  | false |",
-            "+----+----+-----+-------+",
-        ];
-
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+-----+-------+
+            | a1 | b1 | c1  | mark  |
+            +----+----+-----+-------+
+            | 11 | 8  | 110 | false |
+            | 5  | 5  | 50  | true  |
+            | 9  | 8  | 90  | false |
+            +----+----+-----+-------+
+            "#);
 
         Ok(())
     }
diff --git a/datafusion/physical-plan/src/joins/sort_merge_join.rs b/datafusion/physical-plan/src/joins/sort_merge_join.rs
index 7fb8a2d73600..d1442a1be992 100644
--- a/datafusion/physical-plan/src/joins/sort_merge_join.rs
+++ b/datafusion/physical-plan/src/joins/sort_merge_join.rs
@@ -2294,8 +2294,8 @@ fn fetch_right_columns_from_batch_by_idxs(
                 });
             }
 
-            Ok(buffered_cols)
-        }
+                Ok(buffered_cols)
+            }
         // Invalid combination
         (spill, batch) => internal_err!("Unexpected buffered batch spill status. Spill exists: {}. In-memory exists: {}", spill.is_some(), batch.is_some()),
     }
@@ -2547,10 +2547,11 @@ mod tests {
     use arrow::compute::{concat_batches, filter_record_batch, SortOptions};
     use arrow::datatypes::{DataType, Field, Schema};
 
-    use datafusion_common::JoinSide;
     use datafusion_common::JoinType::*;
+    use datafusion_common::{assert_batches_eq, assert_contains, JoinType, Result};
     use datafusion_common::{
-        assert_batches_eq, assert_batches_sorted_eq, assert_contains, JoinType, Result,
+        test_util::{batches_to_sort_string, batches_to_string},
+        JoinSide,
     };
     use datafusion_execution::config::SessionConfig;
     use datafusion_execution::disk_manager::DiskManagerConfig;
@@ -2558,6 +2559,7 @@ mod tests {
     use datafusion_execution::TaskContext;
     use datafusion_expr::Operator;
     use datafusion_physical_expr::expressions::BinaryExpr;
+    use insta::{allow_duplicates, assert_snapshot};
 
     use crate::expressions::Column;
     use crate::joins::sort_merge_join::{get_corrected_filter_mask, JoinedRecordBatches};
@@ -2803,17 +2805,16 @@ mod tests {
 
         let (_, batches) = join_collect(left, right, on, Inner).await?;
 
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b1 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 1  | 4  | 7  | 10 | 4  | 70 |",
-            "| 2  | 5  | 8  | 20 | 5  | 80 |",
-            "| 3  | 5  | 9  | 20 | 5  | 80 |",
-            "+----+----+----+----+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b1 | c2 |
+            +----+----+----+----+----+----+
+            | 1  | 4  | 7  | 10 | 4  | 70 |
+            | 2  | 5  | 8  | 20 | 5  | 80 |
+            | 3  | 5  | 9  | 20 | 5  | 80 |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -2841,17 +2842,17 @@ mod tests {
         ];
 
         let (_columns, batches) = join_collect(left, right, on, Inner).await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b2 | c1 | a1 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 1  | 1  | 7  | 1  | 1  | 70 |",
-            "| 2  | 2  | 8  | 2  | 2  | 80 |",
-            "| 2  | 2  | 9  | 2  | 2  | 80 |",
-            "+----+----+----+----+----+----+",
-        ];
+
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b2 | c1 | a1 | b2 | c2 |
+            +----+----+----+----+----+----+
+            | 1  | 1  | 7  | 1  | 1  | 70 |
+            | 2  | 2  | 8  | 2  | 2  | 80 |
+            | 2  | 2  | 9  | 2  | 2  | 80 |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -2879,18 +2880,18 @@ mod tests {
         ];
 
         let (_columns, batches) = join_collect(left, right, on, Inner).await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b2 | c1 | a1 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 1  | 1  | 7  | 1  | 1  | 70 |",
-            "| 1  | 1  | 7  | 1  | 1  | 80 |",
-            "| 1  | 1  | 8  | 1  | 1  | 70 |",
-            "| 1  | 1  | 8  | 1  | 1  | 80 |",
-            "+----+----+----+----+----+----+",
-        ];
+
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b2 | c1 | a1 | b2 | c2 |
+            +----+----+----+----+----+----+
+            | 1  | 1  | 7  | 1  | 1  | 70 |
+            | 1  | 1  | 7  | 1  | 1  | 80 |
+            | 1  | 1  | 8  | 1  | 1  | 70 |
+            | 1  | 1  | 8  | 1  | 1  | 80 |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -2918,17 +2919,16 @@ mod tests {
         ];
 
         let (_, batches) = join_collect(left, right, on, Inner).await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b2 | c1 | a1 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 1  | 1  |    | 1  | 1  | 70 |",
-            "| 2  | 2  | 8  | 2  | 2  | 80 |",
-            "| 2  | 2  | 9  | 2  | 2  | 80 |",
-            "+----+----+----+----+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b2 | c1 | a1 | b2 | c2 |
+            +----+----+----+----+----+----+
+            | 1  | 1  |    | 1  | 1  | 70 |
+            | 2  | 2  | 8  | 2  | 2  | 80 |
+            | 2  | 2  | 9  | 2  | 2  | 80 |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -2969,18 +2969,17 @@ mod tests {
             true,
         )
         .await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b2 | c1 | a1 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 2  | 2  | 9  | 2  | 2  | 80 |",
-            "| 2  | 2  | 8  | 2  | 2  | 80 |",
-            "| 1  | 1  |    | 1  | 1  | 70 |",
-            "| 1  |    | 1  | 1  |    | 10 |",
-            "+----+----+----+----+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b2 | c1 | a1 | b2 | c2 |
+            +----+----+----+----+----+----+
+            | 2  | 2  | 9  | 2  | 2  | 80 |
+            | 2  | 2  | 8  | 2  | 2  | 80 |
+            | 1  | 1  |    | 1  | 1  | 70 |
+            | 1  |    | 1  | 1  |    | 10 |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3009,20 +3008,19 @@ mod tests {
 
         let (_, batches) =
             join_collect_batch_size_equals_two(left, right, on, Inner).await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b2 | c1 | a1 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 1  | 1  | 7  | 1  | 1  | 70 |",
-            "| 2  | 2  | 8  | 2  | 2  | 80 |",
-            "| 2  | 2  | 9  | 2  | 2  | 80 |",
-            "+----+----+----+----+----+----+",
-        ];
         assert_eq!(batches.len(), 2);
         assert_eq!(batches[0].num_rows(), 2);
         assert_eq!(batches[1].num_rows(), 1);
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b2 | c1 | a1 | b2 | c2 |
+            +----+----+----+----+----+----+
+            | 1  | 1  | 7  | 1  | 1  | 70 |
+            | 2  | 2  | 8  | 2  | 2  | 80 |
+            | 2  | 2  | 9  | 2  | 2  | 80 |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3044,17 +3042,16 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, Left).await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b1 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 1  | 4  | 7  | 10 | 4  | 70 |",
-            "| 2  | 5  | 8  | 20 | 5  | 80 |",
-            "| 3  | 7  | 9  |    |    |    |",
-            "+----+----+----+----+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b1 | c2 |
+            +----+----+----+----+----+----+
+            | 1  | 4  | 7  | 10 | 4  | 70 |
+            | 2  | 5  | 8  | 20 | 5  | 80 |
+            | 3  | 7  | 9  |    |    |    |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3076,17 +3073,16 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, Right).await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b1 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 1  | 4  | 7  | 10 | 4  | 70 |",
-            "| 2  | 5  | 8  | 20 | 5  | 80 |",
-            "|    |    |    | 30 | 6  | 90 |",
-            "+----+----+----+----+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b1 | c2 |
+            +----+----+----+----+----+----+
+            | 1  | 4  | 7  | 10 | 4  | 70 |
+            | 2  | 5  | 8  | 20 | 5  | 80 |
+            |    |    |    | 30 | 6  | 90 |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3108,17 +3104,17 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, Full).await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "|    |    |    | 30 | 6  | 90 |",
-            "| 1  | 4  | 7  | 10 | 4  | 70 |",
-            "| 2  | 5  | 8  | 20 | 5  | 80 |",
-            "| 3  | 7  | 9  |    |    |    |",
-            "+----+----+----+----+----+----+",
-        ];
-        assert_batches_sorted_eq!(expected, &batches);
+        // The output order is important as SMJ preserves sortedness
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b2 | c2 |
+            +----+----+----+----+----+----+
+            |    |    |    | 30 | 6  | 90 |
+            | 1  | 4  | 7  | 10 | 4  | 70 |
+            | 2  | 5  | 8  | 20 | 5  | 80 |
+            | 3  | 7  | 9  |    |    |    |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3140,16 +3136,16 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, LeftAnti).await?;
-        let expected = [
-            "+----+----+----+",
-            "| a1 | b1 | c1 |",
-            "+----+----+----+",
-            "| 3  | 7  | 9  |",
-            "| 5  | 7  | 11 |",
-            "+----+----+----+",
-        ];
+
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+
+            | a1 | b1 | c1 |
+            +----+----+----+
+            | 3  | 7  | 9  |
+            | 5  | 7  | 11 |
+            +----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3168,15 +3164,14 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, RightAnti).await?;
-        let expected = [
-            "+----+----+",
-            "| a2 | b1 |",
-            "+----+----+",
-            "| 30 | 6  |",
-            "+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+
+            | a2 | b1 |
+            +----+----+
+            | 30 | 6  |
+            +----+----+
+            "#);
 
         let left2 = build_table(
             ("a1", &vec![1, 2, 2]),
@@ -3195,15 +3190,14 @@ mod tests {
         )];
 
         let (_, batches2) = join_collect(left2, right2, on, RightAnti).await?;
-        let expected2 = [
-            "+----+----+----+",
-            "| a2 | b1 | c2 |",
-            "+----+----+----+",
-            "| 30 | 6  | 90 |",
-            "+----+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected2, &batches2);
+        assert_snapshot!(batches_to_string(&batches2), @r#"
+            +----+----+----+
+            | a2 | b1 | c2 |
+            +----+----+----+
+            | 30 | 6  | 90 |
+            +----+----+----+
+            "#);
 
         Ok(())
     }
@@ -3229,17 +3223,16 @@ mod tests {
         ];
 
         let (_, batches) = join_collect(left, right, on, RightAnti).await?;
-        let expected = [
-            "+----+----+",
-            "| a2 | b1 |",
-            "+----+----+",
-            "| 10 | 4  |",
-            "| 20 | 5  |",
-            "| 30 | 6  |",
-            "+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+
+            | a2 | b1 |
+            +----+----+
+            | 10 | 4  |
+            | 20 | 5  |
+            | 30 | 6  |
+            +----+----+
+            "#);
 
         let left = build_table(
             ("a1", &vec![1, 2, 2]),
@@ -3316,14 +3309,13 @@ mod tests {
         );
         let (_, batches) =
             join_collect_with_filter(left, right, on, filter, RightAnti).await?;
-        let expected = [
-            "+----+----+----+",
-            "| a1 | b1 | c2 |",
-            "+----+----+----+",
-            "| 1  | 10 | 20 |",
-            "+----+----+----+",
-        ];
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+
+            | a1 | b1 | c2 |
+            +----+----+----+
+            | 1  | 10 | 20 |
+            +----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3351,15 +3343,14 @@ mod tests {
         ];
 
         let (_, batches) = join_collect(left, right, on, RightAnti).await?;
-        let expected = [
-            "+----+----+----+",
-            "| a1 | b1 | c2 |",
-            "+----+----+----+",
-            "| 2  |    | 8  |",
-            "+----+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+
+            | a1 | b1 | c2 |
+            +----+----+----+
+            | 2  |    | 8  |
+            +----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3402,17 +3393,16 @@ mod tests {
         )
         .await?;
 
-        let expected = [
-            "+----+----+----+",
-            "| a1 | b1 | c2 |",
-            "+----+----+----+",
-            "| 3  |    | 9  |",
-            "| 2  | 5  |    |",
-            "| 2  | 5  | 8  |",
-            "+----+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+
+            | a1 | b1 | c2 |
+            +----+----+----+
+            | 3  |    | 9  |
+            | 2  | 5  |    |
+            | 2  | 5  | 8  |
+            +----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3441,19 +3431,18 @@ mod tests {
 
         let (_, batches) =
             join_collect_batch_size_equals_two(left, right, on, LeftAnti).await?;
-        let expected = [
-            "+----+----+----+",
-            "| a1 | b1 | c1 |",
-            "+----+----+----+",
-            "| 1  | 4  | 7  |",
-            "| 2  | 5  | 8  |",
-            "| 2  | 5  | 8  |",
-            "+----+----+----+",
-        ];
         assert_eq!(batches.len(), 2);
         assert_eq!(batches[0].num_rows(), 2);
         assert_eq!(batches[1].num_rows(), 1);
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+
+            | a1 | b1 | c1 |
+            +----+----+----+
+            | 1  | 4  | 7  |
+            | 2  | 5  | 8  |
+            | 2  | 5  | 8  |
+            +----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3475,17 +3464,16 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, LeftSemi).await?;
-        let expected = [
-            "+----+----+----+",
-            "| a1 | b1 | c1 |",
-            "+----+----+----+",
-            "| 1  | 4  | 7  |",
-            "| 2  | 5  | 8  |",
-            "| 2  | 5  | 8  |",
-            "+----+----+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+
+            | a1 | b1 | c1 |
+            +----+----+----+
+            | 1  | 4  | 7  |
+            | 2  | 5  | 8  |
+            | 2  | 5  | 8  |
+            +----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3507,18 +3495,17 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, LeftMark).await?;
-        let expected = [
-            "+----+----+----+-------+",
-            "| a1 | b1 | c1 | mark  |",
-            "+----+----+----+-------+",
-            "| 1  | 4  | 7  | true  |",
-            "| 2  | 5  | 8  | true  |",
-            "| 2  | 5  | 8  | true  |",
-            "| 3  | 7  | 9  | false |",
-            "+----+----+----+-------+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+-------+
+            | a1 | b1 | c1 | mark  |
+            +----+----+----+-------+
+            | 1  | 4  | 7  | true  |
+            | 2  | 5  | 8  | true  |
+            | 2  | 5  | 8  | true  |
+            | 3  | 7  | 9  | false |
+            +----+----+----+-------+
+            "#);
         Ok(())
     }
 
@@ -3541,16 +3528,15 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, Inner).await?;
-        let expected = [
-            "+---+---+---+----+---+----+",
-            "| a | b | c | a  | b | c  |",
-            "+---+---+---+----+---+----+",
-            "| 1 | 4 | 7 | 10 | 1 | 70 |",
-            "| 2 | 5 | 8 | 20 | 2 | 80 |",
-            "+---+---+---+----+---+----+",
-        ];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +---+---+---+----+---+----+
+            | a | b | c | a  | b | c  |
+            +---+---+---+----+---+----+
+            | 1 | 4 | 7 | 10 | 1 | 70 |
+            | 2 | 5 | 8 | 20 | 2 | 80 |
+            +---+---+---+----+---+----+
+            "#);
         Ok(())
     }
 
@@ -3574,15 +3560,16 @@ mod tests {
 
         let (_, batches) = join_collect(left, right, on, Inner).await?;
 
-        let expected = ["+------------+------------+------------+------------+------------+------------+",
-            "| a1         | b1         | c1         | a2         | b1         | c2         |",
-            "+------------+------------+------------+------------+------------+------------+",
-            "| 1970-01-02 | 2022-04-25 | 1970-01-08 | 1970-01-11 | 2022-04-25 | 1970-03-12 |",
-            "| 1970-01-03 | 2022-04-26 | 1970-01-09 | 1970-01-21 | 2022-04-26 | 1970-03-22 |",
-            "| 1970-01-04 | 2022-04-26 | 1970-01-10 | 1970-01-21 | 2022-04-26 | 1970-03-22 |",
-            "+------------+------------+------------+------------+------------+------------+"];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +------------+------------+------------+------------+------------+------------+
+            | a1         | b1         | c1         | a2         | b1         | c2         |
+            +------------+------------+------------+------------+------------+------------+
+            | 1970-01-02 | 2022-04-25 | 1970-01-08 | 1970-01-11 | 2022-04-25 | 1970-03-12 |
+            | 1970-01-03 | 2022-04-26 | 1970-01-09 | 1970-01-21 | 2022-04-26 | 1970-03-22 |
+            | 1970-01-04 | 2022-04-26 | 1970-01-10 | 1970-01-21 | 2022-04-26 | 1970-03-22 |
+            +------------+------------+------------+------------+------------+------------+
+            "#);
         Ok(())
     }
 
@@ -3606,15 +3593,16 @@ mod tests {
 
         let (_, batches) = join_collect(left, right, on, Inner).await?;
 
-        let expected = ["+-------------------------+---------------------+-------------------------+-------------------------+---------------------+-------------------------+",
-            "| a1                      | b1                  | c1                      | a2                      | b1                  | c2                      |",
-            "+-------------------------+---------------------+-------------------------+-------------------------+---------------------+-------------------------+",
-            "| 1970-01-01T00:00:00.001 | 2022-04-23T08:44:01 | 1970-01-01T00:00:00.007 | 1970-01-01T00:00:00.010 | 2022-04-23T08:44:01 | 1970-01-01T00:00:00.070 |",
-            "| 1970-01-01T00:00:00.002 | 2022-04-25T16:17:21 | 1970-01-01T00:00:00.008 | 1970-01-01T00:00:00.030 | 2022-04-25T16:17:21 | 1970-01-01T00:00:00.090 |",
-            "| 1970-01-01T00:00:00.003 | 2022-04-25T16:17:21 | 1970-01-01T00:00:00.009 | 1970-01-01T00:00:00.030 | 2022-04-25T16:17:21 | 1970-01-01T00:00:00.090 |",
-            "+-------------------------+---------------------+-------------------------+-------------------------+---------------------+-------------------------+"];
         // The output order is important as SMJ preserves sortedness
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +-------------------------+---------------------+-------------------------+-------------------------+---------------------+-------------------------+
+            | a1                      | b1                  | c1                      | a2                      | b1                  | c2                      |
+            +-------------------------+---------------------+-------------------------+-------------------------+---------------------+-------------------------+
+            | 1970-01-01T00:00:00.001 | 2022-04-23T08:44:01 | 1970-01-01T00:00:00.007 | 1970-01-01T00:00:00.010 | 2022-04-23T08:44:01 | 1970-01-01T00:00:00.070 |
+            | 1970-01-01T00:00:00.002 | 2022-04-25T16:17:21 | 1970-01-01T00:00:00.008 | 1970-01-01T00:00:00.030 | 2022-04-25T16:17:21 | 1970-01-01T00:00:00.090 |
+            | 1970-01-01T00:00:00.003 | 2022-04-25T16:17:21 | 1970-01-01T00:00:00.009 | 1970-01-01T00:00:00.030 | 2022-04-25T16:17:21 | 1970-01-01T00:00:00.090 |
+            +-------------------------+---------------------+-------------------------+-------------------------+---------------------+-------------------------+
+            "#);
         Ok(())
     }
 
@@ -3636,21 +3624,20 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, Left).await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 0  | 3  | 4  |    |    |    |",
-            "| 1  | 4  | 5  | 10 | 4  | 60 |",
-            "| 2  | 5  | 6  |    |    |    |",
-            "| 3  | 6  | 7  | 20 | 6  | 70 |",
-            "| 3  | 6  | 7  | 30 | 6  | 80 |",
-            "| 4  | 6  | 8  | 20 | 6  | 70 |",
-            "| 4  | 6  | 8  | 30 | 6  | 80 |",
-            "| 5  | 7  | 9  |    |    |    |",
-            "+----+----+----+----+----+----+",
-        ];
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b2 | c2 |
+            +----+----+----+----+----+----+
+            | 0  | 3  | 4  |    |    |    |
+            | 1  | 4  | 5  | 10 | 4  | 60 |
+            | 2  | 5  | 6  |    |    |    |
+            | 3  | 6  | 7  | 20 | 6  | 70 |
+            | 3  | 6  | 7  | 30 | 6  | 80 |
+            | 4  | 6  | 8  | 20 | 6  | 70 |
+            | 4  | 6  | 8  | 30 | 6  | 80 |
+            | 5  | 7  | 9  |    |    |    |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3672,17 +3659,16 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, Right).await?;
-        let expected = [
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "|    |    |    | 0  | 2  | 60 |",
-            "| 1  | 4  | 7  | 10 | 4  | 70 |",
-            "| 2  | 5  | 8  | 20 | 5  | 80 |",
-            "|    |    |    | 30 | 6  | 90 |",
-            "+----+----+----+----+----+----+",
-        ];
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b2 | c2 |
+            +----+----+----+----+----+----+
+            |    |    |    | 0  | 2  | 60 |
+            | 1  | 4  | 7  | 10 | 4  | 70 |
+            | 2  | 5  | 8  | 20 | 5  | 80 |
+            |    |    |    | 30 | 6  | 90 |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3716,22 +3702,21 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, Left).await?;
-        let expected = vec![
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "| 0  | 3  | 4  |    |    |    |",
-            "| 1  | 4  | 5  | 10 | 4  | 60 |",
-            "| 2  | 5  | 6  |    |    |    |",
-            "| 3  | 6  | 7  | 20 | 6  | 70 |",
-            "| 3  | 6  | 7  | 30 | 6  | 80 |",
-            "| 4  | 6  | 8  | 20 | 6  | 70 |",
-            "| 4  | 6  | 8  | 30 | 6  | 80 |",
-            "| 5  | 7  | 9  |    |    |    |",
-            "| 6  | 9  | 9  |    |    |    |",
-            "+----+----+----+----+----+----+",
-        ];
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b2 | c2 |
+            +----+----+----+----+----+----+
+            | 0  | 3  | 4  |    |    |    |
+            | 1  | 4  | 5  | 10 | 4  | 60 |
+            | 2  | 5  | 6  |    |    |    |
+            | 3  | 6  | 7  | 20 | 6  | 70 |
+            | 3  | 6  | 7  | 30 | 6  | 80 |
+            | 4  | 6  | 8  | 20 | 6  | 70 |
+            | 4  | 6  | 8  | 30 | 6  | 80 |
+            | 5  | 7  | 9  |    |    |    |
+            | 6  | 9  | 9  |    |    |    |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3765,22 +3750,21 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, Right).await?;
-        let expected = vec![
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "|    |    |    | 0  | 3  | 4  |",
-            "| 10 | 4  | 60 | 1  | 4  | 5  |",
-            "|    |    |    | 2  | 5  | 6  |",
-            "| 20 | 6  | 70 | 3  | 6  | 7  |",
-            "| 30 | 6  | 80 | 3  | 6  | 7  |",
-            "| 20 | 6  | 70 | 4  | 6  | 8  |",
-            "| 30 | 6  | 80 | 4  | 6  | 8  |",
-            "|    |    |    | 5  | 7  | 9  |",
-            "|    |    |    | 6  | 9  | 9  |",
-            "+----+----+----+----+----+----+",
-        ];
-        assert_batches_eq!(expected, &batches);
+        assert_snapshot!(batches_to_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b2 | c2 |
+            +----+----+----+----+----+----+
+            |    |    |    | 0  | 3  | 4  |
+            | 10 | 4  | 60 | 1  | 4  | 5  |
+            |    |    |    | 2  | 5  | 6  |
+            | 20 | 6  | 70 | 3  | 6  | 7  |
+            | 30 | 6  | 80 | 3  | 6  | 7  |
+            | 20 | 6  | 70 | 4  | 6  | 8  |
+            | 30 | 6  | 80 | 4  | 6  | 8  |
+            |    |    |    | 5  | 7  | 9  |
+            |    |    |    | 6  | 9  | 9  |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -3814,24 +3798,23 @@ mod tests {
         )];
 
         let (_, batches) = join_collect(left, right, on, Full).await?;
-        let expected = vec![
-            "+----+----+----+----+----+----+",
-            "| a1 | b1 | c1 | a2 | b2 | c2 |",
-            "+----+----+----+----+----+----+",
-            "|    |    |    | 0  | 2  | 50 |",
-            "|    |    |    | 40 | 8  | 90 |",
-            "| 0  | 3  | 4  |    |    |    |",
-            "| 1  | 4  | 5  | 10 | 4  | 60 |",
-            "| 2  | 5  | 6  |    |    |    |",
-            "| 3  | 6  | 7  | 20 | 6  | 70 |",
-            "| 3  | 6  | 7  | 30 | 6  | 80 |",
-            "| 4  | 6  | 8  | 20 | 6  | 70 |",
-            "| 4  | 6  | 8  | 30 | 6  | 80 |",
-            "| 5  | 7  | 9  |    |    |    |",
-            "| 6  | 9  | 9  |    |    |    |",
-            "+----+----+----+----+----+----+",
-        ];
-        assert_batches_sorted_eq!(expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +----+----+----+----+----+----+
+            | a1 | b1 | c1 | a2 | b2 | c2 |
+            +----+----+----+----+----+----+
+            |    |    |    | 0  | 2  | 50 |
+            |    |    |    | 40 | 8  | 90 |
+            | 0  | 3  | 4  |    |    |    |
+            | 1  | 4  | 5  | 10 | 4  | 60 |
+            | 2  | 5  | 6  |    |    |    |
+            | 3  | 6  | 7  | 20 | 6  | 70 |
+            | 3  | 6  | 7  | 30 | 6  | 80 |
+            | 4  | 6  | 8  | 20 | 6  | 70 |
+            | 4  | 6  | 8  | 30 | 6  | 80 |
+            | 5  | 7  | 9  |    |    |    |
+            | 6  | 9  | 9  |    |    |    |
+            +----+----+----+----+----+----+
+            "#);
         Ok(())
     }
 
@@ -4431,18 +4414,15 @@ mod tests {
 
         let filtered_rb = filter_record_batch(&output, &corrected_mask)?;
 
-        assert_batches_eq!(
-            &[
-                "+---+----+---+----+",
-                "| a | b  | x | y  |",
-                "+---+----+---+----+",
-                "| 1 | 10 | 1 | 11 |",
-                "| 1 | 11 | 1 | 12 |",
-                "| 1 | 12 | 1 | 13 |",
-                "+---+----+---+----+",
-            ],
-            &[filtered_rb]
-        );
+        assert_snapshot!(batches_to_string(&[filtered_rb]), @r#"
+                +---+----+---+----+
+                | a | b  | x | y  |
+                +---+----+---+----+
+                | 1 | 10 | 1 | 11 |
+                | 1 | 11 | 1 | 12 |
+                | 1 | 12 | 1 | 13 |
+                +---+----+---+----+
+            "#);
 
         // output null rows
 
@@ -4463,17 +4443,14 @@ mod tests {
 
         let null_joined_batch = filter_record_batch(&output, &null_mask)?;
 
-        assert_batches_eq!(
-            &[
-                "+---+----+---+----+",
-                "| a | b  | x | y  |",
-                "+---+----+---+----+",
-                "| 1 | 13 | 1 | 12 |",
-                "| 1 | 14 | 1 | 11 |",
-                "+---+----+---+----+",
-            ],
-            &[null_joined_batch]
-        );
+        assert_snapshot!(batches_to_string(&[null_joined_batch]), @r#"
+                +---+----+---+----+
+                | a | b  | x | y  |
+                +---+----+---+----+
+                | 1 | 13 | 1 | 12 |
+                | 1 | 14 | 1 | 11 |
+                +---+----+---+----+
+            "#);
         Ok(())
     }
 
@@ -4607,18 +4584,15 @@ mod tests {
 
         let filtered_rb = filter_record_batch(&output, &corrected_mask)?;
 
-        assert_batches_eq!(
-            &[
-                "+---+----+---+----+",
-                "| a | b  | x | y  |",
-                "+---+----+---+----+",
-                "| 1 | 10 | 1 | 11 |",
-                "| 1 | 11 | 1 | 12 |",
-                "| 1 | 12 | 1 | 13 |",
-                "+---+----+---+----+",
-            ],
-            &[filtered_rb]
-        );
+        assert_snapshot!(batches_to_string(&[filtered_rb]), @r#"
+                +---+----+---+----+
+                | a | b  | x | y  |
+                +---+----+---+----+
+                | 1 | 10 | 1 | 11 |
+                | 1 | 11 | 1 | 12 |
+                | 1 | 12 | 1 | 13 |
+                +---+----+---+----+
+            "#);
 
         // output null rows
         let null_mask = arrow::compute::not(&corrected_mask)?;
@@ -4638,15 +4612,12 @@ mod tests {
 
         let null_joined_batch = filter_record_batch(&output, &null_mask)?;
 
-        assert_batches_eq!(
-            &[
-                "+---+---+---+---+",
-                "| a | b | x | y |",
-                "+---+---+---+---+",
-                "+---+---+---+---+",
-            ],
-            &[null_joined_batch]
-        );
+        assert_snapshot!(batches_to_string(&[null_joined_batch]), @r#"
+                +---+---+---+---+
+                | a | b | x | y |
+                +---+---+---+---+
+                +---+---+---+---+
+            "#);
         Ok(())
     }
 
@@ -4781,17 +4752,16 @@ mod tests {
 
             let filtered_rb = filter_record_batch(&output, &corrected_mask)?;
 
-            assert_batches_eq!(
-                &[
-                    "+---+----+---+----+",
-                    "| a | b  | x | y  |",
-                    "+---+----+---+----+",
-                    "| 1 | 13 | 1 | 12 |",
-                    "| 1 | 14 | 1 | 11 |",
-                    "+---+----+---+----+",
-                ],
-                &[filtered_rb]
-            );
+            allow_duplicates! {
+                assert_snapshot!(batches_to_string(&[filtered_rb]), @r#"
+                    +---+----+---+----+
+                    | a | b  | x | y  |
+                    +---+----+---+----+
+                    | 1 | 13 | 1 | 12 |
+                    | 1 | 14 | 1 | 11 |
+                    +---+----+---+----+
+            "#);
+            }
 
             // output null rows
             let null_mask = arrow::compute::not(&corrected_mask)?;
@@ -4811,15 +4781,14 @@ mod tests {
 
             let null_joined_batch = filter_record_batch(&output, &null_mask)?;
 
-            assert_batches_eq!(
-                &[
-                    "+---+---+---+---+",
-                    "| a | b | x | y |",
-                    "+---+---+---+---+",
-                    "+---+---+---+---+",
-                ],
-                &[null_joined_batch]
-            );
+            allow_duplicates! {
+                assert_snapshot!(batches_to_string(&[null_joined_batch]), @r#"
+                        +---+---+---+---+
+                        | a | b | x | y |
+                        +---+---+---+---+
+                        +---+---+---+---+
+                "#);
+            }
         }
         Ok(())
     }
diff --git a/datafusion/physical-plan/src/repartition/mod.rs b/datafusion/physical-plan/src/repartition/mod.rs
index c27de77401eb..ebc751201378 100644
--- a/datafusion/physical-plan/src/repartition/mod.rs
+++ b/datafusion/physical-plan/src/repartition/mod.rs
@@ -1078,9 +1078,11 @@ mod tests {
     use arrow::array::{ArrayRef, StringArray, UInt32Array};
     use arrow::datatypes::{DataType, Field, Schema};
     use datafusion_common::cast::as_string_array;
-    use datafusion_common::{arrow_datafusion_err, assert_batches_sorted_eq, exec_err};
+    use datafusion_common::test_util::batches_to_sort_string;
+    use datafusion_common::{arrow_datafusion_err, exec_err};
     use datafusion_common_runtime::JoinSet;
     use datafusion_execution::runtime_env::RuntimeEnvBuilder;
+    use insta::assert_snapshot;
 
     #[tokio::test]
     async fn one_to_many_round_robin() -> Result<()> {
@@ -1333,23 +1335,30 @@ mod tests {
 
         let exec = RepartitionExec::try_new(Arc::new(input), partitioning).unwrap();
 
-        let expected = vec![
-            "+------------------+",
-            "| my_awesome_field |",
-            "+------------------+",
-            "| foo              |",
-            "| bar              |",
-            "| frob             |",
-            "| baz              |",
-            "+------------------+",
-        ];
-
-        assert_batches_sorted_eq!(&expected, &expected_batches);
+        assert_snapshot!(batches_to_sort_string(&expected_batches), @r"
+        +------------------+
+        | my_awesome_field |
+        +------------------+
+        | bar              |
+        | baz              |
+        | foo              |
+        | frob             |
+        +------------------+
+        ");
 
         let output_stream = exec.execute(0, task_ctx).unwrap();
         let batches = crate::common::collect(output_stream).await.unwrap();
 
-        assert_batches_sorted_eq!(&expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r"
+        +------------------+
+        | my_awesome_field |
+        +------------------+
+        | bar              |
+        | baz              |
+        | foo              |
+        | frob             |
+        +------------------+
+        ");
     }
 
     #[tokio::test]
@@ -1383,18 +1392,16 @@ mod tests {
         // output stream 1 should *not* error and have one of the input batches
         let batches = crate::common::collect(output_stream1).await.unwrap();
 
-        let expected = vec![
-            "+------------------+",
-            "| my_awesome_field |",
-            "+------------------+",
-            "| baz              |",
-            "| frob             |",
-            "| gaz              |",
-            "| grob             |",
-            "+------------------+",
-        ];
-
-        assert_batches_sorted_eq!(&expected, &batches);
+        assert_snapshot!(batches_to_sort_string(&batches), @r#"
+            +------------------+
+            | my_awesome_field |
+            +------------------+
+            | baz              |
+            | frob             |
+            | gaz              |
+            | grob             |
+            +------------------+
+            "#);
     }
 
     #[tokio::test]