[SPARK-47582][SQL] Migrate Catalyst logInfo with variables to structu…

…red logging framework

### What changes were proposed in this pull request?

Migrate logInfo with variables of the Catalyst module to structured logging framework. This is part1 which transforms the logInfo entries of the following API
```
def logInfo(msg: => String): Unit
```
to
```
def logInfo(entry: LogEntry): Unit
```

### Why are the changes needed?

To enhance Apache Spark's logging system by implementing structured logging.

### Does this PR introduce _any_ user-facing change?

Yes, Spark core logs will contain additional MDC

### How was this patch tested?

Compiler and scala style checks, as well as code review.

### Was this patch authored or co-authored using generative AI tooling?

No

Closes apache#45866 from dtenedor/catalyst-log-info.

Authored-by: Daniel Tenedorio <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>

common/utils/src/main/scala/org/apache/spark/internal/LogKey.scala

@@ -29,6 +29,7 @@ object LogKey extends Enumeration {
   val BLOCK_MANAGER_ID = Value
   val BROADCAST_ID = Value
   val BUCKET = Value
+  val BYTECODE_SIZE = Value
   val CATEGORICAL_FEATURES = Value
   val CLASS_LOADER = Value
   val CLASS_NAME = Value
@@ -50,6 +51,7 @@ object LogKey extends Enumeration {
   val FAILURES = Value
   val HOST = Value
   val JOB_ID = Value
+  val JOIN_CONDITION = Value
   val LEARNING_RATE = Value
   val LINE = Value
   val LINE_NUM = Value
@@ -72,13 +74,17 @@ object LogKey extends Enumeration {
   val PATHS = Value
   val POD_ID = Value
   val PORT = Value
+  val QUERY_PLAN = Value
   val RANGE = Value
   val RDD_ID = Value
   val REASON = Value
   val REDUCE_ID = Value
   val REMOTE_ADDRESS = Value
   val RETRY_COUNT = Value
   val RPC_ADDRESS = Value
+  val RULE_BATCH_NAME = Value
+  val RULE_NAME = Value
+  val RULE_NUMBER_OF_RUNS = Value
   val SHUFFLE_BLOCK_INFO = Value
   val SHUFFLE_ID = Value
   val SHUFFLE_MERGE_ID = Value
@@ -96,9 +102,12 @@ object LogKey extends Enumeration {
   val THREAD_NAME = Value
   val TID = Value
   val TIMEOUT = Value
+  val TIME_UNITS = Value
   val URI = Value
   val USER_NAME = Value
+  val WATERMARK_CONSTRAINT = Value
   val WORKER_URL = Value
+  val XSD_PATH = Value
 
   type LogKey = Value
 }

common/utils/src/main/scala/org/apache/spark/internal/Logging.scala

@@ -49,6 +49,8 @@ case class MessageWithContext(message: String, context: java.util.HashMap[String
     resultMap.putAll(mdc.context)
     MessageWithContext(message + mdc.message, resultMap)
   }
+
+  def stripMargin: MessageWithContext = copy(message = message.stripMargin)
 }
 
 /**

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/StreamingJoinHelper.scala

@@ -20,6 +20,8 @@ package org.apache.spark.sql.catalyst.analysis
 import scala.util.control.NonFatal
 
 import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKey._
+import org.apache.spark.internal.MDC
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.planning.ExtractEquiJoinKeys
 import org.apache.spark.sql.catalyst.plans.logical.{EventTimeWatermark, LogicalPlan}
@@ -107,7 +109,8 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
         case _ => None
       }
       if (stateWatermark.nonEmpty) {
-        logInfo(s"Condition $joinCondition generated watermark constraint = ${stateWatermark.get}")
+        logInfo(log"Condition ${MDC(JOIN_CONDITION, joinCondition)} generated " +
+          log"watermark constraint = ${MDC(WATERMARK_CONSTRAINT, stateWatermark.get)}")
       }
       stateWatermark
     }
@@ -195,7 +198,8 @@ object StreamingJoinHelper extends PredicateHelper with Logging {
     }.reduceLeft(Add(_, _))
 
     // Calculate the constraint value
-    logInfo(s"Final expression to evaluate constraint:\t$exprWithWatermarkSubstituted")
+    logInfo(log"Final expression to evaluate " +
+      log"constraint:\t${MDC(WATERMARK_CONSTRAINT, exprWithWatermarkSubstituted)}")
     val constraintValue = exprWithWatermarkSubstituted.eval().asInstanceOf[java.lang.Double]
     Some((Double2double(constraintValue) / 1000.0).toLong)
   }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala

@@ -33,6 +33,8 @@ import org.codehaus.janino.util.ClassFile
 import org.apache.spark.{SparkException, SparkIllegalArgumentException, TaskContext, TaskKilledException}
 import org.apache.spark.executor.InputMetrics
 import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKey._
+import org.apache.spark.internal.MDC
 import org.apache.spark.metrics.source.CodegenMetrics
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.encoders.HashableWeakReference
@@ -1593,7 +1595,8 @@ object CodeGenerator extends Logging {
 
             if (byteCodeSize > DEFAULT_JVM_HUGE_METHOD_LIMIT) {
               logInfo("Generated method too long to be JIT compiled: " +
-                s"${cf.getThisClassName}.${method.getName} is $byteCodeSize bytes")
+                log"${MDC(CLASS_NAME, cf.getThisClassName)}.${MDC(METHOD_NAME, method.getName)} " +
+                log"is ${MDC(BYTECODE_SIZE, byteCodeSize)} bytes")
             }
 
             byteCodeSize
@@ -1638,7 +1641,7 @@ object CodeGenerator extends Logging {
         val timeMs: Double = duration.toDouble / NANOS_PER_MILLIS
         CodegenMetrics.METRIC_SOURCE_CODE_SIZE.update(code.body.length)
         CodegenMetrics.METRIC_COMPILATION_TIME.update(timeMs.toLong)
-        logInfo(s"Code generated in $timeMs ms")
+        logInfo(log"Code generated in ${MDC(TIME_UNITS, timeMs)} ms")
         _compileTime.add(duration)
         result
     }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -20,6 +20,8 @@ package org.apache.spark.sql.catalyst.optimizer
 import scala.collection.mutable
 
 import org.apache.spark.SparkException
+import org.apache.spark.internal.LogKey._
+import org.apache.spark.internal.MDC
 import org.apache.spark.sql.catalyst.SQLConfHelper
 import org.apache.spark.sql.catalyst.analysis._
 import org.apache.spark.sql.catalyst.catalog.{InMemoryCatalog, SessionCatalog}
@@ -455,7 +457,8 @@ abstract class Optimizer(catalogManager: CatalogManager)
         val filteredRules = batch.rules.filter { rule =>
           val exclude = excludedRules.contains(rule.ruleName)
           if (exclude) {
-            logInfo(s"Optimization rule '${rule.ruleName}' is excluded from the optimizer.")
+            logInfo(log"Optimization rule '${MDC(RULE_NAME, rule.ruleName)}' " +
+              log"is excluded from the optimizer.")
           }
           !exclude
         }
@@ -464,8 +467,8 @@ abstract class Optimizer(catalogManager: CatalogManager)
         } else if (filteredRules.nonEmpty) {
           Some(Batch(batch.name, batch.strategy, filteredRules: _*))
         } else {
-          logInfo(s"Optimization batch '${batch.name}' is excluded from the optimizer " +
-            s"as all enclosed rules have been excluded.")
+          logInfo(log"Optimization batch '${MDC(RULE_BATCH_NAME, batch.name)}' " +
+            log"is excluded from the optimizer as all enclosed rules have been excluded.")
           None
         }
       }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala

@@ -18,7 +18,9 @@
 package org.apache.spark.sql.catalyst.rules
 
 import org.apache.spark.SparkException
-import org.apache.spark.internal.Logging
+import org.apache.spark.internal.{Logging, MessageWithContext}
+import org.apache.spark.internal.LogKey._
+import org.apache.spark.internal.MDC
 import org.apache.spark.sql.catalyst.QueryPlanningTracker
 import org.apache.spark.sql.catalyst.trees.TreeNode
 import org.apache.spark.sql.catalyst.util.DateTimeConstants.NANOS_PER_SECOND
@@ -56,10 +58,10 @@ class PlanChangeLogger[TreeType <: TreeNode[_]] extends Logging {
   def logRule(ruleName: String, oldPlan: TreeType, newPlan: TreeType): Unit = {
     if (!newPlan.fastEquals(oldPlan)) {
       if (logRules.isEmpty || logRules.get.contains(ruleName)) {
-        def message(): String = {
-          s"""
-             |=== Applying Rule $ruleName ===
-             |${sideBySide(oldPlan.treeString, newPlan.treeString).mkString("\n")}
+        def message(): MessageWithContext = {
+          log"""
+             |=== Applying Rule ${MDC(RULE_NAME, ruleName)} ===
+             |${MDC(QUERY_PLAN, sideBySide(oldPlan.treeString, newPlan.treeString).mkString("\n"))}
            """.stripMargin
         }
 
@@ -70,14 +72,14 @@ class PlanChangeLogger[TreeType <: TreeNode[_]] extends Logging {
 
   def logBatch(batchName: String, oldPlan: TreeType, newPlan: TreeType): Unit = {
     if (logBatches.isEmpty || logBatches.get.contains(batchName)) {
-      def message(): String = {
+      def message(): MessageWithContext = {
         if (!oldPlan.fastEquals(newPlan)) {
-          s"""
-             |=== Result of Batch $batchName ===
-             |${sideBySide(oldPlan.treeString, newPlan.treeString).mkString("\n")}
+          log"""
+             |=== Result of Batch ${MDC(RULE_BATCH_NAME, batchName)} ===
+             |${MDC(QUERY_PLAN, sideBySide(oldPlan.treeString, newPlan.treeString).mkString("\n"))}
           """.stripMargin
         } else {
-          s"Batch $batchName has no effect."
+          log"Batch ${MDC(RULE_BATCH_NAME, batchName)} has no effect."
         }
       }
 
@@ -88,26 +90,26 @@ class PlanChangeLogger[TreeType <: TreeNode[_]] extends Logging {
   def logMetrics(metrics: QueryExecutionMetrics): Unit = {
     val totalTime = metrics.time / NANOS_PER_SECOND.toDouble
     val totalTimeEffective = metrics.timeEffective / NANOS_PER_SECOND.toDouble
-    val message =
-      s"""
+    val message: MessageWithContext =
+      log"""
          |=== Metrics of Executed Rules ===
-         |Total number of runs: ${metrics.numRuns}
-         |Total time: $totalTime seconds
-         |Total number of effective runs: ${metrics.numEffectiveRuns}
-         |Total time of effective runs: $totalTimeEffective seconds
+         |Total number of runs: ${MDC(RULE_NUMBER_OF_RUNS, metrics.numRuns)}
+         |Total time: ${MDC(TIME_UNITS, totalTime)} seconds
+         |Total number of effective runs: ${MDC(RULE_NUMBER_OF_RUNS, metrics.numEffectiveRuns)}
+         |Total time of effective runs: ${MDC(TIME_UNITS, totalTimeEffective)} seconds
       """.stripMargin
 
     logBasedOnLevel(message)
   }
 
-  private def logBasedOnLevel(f: => String): Unit = {
+  private def logBasedOnLevel(f: => MessageWithContext): Unit = {
     logLevel match {
-      case "TRACE" => logTrace(f)
-      case "DEBUG" => logDebug(f)
+      case "TRACE" => logTrace(f.message)
+      case "DEBUG" => logDebug(f.message)
       case "INFO" => logInfo(f)
       case "WARN" => logWarning(f)
       case "ERROR" => logError(f)
-      case _ => logTrace(f)
+      case _ => logTrace(f.message)
     }
   }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/xml/ValidatorUtil.scala

@@ -27,6 +27,8 @@ import org.apache.hadoop.fs.Path
 import org.apache.spark.SparkFiles
 import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKey._
+import org.apache.spark.internal.MDC
 
 /**
  * Utilities for working with XSD validation.
@@ -56,7 +58,8 @@ object ValidatorUtil extends Logging {
       case e: Throwable =>
         // Handle case where it was added with sc.addFile
         // When they are added via sc.addFile, they are always downloaded to local file system
-        logInfo(s"$xsdPath was not found, falling back to look up files added by Spark")
+        logInfo(log"${MDC(XSD_PATH, xsdPath)} was not found, " +
+          log"falling back to look up files added by Spark")
         val f = new File(SparkFiles.get(xsdPath.toString))
         if (f.exists()) {
           new FileInputStream(f)