[SPARK-47688][CORE] Support three methods of the log concatenation in the structured logging framework

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

@@ -48,6 +48,19 @@ case class MessageWithContext(message: String, context: java.util.HashMap[String
     resultMap.putAll(mdc.context)
     MessageWithContext(message + mdc.message, resultMap)
   }
+
+  def ++(s: String): MessageWithContext = {
+    MessageWithContext(message + s, context)
+  }
+}
+
+object MDC {
+
+  implicit class StringImprovements(val s: String) {
+    def ++(mdc: MessageWithContext): MessageWithContext = {
+     MessageWithContext(s + mdc.message, mdc.context)
+    }
+  }
 }
 
 /**
@@ -96,17 +109,21 @@ trait Logging {
   }
 
   implicit class LogStringContext(val sc: StringContext) {
-    def log(args: MDC*): MessageWithContext = {
+    def log(args: Any*): MessageWithContext = {
       val processedParts = sc.parts.iterator
       val sb = new StringBuilder(processedParts.next())
       val context = new java.util.HashMap[String, String]()
 
-      args.foreach { mdc =>
-        sb.append(mdc.value)
-        if (Logging.isStructuredLoggingEnabled) {
-          context.put(mdc.key.toString.toLowerCase(Locale.ROOT), mdc.value)
+      args.foreach { arg =>
+        arg match {
+          case MDC(k, v) =>
+            sb.append(v)
+            if (Logging.isStructuredLoggingEnabled) {
+              context.put(k.toString.toLowerCase(Locale.ROOT), v)
+            }
+          case v: Any =>
+            sb.append(v.toString)
         }
-
         if (processedParts.hasNext) {
           sb.append(processedParts.next())
         }

common/utils/src/test/scala/org/apache/spark/util/PatternLoggingSuite.scala

@@ -40,6 +40,18 @@ class PatternLoggingSuite extends LoggingSuiteBase with BeforeAndAfterAll {
   override def expectedPatternForMsgWithMDCAndException(level: Level): String =
     s""".*$level $className: Error in executor 1.\njava.lang.RuntimeException: OOM\n.*"""
 
+  override def expectedPatternForVariableAndMDC(level: Level): String = {
+     s""".*$level $className: Hello This is a log message, Lost executor 1.\n"""
+  }
+
+  override def expectedPatternForConcatStringAndMDC(level: Level): String = {
+    s""".*$level $className: Hello This is a log message, Lost executor 1.\n"""
+  }
+
+  override def expectedPatternForConcatMDCAndString(level: Level): String = {
+    s""".*$level $className: Lost executor 1. Hello This is a log message.\n"""
+  }
+
   override def verifyMsgWithConcat(level: Level, logOutput: String): Unit = {
     val pattern =
       s""".*$level $className: Min Size: 2, Max Size: 4. Please double check.\n"""

common/utils/src/test/scala/org/apache/spark/util/StructuredLoggingSuite.scala

@@ -26,6 +26,7 @@ import org.scalatest.funsuite.AnyFunSuite // scalastyle:ignore funsuite
 
 import org.apache.spark.internal.{LogEntry, Logging, MDC}
 import org.apache.spark.internal.LogKey.{EXECUTOR_ID, MAX_SIZE, MIN_SIZE}
+import org.apache.spark.internal.MDC._
 
 trait LoggingSuiteBase
     extends AnyFunSuite // scalastyle:ignore funsuite
@@ -63,12 +64,27 @@ trait LoggingSuiteBase
     log"Max Size: ${MDC(MAX_SIZE, "4")}. " +
     log"Please double check."
 
+  def logVariableAndMDC: LogEntry =
+    log"Hello $basicMsg, Lost executor ${MDC(EXECUTOR_ID, "1")}."
+
+  def concatMDCAndString: LogEntry =
+    (log"Lost executor ${MDC(EXECUTOR_ID, "1")}." ++ s" Hello $basicMsg.")
+
+  def concatStringAndMDC: LogEntry =
+    (s"Hello $basicMsg, " ++ log"Lost executor ${MDC(EXECUTOR_ID, "1")}.")
+
   def expectedPatternForMsgWithMDC(level: Level): String
 
   def expectedPatternForMsgWithMDCAndException(level: Level): String
 
   def verifyMsgWithConcat(level: Level, logOutput: String): Unit
 
+  def expectedPatternForVariableAndMDC(level: Level): String
+
+  def expectedPatternForConcatStringAndMDC(level: Level): String
+
+  def expectedPatternForConcatMDCAndString(level: Level): String
+
   test("Basic logging") {
     Seq(
       (Level.ERROR, () => logError(basicMsg)),
@@ -112,6 +128,39 @@ trait LoggingSuiteBase
           verifyMsgWithConcat(level, logOutput)
       }
   }
+
+  test("Logging variable and MDC") {
+    Seq(
+      (Level.ERROR, () => logError(logVariableAndMDC)),
+      (Level.WARN, () => logWarning(logVariableAndMDC)),
+      (Level.INFO, () => logInfo(logVariableAndMDC))).foreach {
+        case (level, logFunc) =>
+          val logOutput = captureLogOutput(logFunc)
+          assert(expectedPatternForVariableAndMDC(level).r.matches(logOutput))
+      }
+  }
+
+  test("Logging concat string and MDC") {
+    Seq(
+      (Level.ERROR, () => logError(concatStringAndMDC)),
+      (Level.WARN, () => logWarning(concatStringAndMDC)),
+      (Level.INFO, () => logInfo(concatStringAndMDC))).foreach {
+        case (level, logFunc) =>
+          val logOutput = captureLogOutput(logFunc)
+          assert(expectedPatternForConcatStringAndMDC(level).r.matches(logOutput))
+      }
+  }
+
+  test("Logging concat MDC and string") {
+    Seq(
+      (Level.ERROR, () => logError(concatMDCAndString)),
+      (Level.WARN, () => logWarning(concatMDCAndString)),
+      (Level.INFO, () => logInfo(concatMDCAndString))).foreach {
+        case (level, logFunc) =>
+          val logOutput = captureLogOutput(logFunc)
+          assert(expectedPatternForConcatMDCAndString(level).r.matches(logOutput))
+      }
+  }
 }
 
 class StructuredLoggingSuite extends LoggingSuiteBase {
@@ -170,6 +219,48 @@ class StructuredLoggingSuite extends LoggingSuiteBase {
         }""")
   }
 
+  override def expectedPatternForVariableAndMDC(level: Level): String = {
+    compactAndToRegexPattern(
+      s"""
+        {
+          "ts": "<timestamp>",
+          "level": "$level",
+          "msg": "Hello This is a log message, Lost executor 1.",
+          "context": {
+             "executor_id": "1"
+          },
+          "logger": "$className"
+        }""")
+  }
+
+  override def expectedPatternForConcatStringAndMDC(level: Level): String = {
+    compactAndToRegexPattern(
+      s"""
+        {
+          "ts": "<timestamp>",
+          "level": "$level",
+          "msg": "Hello This is a log message, Lost executor 1.",
+          "context": {
+             "executor_id": "1"
+          },
+          "logger": "$className"
+        }""")
+  }
+
+  override def expectedPatternForConcatMDCAndString(level: Level): String = {
+    compactAndToRegexPattern(
+      s"""
+        {
+          "ts": "<timestamp>",
+          "level": "$level",
+          "msg": "Lost executor 1. Hello This is a log message.",
+          "context": {
+             "executor_id": "1"
+          },
+          "logger": "$className"
+        }""")
+  }
+
   override def verifyMsgWithConcat(level: Level, logOutput: String): Unit = {
     val pattern1 = compactAndToRegexPattern(
       s"""