fix: ignore filenames when parsing Hive partitions

datafusion/catalog-listing/src/helpers.rs

@@ -475,10 +475,18 @@ pub fn parse_partitions_for_path<'a, I>(
 where
     I: IntoIterator<Item = &'a str>,
 {
-    let subpath = table_path.strip_prefix(file_path)?;
+    let mut subpath = table_path.strip_prefix(file_path)?.peekable();
 
     let mut part_values = vec![];
-    for (part, expected_partition) in subpath.zip(table_partition_cols) {
+    for expected_partition in table_partition_cols {
+        let part = subpath.next()?;
+        // A partition value must come from a directory, never the filename.
+        if subpath.peek().is_none() {
+            debug!(
+                "Ignoring file: file_path='{file_path}', table_path='{table_path}', missing partition directory '{expected_partition}'",
+            );
+            return None;
+        }
         match part.split_once('=') {
             Some((name, val)) if name == expected_partition => {
                 // Preserve the original value if percent-decoding produces invalid UTF-8.
@@ -875,28 +883,27 @@ mod tests {
             ("year".to_string(), DataType::Utf8),
             ("month".to_string(), DataType::Utf8),
         ];
-        let meta = ObjectMeta {
-            location: Path::from("bucket/mytable/year=2024/data.parquet"),
-            last_modified: chrono::Utc::now(),
-            size: 100,
-            e_tag: None,
-            version: None,
-        };
+        for filename in ["data.parquet", "month=02.parquet"] {
+            let meta = ObjectMeta {
+                location: Path::from(format!("bucket/mytable/year=2024/{filename}")),
+                last_modified: chrono::Utc::now(),
+                size: 100,
+                e_tag: None,
+                version: None,
+            };
 
-        let result = try_into_partitioned_file(
-            meta,
-            &partition_cols,
-            &table_path,
-            HIVE_DEFAULT_PARTITION_NAME,
-        )
-        .unwrap();
-        // File has year=2024 but no month= directory — parse_partitions_for_path
-        // returns None because the path component "data.parquet" doesn't match
-        // the expected "month=..." pattern.
-        assert!(
-            result.is_none(),
-            "Files with incomplete partition structure should be skipped"
-        );
+            let result = try_into_partitioned_file(
+                meta,
+                &partition_cols,
+                &table_path,
+                HIVE_DEFAULT_PARTITION_NAME,
+            )
+            .unwrap();
+            assert!(
+                result.is_none(),
+                "A filename must not supply the missing month partition: {filename}"
+            );
+        }
     }
 
     #[test]

datafusion/sqllogictest/test_files/listing_table_partitions.slt

@@ -257,3 +257,52 @@ select a, p from null_part_write order by a;
 
 statement ok
 set datafusion.execution.listing_table_hive_default_partition_name = '__HIVE_DEFAULT_PARTITION__';
+
+# A filename shaped like a partition must not supply a missing directory level.
+query I
+copy (values(3))
+to 'test_files/scratch/listing_table_partitions/filename_partition/year=2024/month=01/a.parquet';
+----
+1
+
+statement ok
+create external table filename_partition (column1 bigint)
+partitioned by (year varchar, month varchar)
+stored as parquet
+location 'test_files/scratch/listing_table_partitions/filename_partition/';
+
+query I
+copy (values(4))
+to 'test_files/scratch/listing_table_partitions/filename_partition/year=2024/month=02.parquet';
+----
+1
+
+query I
+copy (values(5))
+to 'test_files/scratch/listing_table_partitions/filename_partition/year=2024/data.parquet';
+----
+1
+
+# A key=value filename inside a complete partition path remains a valid file.
+query I
+copy (values(6))
+to 'test_files/scratch/listing_table_partitions/filename_partition/year=2024/month=01/month=03.parquet';
+----
+1
+
+statement ok
+set datafusion.runtime.list_files_cache_limit = '0K';
+
+query ITT
+select column1, year, month from filename_partition order by column1;
+----
+3 2024 01
+6 2024 01
+
+query I
+select count(*) from filename_partition where month = '01';
+----
+2
+
+statement ok
+RESET datafusion.catalog.create_default_catalog_and_schema;