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 a4d9766ae713..50dc69dd9a55 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 implements Closeable { private final Iterator 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 lazyInputFiles; private CloseableIterator 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; } } 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 237dfd5c6919..d6ef3bb095ac 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.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 @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 dbfb0b7614e4..a60fbabcaf38 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.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 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 22340408b08f..c54e9c2dab59 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.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(), 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 000000000000..49f55ab35532 --- /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 000000000000..a505910514ad --- /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 57b2c7146b79..442c96ae6cd0 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 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 df = spark.sql(String.format("SELECT * FROM %s", tableName)); + assertThat(df.collectAsList()).hasSize(19999); + + List sparkPlans = + seqAsJavaListConverter(df.queryExecution().executedPlan().collectLeaves()).asJava(); + Map 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 df = spark.sql(String.format("SELECT * FROM %s", tableName)); + df.collect(); + + List sparkPlans = + seqAsJavaListConverter(df.queryExecution().executedPlan().collectLeaves()).asJava(); + // sanity check that this really exercised the row-based path + assertThat(sparkPlans.get(0).supportsColumnar()).isFalse(); + + Map 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) 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 000000000000..9a1f653023a4 --- /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); + } + } +}