Use more compact representation of AssignmentResult in diagnostics

Previously we used toString but it transitively calls out to
ConnectorSplit.toString which may be very verbose. We observed memory
pressure problems while logging diagnostics.

core/trino-main/src/main/java/io/trino/execution/scheduler/faulttolerant/EventDrivenFaultTolerantQueryScheduler.java

@@ -640,6 +640,7 @@ private static Optional<String> getEventDebugInfo(Event event)
                     // Also empty events can push important events out of recorded debug information - making debug logs less useful.
                     return Optional.empty();
                 }
+                return Optional.of(splitAssignmentEvent.debugInfo());
             }
 
             return Optional.of(event.toString());
@@ -3428,6 +3429,15 @@ public String toString()
                     .add("assignmentResult", assignmentResult)
                     .toString();
         }
+
+        public String debugInfo()
+        {
+            // toString is too verbose
+            return toStringHelper(this)
+                    .add("stageId", getStageId())
+                    .add("assignmentResult", assignmentResult.debugInfo())
+                    .toString();
+        }
     }
 
     private static class StageFailureEvent

core/trino-main/src/main/java/io/trino/execution/scheduler/faulttolerant/SplitAssigner.java

@@ -23,9 +23,13 @@
 import io.trino.sql.planner.plan.PlanNodeId;
 
 import java.util.List;
+import java.util.Map;
 
+import static com.google.common.base.MoreObjects.toStringHelper;
 import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.toList;
+import static java.util.stream.Collectors.toMap;
 
 /**
  * An implementation is not required to be thread safe
@@ -61,6 +65,20 @@ record PartitionUpdate(
             checkArgument(!(readyForScheduling && splits.isEmpty()), "partition update with empty splits marked as ready for scheduling");
             splits = ImmutableListMultimap.copyOf(requireNonNull(splits, "splits is null"));
         }
+
+        public String debugInfo()
+        {
+            // toString is too verbose
+            return toStringHelper(this)
+                    .add("partitionId", partitionId)
+                    .add("planNodeId", planNodeId)
+                    .add("readyForScheduling", readyForScheduling)
+                    .add("splits", splits.asMap().entrySet().stream().collect(toMap(
+                            Map.Entry::getKey,
+                            entry -> entry.getValue().size())))
+                    .add("noMoreSplits", noMoreSplits)
+                    .toString();
+        }
     }
 
     record AssignmentResult(
@@ -80,6 +98,17 @@ boolean isEmpty()
             return partitionsAdded.isEmpty() && !noMorePartitions && partitionUpdates.isEmpty() && sealedPartitions.isEmpty();
         }
 
+        public String debugInfo()
+        {
+            // toString is too verbose
+            return toStringHelper(this)
+                    .add("partitionsAdded", partitionsAdded)
+                    .add("noMorePartitions", noMorePartitions)
+                    .add("partitionUpdates", partitionUpdates.stream().map(PartitionUpdate::debugInfo).collect(toList()))
+                    .add("sealedPartitions", sealedPartitions)
+                    .toString();
+        }
+
         public static AssignmentResult.Builder builder()
         {
             return new AssignmentResult.Builder();