Add files_processed and files_scanned metrics to FileStreamMetrics

datafusion/datasource/src/file_stream.rs

@@ -124,6 +124,7 @@ impl FileStream {
                 }
                 FileStreamState::Open { future } => match ready!(future.poll_unpin(cx)) {
                     Ok(reader) => {
+                        self.file_stream_metrics.files_processed.add(1);
                         // include time needed to start opening in `start_next_file`
                         self.file_stream_metrics.time_opening.stop();
                         let next = self.start_next_file().transpose();
@@ -179,6 +180,13 @@ impl FileStream {
                                         batch
                                     } else {
                                         let batch = batch.slice(0, *remain);
+                                        // Count the prefetched next file (if any) and
+                                        // all remaining files we will never open.
+                                        let unprocessed = self.file_iter.len()
+                                            + usize::from(next.is_some());
+                                        self.file_stream_metrics
+                                            .files_processed
+                                            .add(unprocessed);
                                         self.state = FileStreamState::Limit;
                                         *remain = 0;
                                         batch
@@ -223,6 +231,7 @@ impl FileStream {
                             }
                         }
                         None => {
+                            self.file_stream_metrics.files_scanned.add(1);
                             self.file_stream_metrics.time_scanning_until_data.stop();
                             self.file_stream_metrics.time_scanning_total.stop();
 
@@ -399,6 +408,12 @@ pub struct FileStreamMetrics {
     /// If using `OnError::Skip` this will provide a count of the number of files
     /// which were skipped and will not be included in the scan results.
     pub file_scan_errors: Count,
+    /// Count of files processed (opened, pruned, or skipped due to limit).
+    /// When the stream completes, this equals the total number of files
+    /// assigned to this partition.
+    pub files_processed: Count,
+    /// Count of files completely scanned (reader stream fully consumed).
+    pub files_scanned: Count,
 }
 
 impl FileStreamMetrics {
@@ -433,13 +448,21 @@ impl FileStreamMetrics {
         let file_scan_errors =
             MetricBuilder::new(metrics).counter("file_scan_errors", partition);
 
+        let files_processed =
+            MetricBuilder::new(metrics).counter("files_processed", partition);
+
+        let files_scanned =
+            MetricBuilder::new(metrics).counter("files_scanned", partition);
+
         Self {
             time_opening,
             time_scanning_until_data,
             time_scanning_total,
             time_processing,
             file_open_errors,
             file_scan_errors,
+            files_processed,
+            files_scanned,
         }
     }
 }