perf: Add metric for time spent casting in native scan

native/core/src/execution/operators/scan.rs

@@ -39,7 +39,9 @@ use arrow_data::ffi::FFI_ArrowArray;
 use arrow_data::ArrayData;
 use arrow_schema::ffi::FFI_ArrowSchema;
 use arrow_schema::{DataType, Field, Schema, SchemaRef};
-use datafusion::physical_plan::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
+use datafusion::physical_plan::metrics::{
+    BaselineMetrics, ExecutionPlanMetricsSet, MetricBuilder, MetricsSet, Time,
+};
 use datafusion::{
     execution::TaskContext,
     physical_expr::*,
@@ -345,16 +347,20 @@ struct ScanStream<'a> {
     baseline_metrics: BaselineMetrics,
     /// Cast options
     cast_options: CastOptions<'a>,
+    /// elapsed time for casting columns to different data types during scan
+    cast_time: Time,
 }
 
 impl<'a> ScanStream<'a> {
     pub fn new(scan: ScanExec, schema: SchemaRef, partition: usize) -> Self {
         let baseline_metrics = BaselineMetrics::new(&scan.metrics, partition);
+        let cast_time = MetricBuilder::new(&scan.metrics).subset_time("cast_time", partition);
         Self {
             scan,
             schema,
             baseline_metrics,
             cast_options: CastOptions::default(),
+            cast_time,
         }
     }
 
@@ -375,7 +381,10 @@ impl<'a> ScanStream<'a> {
             .zip(schema_fields.iter())
             .map(|(column, f)| {
                 if column.data_type() != f.data_type() {
-                    cast_with_options(column, f.data_type(), &self.cast_options)
+                    let mut timer = self.cast_time.timer();
+                    let cast_array = cast_with_options(column, f.data_type(), &self.cast_options);
+                    timer.stop();
+                    cast_array
                 } else {
                     Ok(Arc::clone(column))
                 }

spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala

@@ -79,6 +79,15 @@ object CometMetricNode {
         "total time (in ms) spent in this operator"))
   }
 
+  /**
+   * SQL Metrics for Comet native ScanExec
+   */
+  def scanMetrics(sc: SparkContext): Map[String, SQLMetric] = {
+    Map(
+      "cast_time" ->
+        SQLMetrics.createNanoTimingMetric(sc, "Total time for casting columns"))
+  }
+
   /**
    * SQL Metrics for DataFusion HashJoin
    */

spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala

@@ -198,9 +198,12 @@ case class CometScanExec(
     // Tracking scan time has overhead, we can't afford to do it for each row, and can only do
     // it for each batch.
     if (supportsColumnar) {
-      Some("scanTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "scan time"))
+      Map(
+        "scanTime" -> SQLMetrics.createNanoTimingMetric(
+          sparkContext,
+          "scan time")) ++ CometMetricNode.scanMetrics(sparkContext)
     } else {
-      None
+      Map.empty
     }
   } ++ {
     relation.fileFormat match {

spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala

@@ -517,6 +517,28 @@ class CometExecSuite extends CometTestBase {
     }
   }
 
+  test("Comet native metrics: scan") {
+    withSQLConf(CometConf.COMET_EXEC_ENABLED.key -> "true") {
+      withTempDir { dir =>
+        val path = new Path(dir.toURI.toString, "native-scan.parquet")
+        makeParquetFileAllTypes(path, dictionaryEnabled = true, 10000)
+        withParquetTable(path.toString, "tbl") {
+          val df = sql("SELECT * FROM tbl WHERE _2 > _3")
+          df.collect()
+
+          val metrics = find(df.queryExecution.executedPlan)(_.isInstanceOf[CometScanExec])
+            .map(_.metrics)
+            .get
+
+          assert(metrics.contains("scanTime"))
+          assert(metrics.contains("cast_time"))
+          assert(metrics("scanTime").value > 0)
+          assert(metrics("cast_time").value > 0)
+        }
+      }
+    }
+  }
+
   test("Comet native metrics: project and filter") {
     withSQLConf(CometConf.COMET_EXEC_ENABLED.key -> "true") {
       withParquetTable((0 until 5).map(i => (i, i + 1)), "tbl") {