Spark: Add custom metric for number of file splits read by a SparkScan

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/NumFiles.java

@@ -0,0 +1,45 @@
+/*
+ * 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;
+
+import java.text.NumberFormat;
+import org.apache.spark.sql.connector.metric.CustomMetric;
+
+public class NumFiles implements CustomMetric {
+
+  @Override
+  public String name() {
+    return "numFiles";
+  }
+
+  @Override
+  public String description() {
+    return "number of files read";
+  }
+
+  @Override
+  public String aggregateTaskMetrics(long[] taskMetrics) {
+    long sum = initialValue;
+    for (int i = 0; i < taskMetrics.length; i++) {
+      sum += taskMetrics[i];
+    }
+    return NumberFormat.getIntegerInstance().format(sum);
+  }
+}

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchQueryScan.java

@@ -93,6 +93,11 @@ class SparkBatchQueryScan extends SparkScan implements SupportsRuntimeFiltering
     }
   }
 
+  @Override
+  public String description() {
+    return "IcebergScan " + super.description();
+  }
+
   Long snapshotId() {
     return snapshotId;
   }

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java

@@ -77,6 +77,11 @@ class SparkCopyOnWriteScan extends SparkScan implements SupportsRuntimeFiltering
     }
   }
 
+  @Override
+  public String description() {
+    return "IcebergCopyOnWriteScan " + super.description();
+  }
+
   Long snapshotId() {
     return snapshot != null ? snapshot.snapshotId() : null;
   }

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkFilesScan.java

@@ -50,6 +50,11 @@ class SparkFilesScan extends SparkScan {
     this.splitOpenFileCost = readConf.splitOpenFileCost();
   }
 
+  @Override
+  public String description() {
+    return "IcebergFilesScan " + super.description();
+  }
+
   @Override
   protected List<CombinedScanTask> tasks() {
     if (tasks == null) {

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkScan.java

@@ -44,6 +44,8 @@
 import org.apache.spark.broadcast.Broadcast;
 import org.apache.spark.sql.SparkSession;
 import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.connector.metric.CustomMetric;
+import org.apache.spark.sql.connector.metric.CustomTaskMetric;
 import org.apache.spark.sql.connector.read.Batch;
 import org.apache.spark.sql.connector.read.InputPartition;
 import org.apache.spark.sql.connector.read.PartitionReader;
@@ -162,6 +164,11 @@ public String description() {
     return String.format("%s [filters=%s]", table, filters);
   }
 
+  @Override
+  public CustomMetric[] supportedCustomMetrics() {
+    return new CustomMetric[] { new NumFiles() };
+  }
+
   static class ReaderFactory implements PartitionReaderFactory {
     private final int batchSize;
 
@@ -193,15 +200,41 @@ public boolean supportColumnarReads(InputPartition partition) {
     }
   }
 
+  static long numFilesToScan(CombinedScanTask scanTask) {
+    long fileCount = 0L;
+    for (FileScanTask file : scanTask.files()) {
+      fileCount += 1L;
+    }
+    return fileCount;
+  }
+
   private static class RowReader extends RowDataReader implements PartitionReader<InternalRow> {
+    private long numFilesToRead;
+
     RowReader(ReadTask task) {
       super(task.task, task.table(), task.expectedSchema(), task.isCaseSensitive());
+      numFilesToRead = numFilesToScan(task.task);
+      LOG.debug("reading {} files for table {}", numFilesToRead, task.table().name());
+    }
+
+    @Override
+    public CustomTaskMetric[] currentMetricsValues() {
+      return new CustomTaskMetric[] { new TaskNumFiles(numFilesToRead) };
     }
   }
 
   private static class BatchReader extends BatchDataReader implements PartitionReader<ColumnarBatch> {
+    private long numFilesToRead;
+
     BatchReader(ReadTask task, int batchSize) {
       super(task.task, task.table(), task.expectedSchema(), task.isCaseSensitive(), batchSize);
+      numFilesToRead = numFilesToScan(task.task);
+      LOG.debug("reading {} files for table {}", numFilesToRead, task.table().name());
+    }
+
+    @Override
+    public CustomTaskMetric[] currentMetricsValues() {
+      return new CustomTaskMetric[] { new TaskNumFiles(numFilesToRead) };
     }
   }
 

spark/v3.2/spark/src/main/java/org/apache/iceberg/spark/source/TaskNumFiles.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;
+
+import org.apache.spark.sql.connector.metric.CustomTaskMetric;
+
+public class TaskNumFiles implements CustomTaskMetric {
+  private long value;
+
+  TaskNumFiles(long value) {
+    this.value = value;
+  }
+
+  @Override
+  public String name() {
+    return "numFiles";
+  }
+
+  @Override
+  public long value() {
+    return value;
+  }
+}