[SPARK-49551][SS] Improve RocksDB log for replayChangelog

### What changes were proposed in this pull request? A better log for replayChangelog() in rocksDB.scala. The code change is straightforward. ### Why are the changes needed? Log improvement ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? No need ### Was this patch authored or co-authored using generative AI tooling? No Closes #48039 from WweiL/SPARK-49551-rocksdb-log-improve. Authored-by: Wei Liu <[email protected]> Signed-off-by: Jungtaek Lim <[email protected]>
apache · Sep 9, 2024 · 14e4908 · 14e4908
1 parent 20643bb
commit 14e4908
Showing 1 changed file with 5 additions and 3 deletions.
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/RocksDB.scala
@@ -407,10 +407,12 @@ class RocksDB(
    * Replay change log from the loaded version to the target version.
    */
   private def replayChangelog(endVersion: Long): Unit = {
+    logInfo(log"Replaying changelog from version " +
+      log"${MDC(LogKeys.LOADED_VERSION, loadedVersion)} -> " +
+      log"${MDC(LogKeys.END_VERSION, endVersion)}")
     for (v <- loadedVersion + 1 to endVersion) {
-      logInfo(log"replaying changelog from version " +
-        log"${MDC(LogKeys.LOADED_VERSION, loadedVersion)} -> " +
-        log"${MDC(LogKeys.END_VERSION, endVersion)}")
+      logInfo(log"Replaying changelog on version " +
+        log"${MDC(LogKeys.VERSION_NUM, v)}")
       var changelogReader: StateStoreChangelogReader = null
       try {
         changelogReader = fileManager.getChangelogReader(v, useColumnFamilies)