This is an automated email from the ASF dual-hosted git repository.
nastra pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 65b3ef316a Spark 4.2: Add scanDuration read metric (#18751)
65b3ef316a is described below
commit 65b3ef316ae0c6fb13a9150b5eaac1258f8cda47
Author: Venkata krishnan Sowrirajan <[email protected]>
AuthorDate: Mon Nov 2 05:57:03 2026 -0800
Spark 4.2: Add scanDuration read metric (#18751)
Port of #17562 to Spark 4.2. Reports how long each task spends reading its
splits as the scanDuration custom metric, in nanoseconds.
Generated-by: Claude Opus 5.5
Co-authored-by: Venkata Krishnan Sowrirajan <[email protected]>
---
.../apache/iceberg/spark/source/BaseReader.java | 10 +++
.../iceberg/spark/source/BatchDataReader.java | 5 +-
.../apache/iceberg/spark/source/RowDataReader.java | 5 +-
.../org/apache/iceberg/spark/source/SparkScan.java | 2 +
.../iceberg/spark/source/metrics/ScanDuration.java | 78 ++++++++++++++++++++++
.../spark/source/metrics/TaskScanDuration.java | 40 +++++++++++
.../iceberg/spark/source/TestSparkReadMetrics.java | 65 ++++++++++++++++++
.../spark/source/metrics/TestScanDuration.java | 75 +++++++++++++++++++++
8 files changed, 278 insertions(+), 2 deletions(-)
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseReader.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseReader.java
index a4d9766ae7..50dc69dd9a 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BaseReader.java
+++
b/spark/v4.2/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 @@ abstract class BaseReader<T, TaskT extends ScanTask>
implements Closeable {
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 @@ abstract class BaseReader<T, TaskT extends ScanTask>
implements Closeable {
LOG.error("Error reading file(s): {}", filePaths, e);
}
throw e;
+ } finally {
+ this.scanDurationNanos += System.nanoTime() - start;
}
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java
index 237dfd5c69..d6ef3bb095 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java
@@ -33,6 +33,7 @@ import org.apache.iceberg.spark.OrcBatchReadConf;
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())
};
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java
index dbfb0b7614..a60fbabcaf 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java
@@ -33,6 +33,7 @@ import org.apache.iceberg.io.InputFile;
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())
};
}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
index 22340408b0..c54e9c2dab 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
@@ -49,6 +49,7 @@ import org.apache.iceberg.spark.source.metrics.NumSplits;
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 @@ abstract class SparkScan implements Scan,
SupportsReportStatistics {
// common
new TotalPlanningDuration(),
+ new ScanDuration(),
// data manifests
new TotalDataManifests(),
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/metrics/ScanDuration.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/metrics/ScanDuration.java
new file mode 100644
index 0000000000..49f55ab355
--- /dev/null
+++
b/spark/v4.2/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));
+ }
+ }
+}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/metrics/TaskScanDuration.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/metrics/TaskScanDuration.java
new file mode 100644
index 0000000000..a505910514
--- /dev/null
+++
b/spark/v4.2/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;
+ }
+}
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java
index 57b2c7146b..442c96ae6c 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java
+++
b/spark/v4.2/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 class TestSparkReadMetrics extends TestBaseWithCatalog
{
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 class TestSparkReadMetrics extends
TestBaseWithCatalog {
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 class TestSparkReadMetrics extends
TestBaseWithCatalog {
"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 class TestSparkReadMetrics extends
TestBaseWithCatalog {
assertThat(metricsMap)
.hasEntrySatisfying(
"totalPlanningDuration", sqlMetric ->
assertThat(sqlMetric.value()).isNotEqualTo(0));
+ assertThat(metricsMap)
+ .hasEntrySatisfying(
+ "scanDuration", sqlMetric ->
assertThat(sqlMetric.value()).isGreaterThan(0));
// data manifests
assertThat(metricsMap)
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/metrics/TestScanDuration.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/metrics/TestScanDuration.java
new file mode 100644
index 0000000000..9a1f653023
--- /dev/null
+++
b/spark/v4.2/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);
+ }
+ }
+}
