Update logging

spark/src/main/scala/org/apache/spark/sql/delta/MetadataCleanup.scala

@@ -128,8 +128,8 @@ trait MetadataCleanup extends DeltaLogging {
           sidecarDeletionMetrics)
         logInfo(log"Sidecar deletion metrics: ${MDC(DeltaLogKeys.METRICS, sidecarDeletionMetrics)}")
       }
-      logInfo(log"Deleted ${MDC(DeltaLogKeys.NUM_FILES, numDeleted)} log files and " +
-        log"${MDC(DeltaLogKeys.NUM_FILES2, numDeletedUnbackfilled)} unbackfilled commit " +
+      logInfo(log"Deleted ${MDC(DeltaLogKeys.NUM_FILES, numDeleted.toLong)} log files and " +
+        log"${MDC(DeltaLogKeys.NUM_FILES2, numDeletedUnbackfilled.toLong)} unbackfilled commit " +
         log"files older than ${MDC(DeltaLogKeys.DATE, formattedDate)}")
     }
   }

spark/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala

@@ -1881,7 +1881,7 @@ trait OptimisticTransactionImpl extends TransactionalWrite
 
     logInfo(log"Committed delta #${MDC(DeltaLogKeys.VERSION, attemptVersion)} to " +
       log"${MDC(DeltaLogKeys.PATH, deltaLog.logPath)}. Wrote " +
-      log"${MDC(DeltaLogKeys.NUM_ACTIONS, commitSize)} actions.")
+      log"${MDC(DeltaLogKeys.NUM_ACTIONS, commitSize.toLong)} actions.")
 
     deltaLog.checkpoint(currentSnapshot)
     currentSnapshot
@@ -2044,7 +2044,7 @@ trait OptimisticTransactionImpl extends TransactionalWrite
              |Detected mismatch in partition values between AddFile and table metadata but
              |commit validation was turned off.
              |To turn it back on set
