[CARMEL-6355][FOLLOWUP] Auto update table statistics based on write metrics (#1142)

Author: wakun

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/BasicWriteStatsTracker.scala

@@ -185,8 +185,12 @@ object BasicWriteJobStatsTracker {
   }
 
   def getWriteStats(mode: SaveMode, metrics: Map[String, SQLMetric]): Option[WriteStats] = {
-    val numBytes = metrics.get(NUM_OUTPUT_BYTES_KEY).map(_.value).map(BigInt(_))
-    val numRows = metrics.get(NUM_OUTPUT_ROWS_KEY).map(_.value).map(BigInt(_))
-    numBytes.map(WriteStats(mode, _, numRows))
+    if (metrics.contains(NUM_OUTPUT_BYTES_KEY) && metrics.contains(NUM_OUTPUT_ROWS_KEY)) {
+      val numBytes = metrics.get(NUM_OUTPUT_BYTES_KEY).map(_.value).map(BigInt(_))
+      val numRows = metrics.get(NUM_OUTPUT_ROWS_KEY).map(_.value).map(BigInt(_))
+      numBytes.map(WriteStats(mode, _, numRows))
+    } else {
+      None
+    }
   }
 }