[SPARK-47743][CORE] Use milliseconds as the time unit in logging

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

Use milliseconds as the time unit in logging

### Why are the changes needed?

During migrations, for the same log key `TIMEOUT` there are [log entry using seconds](https://github.com/apache/spark/blob/master/sql/hive-thriftserver/src/main/scala/org/apache/spark/sql/hive/thriftserver/SparkExecuteStatementOperation.scala#L145) and [another entry using ms](https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/executor/Executor.scala#L1017).
I suggest unify them and use milliseconds for all time related loggings.

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

No

### How was this patch tested?

Exising UT

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

No

Closes apache#45903 from gengliangwang/unify_time_unit.

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

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

@@ -89,7 +89,7 @@ object LogKey extends Enumeration {
   val SHUFFLE_ID = Value
   val SHUFFLE_MERGE_ID = Value
   val SIZE = Value
-  val SLEEP_TIME_SECONDS = Value
+  val SLEEP_TIME = Value
   val STAGE_ID = Value
   val SUBMISSION_ID = Value
   val SUBSAMPLING_RATE = Value
@@ -102,7 +102,8 @@ object LogKey extends Enumeration {
   val THREAD_NAME = Value
   val TID = Value
   val TIMEOUT = Value
-  val TIME_UNITS = Value
+  val TOTAL_EFFECTIVE_TIME = Value
+  val TOTAL_TIME = Value
   val URI = Value
   val USER_NAME = Value
   val WATERMARK_CONSTRAINT = Value

common/utils/src/main/scala/org/apache/spark/internal/README.md

@@ -7,6 +7,7 @@ LogKeys serve as identifiers for mapped diagnostic contexts (MDC) within logs. F
 * Use `UPPER_SNAKE_CASE` for key names.
 * Key names should be both simple and broad, yet include specific identifiers like `STAGE_ID`, `TASK_ID`, and `JOB_ID` when needed for clarity. For instance, use `MAX_ATTEMPTS` as a general key instead of creating separate keys for each scenario such as `EXECUTOR_STATE_SYNC_MAX_ATTEMPTS` and `MAX_TASK_FAILURES`. This balances simplicity with the detail needed for effective logging.
 * Use abbreviations in names if they are widely understood, such as `APP_ID` for APPLICATION_ID, and `K8S` for KUBERNETES.
+* For time-related keys, use milliseconds as the unit of time.
 
 ## Exceptions
 

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

@@ -42,7 +42,7 @@ import org.apache.spark._
 import org.apache.spark.errors.SparkCoreErrors
 import org.apache.spark.executor.DataReadMethod
 import org.apache.spark.internal.{config, Logging, MDC}
-import org.apache.spark.internal.LogKey.{BLOCK_ID, COUNT, SLEEP_TIME_SECONDS}
+import org.apache.spark.internal.LogKey.{BLOCK_ID, COUNT, SLEEP_TIME}
 import org.apache.spark.internal.config.{Network, RDD_CACHE_VISIBILITY_TRACKING_ENABLED, Tests}
 import org.apache.spark.memory.{MemoryManager, MemoryMode}
 import org.apache.spark.metrics.source.Source
@@ -616,7 +616,7 @@ private[spark] class BlockManager(
       shuffleManagerMeta)
 
     val MAX_ATTEMPTS = conf.get(config.SHUFFLE_REGISTRATION_MAX_ATTEMPTS)
-    val SLEEP_TIME_SECS = 5
+    val SLEEP_TIME_MS = 5000
 
     for (i <- 1 to MAX_ATTEMPTS) {
       try {
@@ -628,8 +628,8 @@ private[spark] class BlockManager(
         case e: Exception if i < MAX_ATTEMPTS =>
           logError(log"Failed to connect to external shuffle server, will retry " +
             log"${MDC(COUNT, MAX_ATTEMPTS - i)} more times after waiting " +
-            log"${MDC(SLEEP_TIME_SECONDS, SLEEP_TIME_SECS)} seconds...", e)
-          Thread.sleep(SLEEP_TIME_SECS * 1000L)
+            log"${MDC(SLEEP_TIME, SLEEP_TIME_MS)} ms...", e)
+          Thread.sleep(SLEEP_TIME_MS)
         case NonFatal(e) => throw SparkCoreErrors.unableToRegisterWithExternalShuffleServerError(e)
       }
     }

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

@@ -1641,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(log"Code generated in ${MDC(TIME_UNITS, timeMs)} ms")
+        logInfo(log"Code generated in ${MDC(TOTAL_TIME, timeMs)} ms")
         _compileTime.add(duration)
         result
     }

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

@@ -23,7 +23,7 @@ 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
+import org.apache.spark.sql.catalyst.util.DateTimeConstants.NANOS_PER_MILLIS
 import org.apache.spark.sql.catalyst.util.sideBySide
 import org.apache.spark.sql.errors.QueryExecutionErrors
 import org.apache.spark.sql.internal.SQLConf
@@ -88,15 +88,15 @@ 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 totalTime = metrics.time / NANOS_PER_MILLIS.toDouble
+    val totalTimeEffective = metrics.timeEffective / NANOS_PER_MILLIS.toDouble
     val message: MessageWithContext =
       log"""
          |=== Metrics of Executed Rules ===
          |Total number of runs: ${MDC(RULE_NUMBER_OF_RUNS, metrics.numRuns)}
-         |Total time: ${MDC(TIME_UNITS, totalTime)} seconds
+         |Total time: ${MDC(TOTAL_TIME, totalTime)} ms
          |Total number of effective runs: ${MDC(RULE_NUMBER_OF_RUNS, metrics.numEffectiveRuns)}
-         |Total time of effective runs: ${MDC(TIME_UNITS, totalTimeEffective)} seconds
+         |Total time of effective runs: ${MDC(TOTAL_EFFECTIVE_TIME, totalTimeEffective)} ms
       """.stripMargin
 
     logBasedOnLevel(message)