fix: ignore filenames when parsing Hive partitions

datafusion/catalog-listing/src/helpers.rs

@@ -467,6 +467,15 @@ fn object_meta_to_partitioned_file(
 ///
 /// Partition values are percent-decoded to match Hive-style object-store paths
 /// that encode special characters in path segments.
+///
+/// The final path component is always treated as a filename, even if it contains
+/// `=`. Each requested partition column must match a `name=value` directory in
+/// order; this returns `None` if any of those directories is missing or mismatched,
+/// or if `file_path` is outside `table_path`.
+///
+/// For example, columns `year` and `month` match `year=2024/month=01/data.parquet`,
+/// but not `year=2024/month=01.parquet`: the latter is missing the month directory.
+/// A `key=value` filename inside a complete partition path remains valid.
 pub fn parse_partitions_for_path<'a, I>(
     table_path: &ListingTableUrl,
     file_path: &'a Path,
@@ -475,10 +484,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.
@@ -594,6 +611,30 @@ mod tests {
                 vec!["mypartition"]
             )
         );
+        assert_eq!(
+            None,
+            parse_partitions_for_path(
+                &ListingTableUrl::parse("file:///bucket/mytable").unwrap(),
+                &Path::from("bucket/mytable/mypartition=v1.csv"),
+                vec!["mypartition"]
+            )
+        );
+        assert_eq!(
+            None,
+            parse_partitions_for_path(
+                &ListingTableUrl::parse("file:///bucket/mytable").unwrap(),
+                &Path::from("bucket/mytable/mypartition=v1/otherpartition=v2.csv"),
+                vec!["mypartition", "otherpartition"]
+            )
+        );
+        assert_eq!(
+            Some(vec![Cow::Borrowed("v1")]),
+            parse_partitions_for_path(
+                &ListingTableUrl::parse("file:///bucket/mytable").unwrap(),
+                &Path::from("bucket/mytable/mypartition=v1/mypartition=ignored.csv"),
+                vec!["mypartition"]
+            )
+        );
         for (path, column, expected) in [
             (
                 "bucket/mytable/mypartition=v%2F1/file.csv",
@@ -875,28 +916,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,81 @@ 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 root filename cannot supply an integer partition value. Previously this
+# attempted to cast "2025.parquet" to INT and failed the query; it is now skipped.
+query I
+copy (values(7))
+to 'test_files/scratch/listing_table_partitions/filename_partition_int/year=2024/data.parquet';
+----
+1
+
+statement ok
+create external table filename_partition_int (column1 bigint)
+partitioned by (year int)
+stored as parquet
+location 'test_files/scratch/listing_table_partitions/filename_partition_int/';
+
+# A new stray file arrives after the table's valid layout has been registered.
+query I
+copy (values(8))
+to 'test_files/scratch/listing_table_partitions/filename_partition_int/year=2025.parquet';
+----
+1
+
+statement ok
+set datafusion.runtime.list_files_cache_limit = '0K';
+
+query II
+select column1, year from filename_partition_int order by column1;
+----
+7 2024
+
+# 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;