fix: DH-21978: WorkerHeapSize column is Incorrect. (deephaven#7937)

This is the Core half of the fix, the performance queries are assuming
the heap size we are analyzing is the same size as the current worker
which is not a generally valid assumption. The Core+ half of the fix is
to write the heap size down to permanent logs for later analysis.

Author: cpwright

engine/table/src/main/java/io/deephaven/engine/table/impl/perf/UpdatePerformanceStreamPublisher.java

@@ -9,6 +9,7 @@
 import io.deephaven.engine.table.TableDefinition;
 import io.deephaven.engine.table.impl.perf.UpdatePerformanceTracker.IntervalLevelDetails;
 import io.deephaven.engine.table.impl.sources.ArrayBackedColumnSource;
+import io.deephaven.engine.table.impl.util.HeapSize;
 import io.deephaven.stream.StreamChunkUtils;
 import io.deephaven.stream.StreamConsumer;
 import io.deephaven.stream.StreamPublisher;
@@ -41,7 +42,8 @@ class UpdatePerformanceStreamPublisher implements StreamPublisher {
             ColumnDefinition.ofLong("AllocatedBytes"),
             ColumnDefinition.ofLong("PoolAllocatedBytes"),
             ColumnDefinition.ofString("AuthContext"),
-            ColumnDefinition.ofString("UpdateGraph"));
+            ColumnDefinition.ofString("UpdateGraph"),
+            ColumnDefinition.ofLong("WorkerHeapSize"));
 
     public static TableDefinition definition() {
         return DEFINITION;
@@ -51,9 +53,11 @@ public static TableDefinition definition() {
 
     private WritableChunk<Values>[] chunks;
     private StreamConsumer consumer;
+    private final long heapSize;
 
     public UpdatePerformanceStreamPublisher() {
         chunks = StreamChunkUtils.makeChunksForDefinition(DEFINITION, CHUNK_SIZE);
+        heapSize = HeapSize.getMaximumHeapSizeBytes();
     }
 
     @Override
@@ -111,6 +115,8 @@ public synchronized void add(IntervalLevelDetails intervalLevelDetails, Performa
         chunks[21].<String>asWritableObjectChunk().add(Objects.toString(performanceEntry.getAuthContext()));
         // ColumnDefinition.ofString("UpdateGraph"));
         chunks[22].<String>asWritableObjectChunk().add(Objects.toString(performanceEntry.getUpdateGraphName()));
+        // ColumnDefinition.ofLong("WorkerHeapSize")
+        chunks[23].asWritableLongChunk().add(heapSize);
 
         if (chunks[0].size() == CHUNK_SIZE) {
             flushInternal();

engine/table/src/main/java/io/deephaven/engine/table/impl/util/HeapSize.java

@@ -0,0 +1,20 @@
+//
+// Copyright (c) 2016-2026 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.engine.table.impl.util;
+
+import io.deephaven.util.QueryConstants;
+
+/**
+ * Get this process's maximum heap size for performance logs.
+ */
+public class HeapSize {
+    /**
+     * Get this process's maximum heap size in bytes, or {@link QueryConstants#NULL_LONG}.
+     * 
+     * @return this processes maximum heap size in bytes, or {@link QueryConstants#NULL_LONG}.
+     */
+    public static long getMaximumHeapSizeBytes() {
+        return EngineMetrics.getProcessInfo().getMemoryInfo().heap().max().orElse(QueryConstants.NULL_LONG);
+    }
+}

engine/table/src/main/java/io/deephaven/engine/table/impl/util/QueryOperationPerformanceStreamPublisher.java

@@ -44,7 +44,8 @@ class QueryOperationPerformanceStreamPublisher implements StreamPublisher {
             ColumnDefinition.ofLong("PoolAllocatedBytes"),
             ColumnDefinition.ofLong("InputSizeLong"),
             ColumnDefinition.ofBoolean("WasInterrupted"),
-            ColumnDefinition.ofString("AuthContext"));
+            ColumnDefinition.ofString("AuthContext"),
+            ColumnDefinition.ofLong("WorkerHeapSize"));
     private static final int CHUNK_SIZE = ArrayBackedColumnSource.BLOCK_SIZE;
 
     public static TableDefinition definition() {
@@ -53,9 +54,11 @@ public static TableDefinition definition() {
 
     private WritableChunk<Values>[] chunks;
     private StreamConsumer consumer;
+    private final long heapSize;
 
     QueryOperationPerformanceStreamPublisher() {
         chunks = StreamChunkUtils.makeChunksForDefinition(DEFINITION, CHUNK_SIZE);
+        heapSize = HeapSize.getMaximumHeapSizeBytes();
     }
 
     @Override
@@ -143,6 +146,9 @@ public synchronized void add(final QueryPerformanceNugget nugget) {
         // ColumnDefinition.ofString("AuthContext")
         chunks[24].<String>asWritableObjectChunk().add(Objects.toString(nugget.getAuthContext()));
 
+        // ColumnDefinition.ofLong("WorkerHeapSize")
+        chunks[25].asWritableLongChunk().add(heapSize);
+
         if (chunks[0].size() == CHUNK_SIZE) {
             flushInternal();
         }

engine/table/src/main/java/io/deephaven/engine/table/impl/util/QueryPerformanceStreamPublisher.java

@@ -40,7 +40,8 @@ class QueryPerformanceStreamPublisher implements StreamPublisher {
             ColumnDefinition.ofLong("PoolAllocatedBytes"),
             ColumnDefinition.ofBoolean("WasInterrupted"),
             ColumnDefinition.ofString("Exception"),
-            ColumnDefinition.ofString("AuthContext"));
+            ColumnDefinition.ofString("AuthContext"),
+            ColumnDefinition.ofLong("WorkerHeapSize"));
     private static final int CHUNK_SIZE = ArrayBackedColumnSource.BLOCK_SIZE;
 
     public static TableDefinition definition() {
@@ -49,9 +50,11 @@ public static TableDefinition definition() {
 
     private WritableChunk<Values>[] chunks;
     private StreamConsumer consumer;
+    private final long heapSize;
 
     QueryPerformanceStreamPublisher() {
         chunks = StreamChunkUtils.makeChunksForDefinition(DEFINITION, CHUNK_SIZE);
+        heapSize = HeapSize.getMaximumHeapSizeBytes();
     }
 
     @Override
@@ -126,6 +129,9 @@ public synchronized void add(
         // ColumnDefinition.ofString("AuthContext")
         chunks[19].<String>asWritableObjectChunk().add(Objects.toString(nugget.getAuthContext()));
 
+        // ColumnDefinition.ofLong("WorkerHeapSize")
+        chunks[20].asWritableLongChunk().add(heapSize);
+
         if (chunks[0].size() == CHUNK_SIZE) {
             flushInternal();
         }

extensions/performance/src/main/java/io/deephaven/engine/table/impl/util/PerformanceQueriesGeneral.java

@@ -39,10 +39,8 @@ public static Table queryPerformance(Table queryPerformanceLog, final long evalu
             queryPerformanceLog = queryPerformanceLog.where(whereConditionForEvaluationNumber(evaluationNumber));
         }
 
-        final long workerHeapSizeBytes = getWorkerHeapSizeBytes();
         queryPerformanceLog = queryPerformanceLog
                 .updateView(
-                        "WorkerHeapSize = " + workerHeapSizeBytes + "L",
                         // How long this query ran for, in nanoseconds
                         "DurationNanos = EndTime - StartTime",
                         // How long this query ran for, in seconds
@@ -106,11 +104,9 @@ public static Table queryUpdatePerformance(Table queryUpdatePerformance, final l
             queryUpdatePerformance = queryUpdatePerformance.where(whereConditionForEvaluationNumber(evaluationNumber));
         }
 
-        final long workerHeapSizeBytes = getWorkerHeapSizeBytes();
         queryUpdatePerformance = queryUpdatePerformance
                 .updateView(
                         "IntervalDurationNanos = IntervalEndTime - IntervalStartTime",
-                        "WorkerHeapSize = " + workerHeapSizeBytes + "L",
                         // % of time during this interval that the operation was using CPU
                         "Ratio = UsageNanos / IntervalDurationNanos",
                         // Memory in use by the query. (Only includes active heap memory.)
@@ -370,10 +366,6 @@ private static Table formatColumnsAsPctUpdatePerformance(final Table updatePerfo
         return formatColumnsAsPct(updatePerformanceTable, "Ratio", "QueryMemUsedPct");
     }
 
-    private static long getWorkerHeapSizeBytes() {
-        return EngineMetrics.getProcessInfo().getMemoryInfo().heap().max().orElse(0);
-    }
-
     private static String whereConditionForEvaluationNumber(final long evaluationNumber) {
         return "EvaluationNumber = " + evaluationNumber;
     }