test: update tests for OOM output

Author: wiedld

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -4381,13 +4381,13 @@ mod tests {
             // Asserting that stream-level reservation attempting to overallocate
             assert_contains!(
                 err.to_string(),
-                "Resources exhausted: Additional allocation failed for HashJoinInput[1] with top memory consumers (across reservations) as:\n  HashJoinInput[1]"
+                "Resources exhausted: Additional allocation failed for HashJoinInput[partition=1] with top memory consumers (across reservations) as:\n  HashJoinInput[partition=1]"
 
             );
 
             assert_contains!(
                 err.to_string(),
-                "Failed to allocate additional 120.0 B for HashJoinInput[1]"
+                "Failed to allocate additional 120.0 B for HashJoinInput[partition=1]"
             );
         }
 

datafusion/physical-plan/src/joins/nested_loop_join.rs

@@ -2475,7 +2475,7 @@ pub(crate) mod tests {
 
             assert_contains!(
                 err.to_string(),
-                "Resources exhausted: Additional allocation failed for NestedLoopJoinLoad[0] with top memory consumers (across reservations) as:\n  NestedLoopJoinLoad[0]"
+                "Resources exhausted: Additional allocation failed for NestedLoopJoinLoad[partition=0] with top memory consumers (across reservations) as:\n  NestedLoopJoinLoad[partition=0]"
             );
         }
 

datafusion/physical-plan/src/joins/sort_merge_join/tests.rs

@@ -1799,7 +1799,7 @@ async fn overallocation_single_batch_no_spill() -> Result<()> {
         let err = common::collect(stream).await.unwrap_err();
 
         assert_contains!(err.to_string(), "Failed to allocate additional");
-        assert_contains!(err.to_string(), "SMJStream[0]");
+        assert_contains!(err.to_string(), "SMJStream[partition=0]");
         assert_contains!(err.to_string(), "Disk spilling disabled");
         assert!(join.metrics().is_some());
         assert_eq!(join.metrics().unwrap().spill_count(), Some(0));
@@ -1879,7 +1879,7 @@ async fn overallocation_multi_batch_no_spill() -> Result<()> {
         let err = common::collect(stream).await.unwrap_err();
 
         assert_contains!(err.to_string(), "Failed to allocate additional");
-        assert_contains!(err.to_string(), "SMJStream[0]");
+        assert_contains!(err.to_string(), "SMJStream[partition=0]");
         assert_contains!(err.to_string(), "Disk spilling disabled");
         assert!(join.metrics().is_some());
         assert_eq!(join.metrics().unwrap().spill_count(), Some(0));