[CARMEL-6135] Report Selected Buckets Number and Tasks Count when Bucket Scan Disabled by Planner (#1034)

Author: xingchaozh

sql/core/src/main/scala/org/apache/spark/sql/execution/DataSourceScanExec.scala

@@ -521,22 +521,22 @@ case class FileSourceScanExec(
 
     val withSelectedBucketsCount = relation.bucketSpec.map { spec =>
       val bucketedKey = "Bucketed"
-      if (bucketedScan) {
-        val numSelectedBuckets = optionalBucketSet.map { b =>
-          b.cardinality()
-        } getOrElse {
-          spec.numBuckets
-        }
-        metadata ++ Map(
-          bucketedKey -> "true",
-          "SelectedBucketsCount" -> (s"$numSelectedBuckets out of ${spec.numBuckets}"))
+      val withBucketedScanStatus = if (bucketedScan) {
+        metadata + (bucketedKey -> "true")
       } else if (!relation.sparkSession.sessionState.conf.bucketingEnabled) {
         metadata + (bucketedKey -> "false (disabled by configuration)")
       } else if (disableBucketedScan) {
         metadata + (bucketedKey -> "false (disabled by query planner)")
       } else {
         metadata + (bucketedKey -> "false (bucket column(s) not read)")
       }
+      val numSelectedBuckets = optionalBucketSet.map { b =>
+        b.cardinality()
+      } getOrElse {
+        spec.numBuckets
+      }
+      withBucketedScanStatus + ("SelectedBucketsCount" ->
+        s"$numSelectedBuckets out of ${spec.numBuckets}")
     } getOrElse {
       metadata
     }
@@ -634,6 +634,7 @@ case class FileSourceScanExec(
       val filesSize = sqlContext.queryLoadLimitationManager.
         checkScanSize(groupId, readRDD, bucketedScan, cancelQuery = applyIndexPrune)
       driverMetrics("filesSize") = filesSize
+      driverMetrics("readRDDPartitionsNumber") = readRDD.getNumPartitions
       readRDD
     }
 
@@ -696,7 +697,9 @@ case class FileSourceScanExec(
       SQLMetrics.createMetric(sparkContext, "local files replace time (ms)"),
     "filesSize" -> SQLMetrics.createSizeMetric(sparkContext, "size of files read"),
     "pruningTime" ->
-      SQLMetrics.createTimingMetric(sparkContext, "dynamic partition pruning time")
+      SQLMetrics.createTimingMetric(sparkContext, "dynamic partition pruning time"),
+    "readRDDPartitionsNumber" -> SQLMetrics.createMetric(sparkContext,
+      "the number of partitions of read RDD")
   ) ++ {
     // Tracking scan time has overhead, we can't afford to do it for each row, and can only do
     // it for each batch.