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 cbfabba93b Spark 4.1: Add per-task scanDuration metric (#17562)
cbfabba93b is described below
commit cbfabba93b6c59ba84984269c332a22312435235
Author: Venkata krishnan Sowrirajan <[email protected]>
AuthorDate: Sun Nov 1 23:05:32 2026 -0800
Spark 4.1: Add per-task scanDuration metric (#17562)
* Spark 4.1: Add per-task scanDuration metric
Iceberg reports scan bytes and file/manifest counts per task, but the only
timer in the metrics system is totalPlanningDuration, which measures
driver-side planning. There is no metric for how long a task spent actually
reading data, so a scan's throughput cannot be derived from Iceberg metrics
alone.
Add scanDuration: wall time a task spends in BaseReader.next(), accumulated
per call rather than per row so the timer costs nothing measurable relative
to the read. Emitted as TaskScanDuration from BatchDataReader and
RowDataReader, aggregated by ScanDuration, mirroring the existing
TotalPlanningDuration pair.
Values are nanoseconds: a single split can be read in well under a
millisecond, so a millisecond timer would round most per-task reads to zero.
ScanDuration formats the aggregate as ns/us/ms/s for the UI.
Generated-by: Claude Opus 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.1/spark/src/main/java/org/apache/iceberg/spark/source/BaseReader.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/BaseReader.java
index a4d9766ae7..50dc69dd9a 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/BaseReader.java
+++
b/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 @@ 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.1/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java
index 237dfd5c69..d6ef3bb095 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java
+++
b/spark/v4.1/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.1/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java
index dbfb0b7614..a60fbabcaf 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/RowDataReader.java
+++
b/spark/v4.1/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.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
index 22340408b0..c54e9c2dab 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java
+++
b/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.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.1/spark/src/main/java/org/apache/iceberg/spark/source/metrics/ScanDuration.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/metrics/ScanDuration.java
new file mode 100644
index 0000000000..49f55ab355
--- /dev/null
+++
b/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));
+ }
+ }
+}
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/metrics/TaskScanDuration.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/metrics/TaskScanDuration.java
new file mode 100644
index 0000000000..a505910514
--- /dev/null
+++
b/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;
+ }
+}
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java
index 57b2c7146b..442c96ae6c 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReadMetrics.java
+++
b/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 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.1/spark/src/test/java/org/apache/iceberg/spark/source/metrics/TestScanDuration.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/metrics/TestScanDuration.java
new file mode 100644
index 0000000000..9a1f653023
--- /dev/null
+++
b/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);
+ }
+ }
+}
