[SPARK-41759][CORE] Use weakIntern on string values in create new objects during deserialization

core/src/main/scala/org/apache/spark/status/protobuf/AccumulableInfoSerializer.scala

@@ -23,6 +23,7 @@ import scala.collection.mutable.ArrayBuffer
 
 import org.apache.spark.status.api.v1.AccumulableInfo
 import org.apache.spark.status.protobuf.Utils.getOptional
+import org.apache.spark.util.Utils.weakIntern
 
 private[protobuf] object AccumulableInfoSerializer {
 
@@ -40,10 +41,11 @@ private[protobuf] object AccumulableInfoSerializer {
     updates.forEach { update =>
       accumulatorUpdates.append(new AccumulableInfo(
         id = update.getId,
-        name = update.getName,
-        update = getOptional(update.hasUpdate, update.getUpdate),
-        value = update.getValue))
+        name = weakIntern(update.getName),
+        update = getOptional(update.hasUpdate, () => weakIntern(update.getUpdate)),
+        value = weakIntern(update.getValue)))
     }
+
     accumulatorUpdates
   }
 }

core/src/main/scala/org/apache/spark/status/protobuf/ApplicationEnvironmentInfoWrapperSerializer.scala

@@ -22,6 +22,7 @@ import collection.JavaConverters._
 import org.apache.spark.resource.{ExecutorResourceRequest, TaskResourceRequest}
 import org.apache.spark.status.ApplicationEnvironmentInfoWrapper
 import org.apache.spark.status.api.v1.{ApplicationEnvironmentInfo, ResourceProfileInfo, RuntimeInfo}
