[BugFix] fix counter QuerySumMemoryUsage (backport #52023) (#52060)

Co-authored-by: Murphy <[email protected]>

fe/fe-core/src/main/java/com/starrocks/qe/scheduler/QueryRuntimeProfile.java

@@ -401,9 +401,9 @@ public RuntimeProfile buildQueryProfile(boolean needMerge) {
  newQueryProfile.copyAllInfoStringsFrom(queryProfile, null);
  newQueryProfile.copyAllCountersFrom(queryProfile);
 
+ Map<String, Long> peakMemoryEachBE = Maps.newHashMap();
  long sumQueryCumulativeCpuTime = 0;
  long sumQuerySpillBytes = 0;
- long sumQueryPeakMemoryBytes = 0;
  long maxQueryPeakMemoryUsage = 0;
  long maxQueryExecutionWallTime = 0;
 
@@ -443,7 +443,8 @@ public RuntimeProfile buildQueryProfile(boolean needMerge) {
  toBeRemove = instanceProfile.getCounter("QueryPeakMemoryUsage");
  if (toBeRemove != null) {
  maxQueryPeakMemoryUsage = Math.max(maxQueryPeakMemoryUsage, toBeRemove.getValue());
- sumQueryPeakMemoryBytes += toBeRemove.getValue();
+ String beAddress = instanceProfile.getInfoString("Address");
+ peakMemoryEachBE.merge(beAddress, toBeRemove.getValue(), Long::max);
  }
  instanceProfile.removeCounter("QueryPeakMemoryUsage");
 
@@ -583,7 +584,7 @@ public RuntimeProfile buildQueryProfile(boolean needMerge) {
  Counter queryPeakMemoryUsage = newQueryProfile.addCounter("QueryPeakMemoryUsagePerNode", TUnit.BYTES, null);
  queryPeakMemoryUsage.setValue(maxQueryPeakMemoryUsage);
  Counter sumQueryPeakMemoryUsage = newQueryProfile.addCounter("QuerySumMemoryUsage", TUnit.BYTES, null);
- sumQueryPeakMemoryUsage.setValue(sumQueryPeakMemoryBytes);
+ sumQueryPeakMemoryUsage.setValue(peakMemoryEachBE.values().stream().reduce(0L, Long::sum));
  Counter queryExecutionWallTime = newQueryProfile.addCounter("QueryExecutionWallTime", TUnit.TIME_NS, null);
  queryExecutionWallTime.setValue(maxQueryExecutionWallTime);
  Counter querySpillBytes = newQueryProfile.addCounter("QuerySpillBytes", TUnit.BYTES, null);