@@ -61,11 +61,16 @@ pub struct ParquetFileMetrics {
     /// the initial pruning but were proved unreachable mid-scan after the
     /// dynamic filter tightened.
     pub row_groups_pruned_dynamic_filter: Count,
-    /// Number of row groups for which the per-row
-    /// [`RowFilter`](parquet::arrow::arrow_reader::RowFilter) was skipped
-    /// because the static stats proved every row of the RG satisfies the
-    /// predicate. The decoder is rebuilt at the boundary with an empty
-    /// row filter so the upcoming RG decodes without per-row evaluation.
+    /// Number of times the per-row
+    /// [`RowFilter`](parquet::arrow::arrow_reader::RowFilter) was toggled
+    /// off at a row-group boundary because static stats proved every row
+    /// of the upcoming row group(s) satisfies the predicate. The decoder
+    /// is rebuilt at that boundary with an empty row filter so the
+    /// fully-matched run decodes without per-row evaluation.
+    ///
+    /// Note this counts *suppression events*, not row groups: a run of
+    /// consecutive fully-matched row groups shares a single toggle (the
+    /// filter stays off across the run with no further rebuilds).
     pub row_filter_skipped_fully_matched: Count,
     /// Total number of bytes scanned
     pub bytes_scanned: Count,
 
@@ -40,6 +40,7 @@ use arrow::datatypes::DataType;
 use datafusion_datasource::morsel::{Morsel, MorselPlan, MorselPlanner, Morselizer};
 use datafusion_physical_expr::projection::ProjectionExprs;
 use datafusion_physical_expr_adapter::replace_columns_with_literals;
+use datafusion_physical_expr_adapter::rewrite::rewrite_input_file_name_in_projection;
 use std::collections::{HashMap, VecDeque};
 use std::fmt;
 use std::future::Future;
@@ -795,6 +796,9 @@ impl ParquetMorselizer {
                 .transpose()?;
         }
 
+        // Replace any `input_file_name()` UDFs in the projection with a literal for this file.
+        projection = rewrite_input_file_name_in_projection(projection, &file_name)?;
+
         let predicate_creation_errors = MetricBuilder::new(&self.metrics)
             .with_category(MetricCategory::Rows)
             .global_counter("num_predicate_creation_errors");
@@ -3336,8 +3340,12 @@ mod test {
     /// (e.g. `row_number`) plumbed through `TableSchema`/`ParquetOpener`.
     mod virtual_columns {
         use super::*;
-        use arrow::array::{Array, Int64Array};
+        use arrow::array::{Array, Int64Array, StringArray};
         use arrow::datatypes::FieldRef;
+        use datafusion_common::config::ConfigOptions;
+        use datafusion_expr::ScalarUDF;
+        use datafusion_functions::core::input_file_name::InputFileNameFunc;
+        use datafusion_physical_expr::{ScalarFunctionExpr, projection::ProjectionExpr};
         use parquet::arrow::RowNumber;
 
         /// Build a parquet `row_number` virtual column field. Spark's
@@ -3351,6 +3359,54 @@ mod test {
             )
         }
 
+        fn input_file_name_expr() -> Arc<dyn PhysicalExpr> {
+            Arc::new(ScalarFunctionExpr::new(
+                "input_file_name",
+                Arc::new(ScalarUDF::from(InputFileNameFunc::new())),
+                vec![],
+                Arc::new(Field::new("input_file_name", DataType::Utf8, true)),
+                Arc::new(ConfigOptions::default()),
+            ))
+        }
+
+        #[tokio::test]
+        async fn test_input_file_name_projection() {
+            let store = Arc::new(InMemory::new()) as Arc<dyn ObjectStore>;
+            let path = "dir/input_file_name.parquet";
+            let (file_schema, data_size) = write_grouped_file(&store, path, 1, 3).await;
+
+            let projection = ProjectionExprs::new([
+                ProjectionExpr::new(Arc::new(Column::new("value", 0)), "value"),
+                ProjectionExpr::new(input_file_name_expr(), "file_name"),
+            ]);
+
+            let morselizer = ParquetMorselizerBuilder::new()
+                .with_store(Arc::clone(&store))
+                .with_schema(file_schema)
+                .with_projection(projection)
+                .build();
+
+            let file =
+                PartitionedFile::new(path.to_string(), u64::try_from(data_size).unwrap());
+            let mut stream = open_file(&morselizer, file).await.unwrap();
+            let batch = stream.next().await.unwrap().unwrap();
+            assert!(stream.next().await.is_none());
+
+            assert_eq!(batch.num_columns(), 2);
+            assert_eq!(batch.schema().field(0).name(), "value");
+            assert_eq!(batch.schema().field(1).name(), "file_name");
+
+            let file_names = batch
+                .column(1)
+                .as_any()
+                .downcast_ref::<StringArray>()
+                .expect("file_name column should be Utf8");
+            assert_eq!(file_names.len(), 3);
+            for i in 0..file_names.len() {
+                assert_eq!(file_names.value(i), path);
+            }
+        }
+
         /// Collect every `Int64` value from the given column in every batch
         /// of a stream. Used to verify the `row_number` column end to end.
         async fn collect_int64_values(