Spark 4.1: Add per-task scanDuration metric

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/BaseReader.java

@@ -77,6 +77,8 @@ abstract class BaseReader<T, TaskT extends ScanTask> implements Closeable {
   private final Iterator<TaskT> tasks;
   private final DeleteCounter counter;
   private final boolean cacheDeleteFilesOnExecutors;
+  // nanos, not millis: a single split can be read in well under a millisecond
+  private long scanDurationNanos = 0L;
 
   private Map<String, InputFile> lazyInputFiles;
   private CloseableIterator<T> currentIterator;
@@ -132,7 +134,13 @@ protected DeleteCounter counter() {
     return counter;
   }
 
+  /** Wall time spent reading in this task, in nanoseconds. */
+  protected long scanDurationNanos() {
+    return scanDurationNanos;
+  }
+
   public boolean next() throws IOException {
+    long start = System.nanoTime();
     try {
       while (true) {
         if (currentIterator.hasNext()) {
@@ -156,6 +164,8 @@ public boolean next() throws IOException {
         LOG.error("Error reading file(s): {}", filePaths, e);
       }
       throw e;
+    } finally {
+      this.scanDurationNanos += System.nanoTime() - start;
     }
   }
 

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java

@@ -33,6 +33,7 @@
 import org.apache.iceberg.spark.ParquetBatchReadConf;
 import org.apache.iceberg.spark.source.metrics.TaskNumDeletes;
 import org.apache.iceberg.spark.source.metrics.TaskNumSplits;
+import org.apache.iceberg.spark.source.metrics.TaskScanDuration;
 import org.apache.spark.rdd.InputFileBlockHolder;
 import org.apache.spark.sql.connector.metric.CustomTaskMetric;
 import org.apache.spark.sql.connector.read.PartitionReader;
@@ -88,7 +89,9 @@ class BatchDataReader extends BaseBatchReader<FileScanTask>
   @Override
   public CustomTaskMetric[] currentMetricsValues() {
     return new CustomTaskMetric[] {
-      new TaskNumSplits(numSplits), new TaskNumDeletes(counter().get())
+      new TaskNumSplits(numSplits),
+      new TaskNumDeletes(counter().get()),
+      new TaskScanDuration(scanDurationNanos())
     };
   }
 

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java

@@ -33,6 +33,7 @@
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.spark.source.metrics.TaskNumDeletes;
 import org.apache.iceberg.spark.source.metrics.TaskNumSplits;
+import org.apache.iceberg.spark.source.metrics.TaskScanDuration;
 import org.apache.spark.rdd.InputFileBlockHolder;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.connector.metric.CustomTaskMetric;
@@ -72,7 +73,9 @@ class RowDataReader extends BaseRowReader<FileScanTask> implements PartitionRead
   @Override
   public CustomTaskMetric[] currentMetricsValues() {
     return new CustomTaskMetric[] {
-      new TaskNumSplits(numSplits), new TaskNumDeletes(counter().get())
+      new TaskNumSplits(numSplits),
+      new TaskNumDeletes(counter().get()),
+      new TaskScanDuration(scanDurationNanos())
     };
   }
 

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java

@@ -49,6 +49,7 @@
 import org.apache.iceberg.spark.source.metrics.PositionalDeleteFiles;
 import org.apache.iceberg.spark.source.metrics.ResultDataFiles;
 import org.apache.iceberg.spark.source.metrics.ResultDeleteFiles;
+import org.apache.iceberg.spark.source.metrics.ScanDuration;
 import org.apache.iceberg.spark.source.metrics.ScannedDataManifests;
 import org.apache.iceberg.spark.source.metrics.ScannedDeleteManifests;
 import org.apache.iceberg.spark.source.metrics.SkippedDataFiles;
@@ -345,6 +346,7 @@ public CustomMetric[] supportedCustomMetrics() {
 
       // common
       new TotalPlanningDuration(),
+      new ScanDuration(),
 
       // data manifests
       new TotalDataManifests(),

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/metrics/ScanDuration.java

@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source.metrics;
+
+import java.util.Arrays;
+import java.util.Locale;
+import java.util.concurrent.TimeUnit;
+import org.apache.spark.sql.connector.metric.CustomSumMetric;
+
+public class ScanDuration extends CustomSumMetric {
+
+  public static final String NAME = "scanDuration";
+
+  @Override
+  public String name() {
+    return NAME;
+  }
+
+  @Override
+  public String description() {
+    return "total scan duration";
+  }
+
+  @Override
+  public String aggregateTaskMetrics(long[] taskMetrics) {
+    if (taskMetrics.length == 0) {
+      return format(0L);
+    }
+
+    long totalNanos = 0L;
+    for (long taskMetric : taskMetrics) {
+      totalNanos += taskMetric;
+    }
+
+    // sort a copy: Spark reuses the array it hands us
+    long[] sorted = Arrays.copyOf(taskMetrics, taskMetrics.length);
+    Arrays.sort(sorted);
+
+    // mirrors MetricUtils.stringValue for Spark's own timing metrics, so a straggler task is
+    // visible instead of being hidden by the total
+    return String.format(
+        Locale.ROOT,
+        "total (min, med, max)\n%s (%s, %s, %s)",
+        format(totalNanos),
+        format(sorted[0]),
+        format(sorted[sorted.length / 2]),
+        format(sorted[sorted.length - 1]));
+  }
+
+  private String format(long nanos) {
+    // raw nanos are unreadable on the UI, scale to the largest meaningful unit
+    if (nanos < TimeUnit.MICROSECONDS.toNanos(1)) {
+      return nanos + " ns";
+    } else if (nanos < TimeUnit.MILLISECONDS.toNanos(1)) {
+      return TimeUnit.NANOSECONDS.toMicros(nanos) + " us";
+    } else if (nanos < TimeUnit.SECONDS.toNanos(1)) {
+      return TimeUnit.NANOSECONDS.toMillis(nanos) + " ms";
+    } else {
+      return String.format(Locale.ROOT, "%.1f s", nanos / (double) TimeUnit.SECONDS.toNanos(1));
+    }
+  }
+}

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/metrics/TaskScanDuration.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source.metrics;
+
+import org.apache.spark.sql.connector.metric.CustomTaskMetric;
+
+public class TaskScanDuration implements CustomTaskMetric {
+
+  private final long value;
+
+  public TaskScanDuration(long value) {
+    this.value = value;
+  }
+
+  @Override
+  public String name() {
+    return ScanDuration.NAME;
+  }
+
+  @Override
+  public long value() {
+    return value;
+  }
+}

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java

@@ -18,13 +18,16 @@
  */
 package org.apache.iceberg.spark.source;
 
+import static org.apache.iceberg.spark.source.SparkSQLExecutionHelper.lastExecutedMetricValue;
 import static org.assertj.core.api.Assertions.assertThat;
 import static scala.collection.JavaConverters.seqAsJavaListConverter;
 
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.spark.TestBaseWithCatalog;
+import org.apache.iceberg.spark.source.metrics.ScanDuration;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
@@ -63,6 +66,9 @@ public void testReadMetricsForV1Table() throws NoSuchTableException {
     assertThat(metricsMap)
         .hasEntrySatisfying(
             "totalPlanningDuration", sqlMetric -> assertThat(sqlMetric.value()).isNotEqualTo(0));
+    assertThat(metricsMap)
+        .hasEntrySatisfying(
+            "scanDuration", sqlMetric -> assertThat(sqlMetric.value()).isGreaterThan(0));
 
     // data manifests
     assertThat(metricsMap)
@@ -139,6 +145,9 @@ public void testReadMetricsForV2Table() throws NoSuchTableException {
     assertThat(metricsMap)
         .hasEntrySatisfying(
             "totalPlanningDuration", sqlMetric -> assertThat(sqlMetric.value()).isNotEqualTo(0));
+    assertThat(metricsMap)
+        .hasEntrySatisfying(
+            "scanDuration", sqlMetric -> assertThat(sqlMetric.value()).isGreaterThan(0));
 
     // data manifests
     assertThat(metricsMap)
@@ -194,6 +203,59 @@ public void testReadMetricsForV2Table() throws NoSuchTableException {
             "skippedDeleteFiles", sqlMetric -> assertThat(sqlMetric.value()).isEqualTo(0));
   }
 
+  @TestTemplate
+  public void scanDurationAggregatedAcrossTasks() throws NoSuchTableException {
+    sql("CREATE TABLE %s (id BIGINT) USING iceberg", tableName);
+
+    spark.range(10000).coalesce(1).writeTo(tableName).append();
+    spark.range(10001, 20000).coalesce(1).writeTo(tableName).append();
+
+    Dataset<Row> df = spark.sql(String.format("SELECT * FROM %s", tableName));
+    assertThat(df.collectAsList()).hasSize(19999);
+
+    List<SparkPlan> sparkPlans =
+        seqAsJavaListConverter(df.queryExecution().executedPlan().collectLeaves()).asJava();
+    Map<String, SQLMetric> metricsMap =
+        JavaConverters.mapAsJavaMapConverter(sparkPlans.get(0).metrics()).asJava();
+
+    // the driver-side accumulator sums TaskScanDuration across all tasks of the scan
+    long aggregatedNanos = metricsMap.get("scanDuration").value();
+    assertThat(aggregatedNanos).isGreaterThan(0);
+
+    // the value the UI renders, read back from the SQL status store
+    assertThat(lastExecutedMetricValue(spark, new ScanDuration().description()))
+        .containsPattern(
+            "total \\(min, med, max\\)\\R"
+                + "\\d+(\\.\\d+)? (ns|us|ms|s) "
+                + "\\(\\d+(\\.\\d+)? (ns|us|ms|s), "
+                + "\\d+(\\.\\d+)? (ns|us|ms|s), "
+                + "\\d+(\\.\\d+)? (ns|us|ms|s)\\)");
+  }
+
+  @TestTemplate
+  public void scanDurationForNonVectorizedRead() throws NoSuchTableException {
+    // the other tests read through BatchDataReader, disable vectorization to cover RowDataReader
+    sql(
+        "CREATE TABLE %s (id BIGINT) USING iceberg TBLPROPERTIES ('%s'='false')",
+        tableName, TableProperties.PARQUET_VECTORIZATION_ENABLED);
+
+    spark.range(10000).coalesce(1).writeTo(tableName).append();
+
+    Dataset<Row> df = spark.sql(String.format("SELECT * FROM %s", tableName));
+    df.collect();
+
+    List<SparkPlan> sparkPlans =
+        seqAsJavaListConverter(df.queryExecution().executedPlan().collectLeaves()).asJava();
+    // sanity check that this really exercised the row-based path
+    assertThat(sparkPlans.get(0).supportsColumnar()).isFalse();
+
+    Map<String, SQLMetric> metricsMap =
+        JavaConverters.mapAsJavaMapConverter(sparkPlans.get(0).metrics()).asJava();
+    assertThat(metricsMap)
+        .hasEntrySatisfying(
+            "scanDuration", sqlMetric -> assertThat(sqlMetric.value()).isGreaterThan(0));
+  }
+
   @TestTemplate
   public void testDeleteMetrics() throws NoSuchTableException {
     sql(
@@ -222,6 +284,9 @@ public void testDeleteMetrics() throws NoSuchTableException {
     assertThat(metricsMap)
         .hasEntrySatisfying(
             "totalPlanningDuration", sqlMetric -> assertThat(sqlMetric.value()).isNotEqualTo(0));
+    assertThat(metricsMap)
+        .hasEntrySatisfying(
+            "scanDuration", sqlMetric -> assertThat(sqlMetric.value()).isGreaterThan(0));
 
     // data manifests
     assertThat(metricsMap)

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/metrics/TestScanDuration.java

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark.source.metrics;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.Locale;
+import org.junit.jupiter.api.Test;
+
+class TestScanDuration {
+
+  @Test
+  void aggregateTaskMetrics() {
+    ScanDuration metric = new ScanDuration();
+
+    assertThat(metric.aggregateTaskMetrics(new long[] {300L, 400L}))
+        .isEqualTo("total (min, med, max)\n700 ns (300 ns, 400 ns, 400 ns)");
+    assertThat(metric.aggregateTaskMetrics(new long[] {1_500L, 2_500L}))
+        .isEqualTo("total (min, med, max)\n4 us (1 us, 2 us, 2 us)");
+    assertThat(metric.aggregateTaskMetrics(new long[] {1_000_000L, 500_000L}))
+        .isEqualTo("total (min, med, max)\n1 ms (500 us, 1 ms, 1 ms)");
+    assertThat(metric.aggregateTaskMetrics(new long[] {1_500_000_000L, 1_000_000_000L}))
+        .isEqualTo("total (min, med, max)\n2.5 s (1.0 s, 1.5 s, 1.5 s)");
+  }
+
+  @Test
+  void aggregateTaskMetricsWithNoTasks() {
+    assertThat(new ScanDuration().aggregateTaskMetrics(new long[] {})).isEqualTo("0 ns");
+  }
+
+  @Test
+  void aggregateTaskMetricsSurfacesStragglers() {
+    // the total alone would hide that one task took the bulk of the time
+    assertThat(
+            new ScanDuration()
+                .aggregateTaskMetrics(new long[] {1_000_000L, 1_000_000L, 8_000_000L}))
+        .isEqualTo("total (min, med, max)\n10 ms (1 ms, 1 ms, 8 ms)");
+  }
+
+  @Test
+  void aggregateTaskMetricsDoesNotMutateInput() {
+    long[] taskMetrics = new long[] {3_000_000L, 1_000_000L, 2_000_000L};
+    new ScanDuration().aggregateTaskMetrics(taskMetrics);
+    assertThat(taskMetrics).containsExactly(3_000_000L, 1_000_000L, 2_000_000L);
+  }
+
+  @Test
+  void aggregateTaskMetricsIsLocaleIndependent() {
+    Locale previous = Locale.getDefault();
+    try {
+      // a comma-decimal locale would render "2,5 s" without an explicit Locale.ROOT
+      Locale.setDefault(Locale.GERMANY);
+      assertThat(new ScanDuration().aggregateTaskMetrics(new long[] {2_500_000_000L}))
+          .isEqualTo("total (min, med, max)\n2.5 s (2.5 s, 2.5 s, 2.5 s)");
+    } finally {
+      Locale.setDefault(previous);
+    }
+  }
+}