-             |${MDC(DeltaLogKeys.CONFIG_KEY, DeltaSQLConf.DELTA_COMMIT_VALIDATION_ENABLED)}
+             |${MDC(DeltaLogKeys.CONFIG_KEY, DeltaSQLConf.DELTA_COMMIT_VALIDATION_ENABLED.key)}
              |to "true"
           """.stripMargin)
         a
@@ -2284,7 +2284,7 @@ trait OptimisticTransactionImpl extends TransactionalWrite
     val actions = currentTransactionInfo.finalActionsToCommit
     logInfo(
       log"Attempting to commit version ${MDC(DeltaLogKeys.VERSION, attemptVersion)} with " +
-      log"${MDC(DeltaLogKeys.NUM_ACTIONS, actions.size)} actions with " +
+      log"${MDC(DeltaLogKeys.NUM_ACTIONS, actions.size.toLong)} actions with " +
       log"${MDC(DeltaLogKeys.ISOLATION_LEVEL, isolationLevel)} isolation level")
 
     if (readVersion > -1 && metadata.id != snapshot.metadata.id) {
@@ -2588,7 +2588,7 @@ trait OptimisticTransactionImpl extends TransactionalWrite
         log"${MDC(DeltaLogKeys.NUM_ACTIONS, adds)} adds, " +
         log"${MDC(DeltaLogKeys.NUM_ACTIONS2, removes)} removes, " +
         log"${MDC(DeltaLogKeys.NUM_PREDICATES, readPredicates.size)} read predicates, " +
-        log"${MDC(DeltaLogKeys.NUM_FILES, readFiles.size)} read files"
+        log"${MDC(DeltaLogKeys.NUM_FILES, readFiles.size.toLong)} read files"
       }
 
       logInfo(logPrefix +

spark/src/main/scala/org/apache/spark/sql/delta/commands/ConvertToDeltaCommand.scala

@@ -304,7 +304,8 @@ abstract class ConvertToDeltaCommandBase(
           _, txn.deltaLog.dataPath, fs, conf, Some(partitionSchema), deltaPath.isDefined))
       if (shouldCollectStats) {
         logInfo(
-          log"Collecting stats for a batch of ${MDC(DeltaLogKeys.NUM_FILES, batch.size)} files; " +
+          log"Collecting stats for a batch of " +
+          log"${MDC(DeltaLogKeys.NUM_FILES, batch.size.toLong)} files; " +
           log"finished ${MDC(DeltaLogKeys.NUM_FILES2, numFiles)} so far"
           )
         numFiles += statsBatchSize

spark/src/main/scala/org/apache/spark/sql/delta/commands/ReorgTableForUpgradeUniformHelper.scala

@@ -114,7 +114,7 @@ trait ReorgTableForUpgradeUniformHelper extends DeltaLogging {
       log"at version ${MDC(DeltaLogKeys.VERSION, snapshot.version)}, there are " +
       log"${MDC(DeltaLogKeys.NUM_FILES, numOfAddFiles)} addFiles, and " +
       log"${MDC(DeltaLogKeys.NUM_FILES2, numOfAddFilesWithTag)} addFiles with " +
-      log"ICEBERG_COMPAT_VERSION=${MDC(DeltaLogKeys.VERSION2, icebergCompatVersion)} tag.")
+      log"ICEBERG_COMPAT_VERSION=${MDC(DeltaLogKeys.VERSION2, icebergCompatVersion.toLong)} tag.")
     (numOfAddFiles, numOfAddFilesWithTag)
   }
 

spark/src/main/scala/org/apache/spark/sql/delta/commands/VacuumCommand.scala

@@ -423,10 +423,10 @@ object VacuumCommand extends VacuumCommandImpl with Serializable {
             )
 
             recordDeltaEvent(deltaLog, "delta.gc.stats", data = stats)
-            logInfo(log"Found ${MDC(DeltaLogKeys.NUM_FILES, numFiles)} files " +
+            logInfo(log"Found ${MDC(DeltaLogKeys.NUM_FILES, numFiles.toLong)} files " +
               log"(${MDC(DeltaLogKeys.NUM_BYTES, sizeOfDataToDelete)} bytes) and directories in " +
               log"a total of ${MDC(DeltaLogKeys.NUM_DIRS, dirCounts)} directories " +
-              log"that are safe to delete. Vacuum stats: ${MDC(DeltaLogKeys.STATS, stats)}")
+              log"that are safe to delete. Vacuum stats: ${MDC(DeltaLogKeys.VACUUM_STATS, stats)}")
 
             return diffFiles.map(f => urlEncodedStringToPath(f).toString).toDF("path")
           }

spark/src/main/scala/org/apache/spark/sql/delta/commands/backfill/BackfillBatch.scala

@@ -68,8 +68,8 @@ trait BackfillBatch extends DeltaLogging {
       )
     }
 
-    logInfo(log"Batch ${MDC(DeltaLogKeys.BATCH_ID, batchId)} starting, committing " +
-      log"${MDC(DeltaLogKeys.NUM_FILES, filesInBatch.size)} candidate files")
+    logInfo(log"Batch ${MDC(DeltaLogKeys.BATCH_ID, batchId.toLong)} starting, committing " +
+      log"${MDC(DeltaLogKeys.NUM_FILES, filesInBatch.size.toLong)} candidate files")
     // This step is necessary to mark all files in this batch as "read" in the
     // child transaction object `txn` and to set the read transactions ids to be the same as the
     // parent transaction object `origTxn`, for proper conflict checking.

spark/src/main/scala/org/apache/spark/sql/delta/hooks/GenerateSymlinkManifest.scala

@@ -337,7 +337,7 @@ trait GenerateSymlinkManifestImpl extends PostCommitHook with DeltaLogging with
     }
 
     logInfo(log"Deleted manifest partitions [" +
-      log"${MDC(DeltaLogKeys.NUM_FILES, partitionRelativePathsToDelete.size)}]:\n\t" +
+      log"${MDC(DeltaLogKeys.NUM_FILES, partitionRelativePathsToDelete.size.toLong)}]:\n\t" +
       log"${MDC(DeltaLogKeys.PATHS, partitionRelativePathsToDelete.mkString("\n\t"))}")
   }
 

spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSink.scala

@@ -93,8 +93,8 @@ case class DeltaSink(
       logInfo(
         log"Committed transaction, batchId=${MDC(DeltaLogKeys.BATCH_ID, batchId)}, " +
         log"duration=${MDC(DeltaLogKeys.DURATION, durationMs)} ms, " +
-        log"added ${MDC(DeltaLogKeys.NUM_FILES, newFiles.size)} files, " +
-        log"removed ${MDC(DeltaLogKeys.NUM_FILES2, deletedFiles.size)} files.")
+        log"added ${MDC(DeltaLogKeys.NUM_FILES, newFiles.size.toLong)} files, " +
+        log"removed ${MDC(DeltaLogKeys.NUM_FILES2, deletedFiles.size.toLong)} files.")
       val executionId = sc.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
       SQLMetrics.postDriverMetricUpdates(sc, executionId, metrics.values.toSeq)
     }
@@ -153,8 +153,8 @@ case class DeltaSink(
     val totalSize = newFiles.map(_.getFileSize).sum
     val totalLogicalRecords = newFiles.map(_.numLogicalRecords.getOrElse(0L)).sum
     logInfo(
-      log"Wrote ${MDC(DeltaLogKeys.NUM_FILES, newFiles.size)} files, with total size " +
-      log"${MDC(DeltaLogKeys.NUM_BYTES, totalSize)}, " +
+      log"Wrote ${MDC(DeltaLogKeys.NUM_FILES, newFiles.size.toLong)} files, " +
+        log"with total size ${MDC(DeltaLogKeys.NUM_BYTES, totalSize)}, " +
       log"${MDC(DeltaLogKeys.NUM_RECORDS, totalLogicalRecords)} logical records, " +
       log"duration=${MDC(DeltaLogKeys.DURATION, writeFilesTimeMs)} ms.")
 

spark/src/main/scala/org/apache/spark/sql/delta/util/DeltaFileOperations.scala

@@ -128,7 +128,7 @@ object DeltaFileOperations extends DeltaLogging {
       base: Int = 100,
       jitter: Int = 1000): Unit = {
     val sleepTime = Random.nextInt(jitter) + base
-    logWarning(log"Sleeping for ${MDC(DeltaLogKeys.TIME_MS, sleepTime)} ms to rate limit " +
+    logWarning(log"Sleeping for ${MDC(DeltaLogKeys.TIME_MS, sleepTime.toLong)} ms to rate limit " +
       log"${MDC(DeltaLogKeys.OP_NAME, opName)}", t)
     Thread.sleep(sleepTime)
   }