+import org.apache.spark.util.Utils.weakIntern
 
 class ApplicationEnvironmentInfoWrapperSerializer
   extends ProtobufSerDe[ApplicationEnvironmentInfoWrapper] {
@@ -73,9 +74,9 @@ class ApplicationEnvironmentInfoWrapperSerializer
   private def deserializeApplicationEnvironmentInfo(info: StoreTypes.ApplicationEnvironmentInfo):
     ApplicationEnvironmentInfo = {
     val runtime = new RuntimeInfo (
-      javaVersion = info.getRuntime.getJavaVersion,
-      javaHome = info.getRuntime.getJavaHome,
-      scalaVersion = info.getRuntime.getScalaVersion
+      javaVersion = weakIntern(info.getRuntime.getJavaVersion),
+      javaHome = weakIntern(info.getRuntime.getJavaHome),
+      scalaVersion = weakIntern(info.getRuntime.getScalaVersion)
     )
     val pairSSToTuple = (pair: StoreTypes.PairStrings) => {
       (pair.getValue1, pair.getValue2)
@@ -134,15 +135,16 @@ class ApplicationEnvironmentInfoWrapperSerializer
   private def deserializeExecutorResourceRequest(info: StoreTypes.ExecutorResourceRequest):
     ExecutorResourceRequest = {
     new ExecutorResourceRequest(
-      resourceName = info.getResourceName,
+      resourceName = weakIntern(info.getResourceName),
       amount = info.getAmount,
-      discoveryScript = info.getDiscoveryScript,
-      vendor = info.getVendor
+      discoveryScript = weakIntern(info.getDiscoveryScript),
+      vendor = weakIntern(info.getVendor)
     )
   }
 
   private def deserializeTaskResourceRequest(info: StoreTypes.TaskResourceRequest):
     TaskResourceRequest = {
-    new TaskResourceRequest(resourceName = info.getResourceName, amount = info.getAmount)
+    new TaskResourceRequest(resourceName = weakIntern(info.getResourceName),
+      amount = info.getAmount)
   }
 }

core/src/main/scala/org/apache/spark/status/protobuf/ApplicationInfoWrapperSerializer.scala

@@ -24,7 +24,7 @@ import collection.JavaConverters._
 import org.apache.spark.status.ApplicationInfoWrapper
 import org.apache.spark.status.api.v1.{ApplicationAttemptInfo, ApplicationInfo}
 import org.apache.spark.status.protobuf.Utils.getOptional
-
+import org.apache.spark.util.Utils.weakIntern
 
 class ApplicationInfoWrapperSerializer extends ProtobufSerDe[ApplicationInfoWrapper] {
 
@@ -71,8 +71,8 @@ class ApplicationInfoWrapperSerializer extends ProtobufSerDe[ApplicationInfoWrap
     val memoryPerExecutorMB = getOptional(info.hasMemoryPerExecutorMb, info.getMemoryPerExecutorMb)
     val attempts = info.getAttemptsList.asScala.map(deserializeApplicationAttemptInfo)
     ApplicationInfo(
-      id = info.getId,
-      name = info.getName,
+      id = weakIntern(info.getId),
+      name = weakIntern(info.getName),
       coresGranted = coresGranted,
       maxCores = maxCores,
       coresPerExecutor = coresPerExecutor,
@@ -99,17 +99,17 @@ class ApplicationInfoWrapperSerializer extends ProtobufSerDe[ApplicationInfoWrap
 
   private def deserializeApplicationAttemptInfo(info: StoreTypes.ApplicationAttemptInfo):
     ApplicationAttemptInfo = {
-    val attemptId = getOptional(info.hasAttemptId, info.getAttemptId)
+    val attemptId = getOptional(info.hasAttemptId, () => weakIntern(info.getAttemptId))
 
     ApplicationAttemptInfo(
       attemptId = attemptId,
       startTime = new Date(info.getStartTime),
       endTime = new Date(info.getEndTime),
       lastUpdated = new Date(info.getLastUpdated),
       duration = info.getDuration,
-      sparkUser = info.getSparkUser,
+      sparkUser = weakIntern(info.getSparkUser),
       completed = info.getCompleted,
-      appSparkVersion = info.getAppSparkVersion
+      appSparkVersion = weakIntern(info.getAppSparkVersion)
     )
   }
 }

core/src/main/scala/org/apache/spark/status/protobuf/CachedQuantileSerializer.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.status.protobuf
 
 import org.apache.spark.status.CachedQuantile
+import org.apache.spark.util.Utils.weakIntern
 
 class CachedQuantileSerializer extends ProtobufSerDe[CachedQuantile] {
 
@@ -73,7 +74,7 @@ class CachedQuantileSerializer extends ProtobufSerDe[CachedQuantile] {
     new CachedQuantile(
       stageId = binary.getStageId.toInt,
       stageAttemptId = binary.getStageAttemptId,
-      quantile = binary.getQuantile,
+      quantile = weakIntern(binary.getQuantile),
       taskCount = binary.getTaskCount,
       duration = binary.getDuration,
       executorDeserializeTime = binary.getExecutorDeserializeTime,

core/src/main/scala/org/apache/spark/status/protobuf/ExecutorStageSummaryWrapperSerializer.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.status.protobuf
 
 import org.apache.spark.status.ExecutorStageSummaryWrapper
+import org.apache.spark.util.Utils.weakIntern
 
 class ExecutorStageSummaryWrapperSerializer
   extends ProtobufSerDe[ExecutorStageSummaryWrapper] {
@@ -38,7 +39,7 @@ class ExecutorStageSummaryWrapperSerializer
     new ExecutorStageSummaryWrapper(
       stageId = binary.getStageId.toInt,
       stageAttemptId = binary.getStageAttemptId,
-      executorId = binary.getExecutorId,
+      executorId = weakIntern(binary.getExecutorId),
       info = info)
   }
 }

core/src/main/scala/org/apache/spark/status/protobuf/ExecutorSummaryWrapperSerializer.scala

@@ -25,6 +25,7 @@ import org.apache.spark.resource.ResourceInformation
 import org.apache.spark.status.ExecutorSummaryWrapper
 import org.apache.spark.status.api.v1.{ExecutorSummary, MemoryMetrics}
 import org.apache.spark.status.protobuf.Utils.getOptional
+import org.apache.spark.util.Utils.weakIntern
 
 class ExecutorSummaryWrapperSerializer extends ProtobufSerDe[ExecutorSummaryWrapper] {
 
@@ -104,13 +105,13 @@ class ExecutorSummaryWrapperSerializer extends ProtobufSerDe[ExecutorSummaryWrap
       getOptional(binary.hasPeakMemoryMetrics,
         () => ExecutorMetricsSerializer.deserialize(binary.getPeakMemoryMetrics))
     val removeTime = getOptional(binary.hasRemoveTime, () => new Date(binary.getRemoveTime))
-    val removeReason = getOptional(binary.hasRemoveReason, () => binary.getRemoveReason)
+    val removeReason = getOptional(binary.hasRemoveReason, () => weakIntern(binary.getRemoveReason))
     val memoryMetrics =
       getOptional(binary.hasMemoryMetrics,
         () => deserializeMemoryMetrics(binary.getMemoryMetrics))
     new ExecutorSummary(
-      id = binary.getId,
-      hostPort = binary.getHostPort,
+      id = weakIntern(binary.getId),
+      hostPort = weakIntern(binary.getHostPort),
       isActive = binary.getIsActive,
       rddBlocks = binary.getRddBlocks,
       memoryUsed = binary.getMemoryUsed,
@@ -171,7 +172,7 @@ class ExecutorSummaryWrapperSerializer extends ProtobufSerDe[ExecutorSummaryWrap
   private def deserializeResourceInformation(binary: StoreTypes.ResourceInformation):
     ResourceInformation = {
     new ResourceInformation(
-      name = binary.getName,
-      addresses = binary.getAddressesList.asScala.toArray)
+      name = weakIntern(binary.getName),
+      addresses = binary.getAddressesList.asScala.map(weakIntern).toArray)
   }
 }

core/src/main/scala/org/apache/spark/status/protobuf/JobDataWrapperSerializer.scala

@@ -24,6 +24,7 @@ import collection.JavaConverters._
 import org.apache.spark.status.JobDataWrapper
 import org.apache.spark.status.api.v1.JobData
 import org.apache.spark.status.protobuf.Utils.getOptional
+import org.apache.spark.util.Utils.weakIntern
 
 class JobDataWrapperSerializer extends ProtobufSerDe[JobDataWrapper] {
 
@@ -79,16 +80,16 @@ class JobDataWrapperSerializer extends ProtobufSerDe[JobDataWrapper] {
   }
 
   private def deserializeJobData(info: StoreTypes.JobData): JobData = {
-    val description = getOptional(info.hasDescription, info.getDescription)
+    val description = getOptional(info.hasDescription, () => weakIntern(info.getDescription))
     val submissionTime =
       getOptional(info.hasSubmissionTime, () => new Date(info.getSubmissionTime))
     val completionTime = getOptional(info.hasCompletionTime, () => new Date(info.getCompletionTime))
-    val jobGroup = getOptional(info.hasJobGroup, info.getJobGroup)
+    val jobGroup = getOptional(info.hasJobGroup, () => weakIntern(info.getJobGroup))
     val status = JobExecutionStatusSerializer.deserialize(info.getStatus)
 
     new JobData(
       jobId = info.getJobId.toInt,
-      name = info.getName,
+      name = weakIntern(info.getName),
       description = description,
       submissionTime = submissionTime,
       completionTime = completionTime,

core/src/main/scala/org/apache/spark/status/protobuf/ProcessSummaryWrapperSerializer.scala

@@ -24,6 +24,7 @@ import scala.collection.JavaConverters._
 import org.apache.spark.status.ProcessSummaryWrapper
 import org.apache.spark.status.api.v1.ProcessSummary
 import org.apache.spark.status.protobuf.Utils.getOptional
+import org.apache.spark.util.Utils.weakIntern
 
 class ProcessSummaryWrapperSerializer extends ProtobufSerDe[ProcessSummaryWrapper] {
 
@@ -59,8 +60,8 @@ class ProcessSummaryWrapperSerializer extends ProtobufSerDe[ProcessSummaryWrappe
   private def deserializeProcessSummary(info: StoreTypes.ProcessSummary): ProcessSummary = {
     val removeTime = getOptional(info.hasRemoveTime, () => new Date(info.getRemoveTime))
     new ProcessSummary(
-      id = info.getId,
-      hostPort = info.getHostPort,
+      id = weakIntern(info.getId),
+      hostPort = weakIntern(info.getHostPort),
       isActive = info.getIsActive,
       totalCores = info.getTotalCores,
       addTime = new Date(info.getAddTime),

core/src/main/scala/org/apache/spark/status/protobuf/RDDOperationGraphWrapperSerializer.scala

@@ -23,6 +23,7 @@ import org.apache.spark.rdd.DeterministicLevel
 import org.apache.spark.status.{RDDOperationClusterWrapper, RDDOperationGraphWrapper}
 import org.apache.spark.status.protobuf.StoreTypes.{DeterministicLevel => GDeterministicLevel}
 import org.apache.spark.ui.scope.{RDDOperationEdge, RDDOperationNode}
+import org.apache.spark.util.Utils.weakIntern
 
 class RDDOperationGraphWrapperSerializer extends ProtobufSerDe[RDDOperationGraphWrapper] {
 
@@ -70,8 +71,8 @@ class RDDOperationGraphWrapperSerializer extends ProtobufSerDe[RDDOperationGraph
   private def deserializeRDDOperationClusterWrapper(op: StoreTypes.RDDOperationClusterWrapper):
     RDDOperationClusterWrapper = {
     new RDDOperationClusterWrapper(
-      id = op.getId,
-      name = op.getName,
+      id = weakIntern(op.getId),
+      name = weakIntern(op.getName),
       childNodes = op.getChildNodesList.asScala.map(deserializeRDDOperationNode),
       childClusters =
         op.getChildClustersList.asScala.map(deserializeRDDOperationClusterWrapper)
@@ -94,10 +95,10 @@ class RDDOperationGraphWrapperSerializer extends ProtobufSerDe[RDDOperationGraph
   private def deserializeRDDOperationNode(node: StoreTypes.RDDOperationNode): RDDOperationNode = {
     RDDOperationNode(
       id = node.getId,
-      name = node.getName,
+      name = weakIntern(node.getName),
       cached = node.getCached,
       barrier = node.getBarrier,
-      callsite = node.getCallsite,
+      callsite = weakIntern(node.getCallsite),
       outputDeterministicLevel = DeterministicLevelSerializer.deserialize(
         node.getOutputDeterministicLevel)
     )

core/src/main/scala/org/apache/spark/status/protobuf/RDDStorageInfoWrapperSerializer.scala

@@ -22,6 +22,7 @@ import scala.collection.JavaConverters._
 import org.apache.spark.status.RDDStorageInfoWrapper
 import org.apache.spark.status.api.v1.{RDDDataDistribution, RDDPartitionInfo, RDDStorageInfo}
 import org.apache.spark.status.protobuf.Utils.getOptional
+import org.apache.spark.util.Utils.weakIntern
 
 class RDDStorageInfoWrapperSerializer extends ProtobufSerDe[RDDStorageInfoWrapper] {
 
@@ -81,10 +82,10 @@ class RDDStorageInfoWrapperSerializer extends ProtobufSerDe[RDDStorageInfoWrappe
   private def deserializeRDDStorageInfo(info: StoreTypes.RDDStorageInfo): RDDStorageInfo = {
     new RDDStorageInfo(
       id = info.getId,
-      name = info.getName,
+      name = weakIntern(info.getName),
       numPartitions = info.getNumPartitions,
       numCachedPartitions = info.getNumCachedPartitions,
-      storageLevel = info.getStorageLevel,
+      storageLevel = weakIntern(info.getStorageLevel),
       memoryUsed = info.getMemoryUsed,
       diskUsed = info.getDiskUsed,
       dataDistribution =
@@ -102,7 +103,7 @@ class RDDStorageInfoWrapperSerializer extends ProtobufSerDe[RDDStorageInfoWrappe
     RDDDataDistribution = {
 
     new RDDDataDistribution(
-      address = info.getAddress,
+      address = weakIntern(info.getAddress),
       memoryUsed = info.getMemoryUsed,
       memoryRemaining = info.getMemoryRemaining,
       diskUsed = info.getDiskUsed,
@@ -117,11 +118,11 @@ class RDDStorageInfoWrapperSerializer extends ProtobufSerDe[RDDStorageInfoWrappe
 
   private def deserializeRDDPartitionInfo(info: StoreTypes.RDDPartitionInfo): RDDPartitionInfo = {
     new RDDPartitionInfo(
-      blockName = info.getBlockName,
-      storageLevel = info.getStorageLevel,
+      blockName = weakIntern(info.getBlockName),
+      storageLevel = weakIntern(info.getStorageLevel),
       memoryUsed = info.getMemoryUsed,
       diskUsed = info.getDiskUsed,
-      executors = info.getExecutorsList.asScala
+      executors = info.getExecutorsList.asScala.map(weakIntern)
     )
   }
 }

core/src/main/scala/org/apache/spark/status/protobuf/StreamBlockDataSerializer.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.status.protobuf
 
 import org.apache.spark.status.StreamBlockData
+import org.apache.spark.util.Utils.weakIntern
 
 class StreamBlockDataSerializer extends ProtobufSerDe[StreamBlockData] {
 
@@ -38,10 +39,10 @@ class StreamBlockDataSerializer extends ProtobufSerDe[StreamBlockData] {
   override def deserialize(bytes: Array[Byte]): StreamBlockData = {
     val binary = StoreTypes.StreamBlockData.parseFrom(bytes)
     new StreamBlockData(
-      name = binary.getName,
-      executorId = binary.getExecutorId,
-      hostPort = binary.getHostPort,
-      storageLevel = binary.getStorageLevel,
+      name = weakIntern(binary.getName),
+      executorId = weakIntern(binary.getExecutorId),
+      hostPort = weakIntern(binary.getHostPort),
+      storageLevel = weakIntern(binary.getStorageLevel),
       useMemory = binary.getUseMemory,
       useDisk = binary.getUseDisk,
       deserialized = binary.getDeserialized,

core/src/main/scala/org/apache/spark/status/protobuf/TaskDataWrapperSerializer.scala

@@ -98,7 +98,7 @@ class TaskDataWrapperSerializer extends ProtobufSerDe[TaskDataWrapper] {
       taskLocality = weakIntern(binary.getTaskLocality),
       speculative = binary.getSpeculative,
       accumulatorUpdates = accumulatorUpdates,
-      errorMessage = getOptional(binary.hasErrorMessage, binary.getErrorMessage),
+      errorMessage = getOptional(binary.hasErrorMessage, () => weakIntern(binary.getErrorMessage)),
       hasMetrics = binary.getHasMetrics,
       executorDeserializeTime = binary.getExecutorDeserializeTime,
       executorDeserializeCpuTime = binary.getExecutorDeserializeCpuTime,

sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SQLExecutionUIDataSerializer.scala

@@ -24,6 +24,7 @@ import collection.JavaConverters._
 import org.apache.spark.sql.execution.ui.SQLExecutionUIData
 import org.apache.spark.status.protobuf.{JobExecutionStatusSerializer, ProtobufSerDe, StoreTypes}
 import org.apache.spark.status.protobuf.Utils.getOptional
+import org.apache.spark.util.Utils.weakIntern
 
 class SQLExecutionUIDataSerializer extends ProtobufSerDe[SQLExecutionUIData] {
 
@@ -64,7 +65,7 @@ class SQLExecutionUIDataSerializer extends ProtobufSerDe[SQLExecutionUIData] {
     val ui = StoreTypes.SQLExecutionUIData.parseFrom(bytes)
     val completionTime =
       getOptional(ui.hasCompletionTime, () => new Date(ui.getCompletionTime))
-    val errorMessage = getOptional(ui.hasErrorMessage, () => ui.getErrorMessage)
+    val errorMessage = getOptional(ui.hasErrorMessage, () => weakIntern(ui.getErrorMessage))
     val metrics =
       ui.getMetricsList.asScala.map(m => SQLPlanMetricSerializer.deserialize(m))
     val jobs = ui.getJobsMap.asScala.map {
@@ -81,9 +82,9 @@ class SQLExecutionUIDataSerializer extends ProtobufSerDe[SQLExecutionUIData] {
     new SQLExecutionUIData(
       executionId = ui.getExecutionId,
       rootExecutionId = ui.getRootExecutionId,
-      description = ui.getDescription,
-      details = ui.getDetails,
-      physicalPlanDescription = ui.getPhysicalPlanDescription,
+      description = weakIntern(ui.getDescription),
+      details = weakIntern(ui.getDetails),
+      physicalPlanDescription = weakIntern(ui.getPhysicalPlanDescription),
       modifiedConfigs = ui.getModifiedConfigsMap.asScala.toMap,
       metrics = metrics,
       submissionTime = ui.getSubmissionTime,

sql/core/src/main/scala/org/apache/spark/status/protobuf/sql/SQLPlanMetricSerializer.scala

@@ -19,6 +19,7 @@ package org.apache.spark.status.protobuf.sql
 
 import org.apache.spark.sql.execution.ui.SQLPlanMetric
 import org.apache.spark.status.protobuf.StoreTypes
+import org.apache.spark.util.Utils.weakIntern
 
 object SQLPlanMetricSerializer {
 
@@ -31,6 +32,7 @@ object SQLPlanMetricSerializer {
   }
 
   def deserialize(metrics: StoreTypes.SQLPlanMetric): SQLPlanMetric = {
-    SQLPlanMetric(metrics.getName, metrics.getAccumulatorId, metrics.getMetricType)
+    SQLPlanMetric(weakIntern(metrics.getName), metrics.getAccumulatorId,
+      weakIntern(metrics.getMetricType))
   }
 }