[VL] Add write IO metrics for WriteFiles

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxMetricsApi.scala

@@ -341,6 +341,7 @@ class VeloxMetricsApi extends MetricsApi with Logging {
       "physicalWrittenBytes" -> SQLMetrics.createSizeMetric(
         sparkContext,
         "number of written bytes"),
+      "writeIONanos" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time of write IO"),
       "wallNanos" -> SQLMetrics.createNanoTimingMetric(sparkContext, "time of write"),
       "numWrittenFiles" -> SQLMetrics.createMetric(sparkContext, "number of written files")
     )

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxMetricsSuite.scala

@@ -195,6 +195,7 @@ class VeloxMetricsSuite extends VeloxWholeStageTransformerSuite with AdaptiveSpa
             assert(write.isDefined)
             val metrics = write.get.metrics
             assert(metrics("physicalWrittenBytes").value > 0)
+            assert(metrics("writeIONanos").value > 0)
             assert(metrics("numWrittenFiles").value == 1)
         }
       }

cpp/core/jni/JniWrapper.cc

@@ -175,7 +175,7 @@ jint JNI_OnLoad(JavaVM* vm, void* reserved) {
   metricsBuilderClass = createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/metrics/Metrics;");
 
   metricsBuilderConstructor = getMethodIdOrError(
-      env, metricsBuilderClass, "<init>", "([J[J[J[J[J[J[J[J[J[JJ[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J)V");
+      env, metricsBuilderClass, "<init>", "([J[J[J[J[J[J[J[J[J[JJ[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J[J)V");
 
   nativeColumnarToRowInfoClass =
       createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/vectorized/NativeColumnarToRowInfo;");
@@ -496,6 +496,7 @@ JNIEXPORT jobject JNICALL Java_org_apache_gluten_vectorized_ColumnarBatchOutIter
       longArray[Metrics::kIoWaitTime],
       longArray[Metrics::kPreloadSplits],
       longArray[Metrics::kPhysicalWrittenBytes],
+      longArray[Metrics::kWriteIOTime],
       longArray[Metrics::kNumWrittenFiles]);
 
   JNI_METHOD_END(nullptr)

cpp/core/utils/metrics.h

@@ -76,6 +76,7 @@ struct Metrics {
 
     // Write metrics.
     kPhysicalWrittenBytes,
+    kWriteIOTime,
     kNumWrittenFiles,
 
     // The end of enum items.

cpp/velox/compute/WholeStageResultIterator.cc

@@ -46,6 +46,7 @@ const std::string kRemainingFilterTime = "totalRemainingFilterTime";
 const std::string kIoWaitTime = "ioWaitNanos";
 const std::string kPreloadSplits = "readyPreloadedSplits";
 const std::string kNumWrittenFiles = "numWrittenFiles";
+const std::string kWriteIOTime = "writeIOTime";
 
 // others
 const std::string kHiveDefaultPartition = "__HIVE_DEFAULT_PARTITION__";
@@ -391,6 +392,7 @@ void WholeStageResultIterator::collectMetrics() {
       metrics_->get(Metrics::kNumWrittenFiles)[metricIndex] =
           runtimeMetric("sum", entry.second->customStats, kNumWrittenFiles);
       metrics_->get(Metrics::kPhysicalWrittenBytes)[metricIndex] = second->physicalWrittenBytes;
+      metrics_->get(Metrics::kWriteIOTime)[metricIndex] = runtimeMetric("sum", second->customStats, kWriteIOTime);
 
       metricIndex += 1;
     }

gluten-data/src/main/java/org/apache/gluten/metrics/Metrics.java

@@ -50,7 +50,7 @@ public class Metrics implements IMetrics {
   public long[] preloadSplits;
 
   public long[] physicalWrittenBytes;
-
+  public long[] writeIOTime;
   public long[] numWrittenFiles;
 
   public SingleMetric singleMetric = new SingleMetric();
@@ -88,6 +88,7 @@ public Metrics(
       long[] ioWaitTime,
       long[] preloadSplits,
       long[] physicalWrittenBytes,
+      long[] writeIOTime,
       long[] numWrittenFiles) {
     this.inputRows = inputRows;
     this.inputVectors = inputVectors;
@@ -120,6 +121,7 @@ public Metrics(
     this.ioWaitTime = ioWaitTime;
     this.preloadSplits = preloadSplits;
     this.physicalWrittenBytes = physicalWrittenBytes;
+    this.writeIOTime = writeIOTime;
     this.numWrittenFiles = numWrittenFiles;
   }
 
@@ -159,6 +161,7 @@ public OperatorMetrics getOperatorMetrics(int index) {
         ioWaitTime[index],
         preloadSplits[index],
         physicalWrittenBytes[index],
+        writeIOTime[index],
         numWrittenFiles[index]);
   }
 

gluten-data/src/main/java/org/apache/gluten/metrics/OperatorMetrics.java

@@ -48,7 +48,7 @@ public class OperatorMetrics implements IOperatorMetrics {
   public long preloadSplits;
 
   public long physicalWrittenBytes;
-
+  public long writeIOTime;
   public long numWrittenFiles;
 
   /** Create an instance for operator metrics. */
@@ -83,6 +83,7 @@ public OperatorMetrics(
       long ioWaitTime,
       long preloadSplits,
       long physicalWrittenBytes,
+      long writeIOTime,
       long numWrittenFiles) {
     this.inputRows = inputRows;
     this.inputVectors = inputVectors;
@@ -114,6 +115,7 @@ public OperatorMetrics(
     this.ioWaitTime = ioWaitTime;
     this.preloadSplits = preloadSplits;
     this.physicalWrittenBytes = physicalWrittenBytes;
+    this.writeIOTime = writeIOTime;
     this.numWrittenFiles = numWrittenFiles;
   }
 }

gluten-data/src/main/scala/org/apache/gluten/metrics/MetricsUtil.scala

@@ -100,6 +100,7 @@ object MetricsUtil extends Logging {
     val outputBytes = operatorMetrics.get(0).outputBytes
 
     val physicalWrittenBytes = operatorMetrics.get(0).physicalWrittenBytes
+    val writeIOTime = operatorMetrics.get(0).writeIOTime
 
     var cpuCount: Long = 0
     var wallNanos: Long = 0
@@ -182,6 +183,7 @@ object MetricsUtil extends Logging {
       ioWaitTime,
       preloadSplits,
       physicalWrittenBytes,
+      writeIOTime,
       numWrittenFiles
     )
   }

gluten-data/src/main/scala/org/apache/gluten/metrics/WriteFilesMetricsUpdater.scala

@@ -24,6 +24,7 @@ class WriteFilesMetricsUpdater(val metrics: Map[String, SQLMetric]) extends Metr
     if (opMetrics != null) {
       val operatorMetrics = opMetrics.asInstanceOf[OperatorMetrics]
       metrics("physicalWrittenBytes") += operatorMetrics.physicalWrittenBytes
+      metrics("writeIONanos") += operatorMetrics.writeIOTime
       metrics("wallNanos") += operatorMetrics.wallNanos
       metrics("numWrittenFiles") += operatorMetrics.numWrittenFiles
     }