Decode Hive partition values in listing tables (apache#23226)

## Which issue does this PR close?

- Closes apache#19650.

## Rationale for this change

Hive-style partition values can contain percent-encoded characters in
object-store paths, such as `%2F` for `/` or `%20` for a space.
`parse_partitions_for_path` currently returns those encoded bytes
literally, so listing tables expose `foo%2Fbar` instead of `foo/bar`.

## What changes are included in this PR?

- Percent-decode extracted partition values in
`parse_partitions_for_path`.
- Return `Cow<str>` from the parser so unchanged values keep the
borrowed fast path and decoded values can be owned only when needed.
- Fall back to the original raw partition value if percent decoding does
not produce valid UTF-8, rather than dropping the file from listing
results.
- Add helper-level and `PartitionedFile` conversion tests for decoded
partition values.

## Are these changes tested?

- `cargo fmt --all --check`
- `cargo test -p datafusion-catalog-listing`

---------

Signed-off-by: Kevin-Li-2025 <2242139@qq.com>
Co-authored-by: Kevin-Li-2025 <2242139@qq.com>

Author: yinli-systems

Cargo.lock

@@ -185,6 +185,7 @@ parquet = { version = "59.1.0", default-features = false, features = [
 ] }
 pbjson = { version = "0.9.0" }
 pbjson-types = "0.9"
+percent-encoding = "2.3"
 pin-project = "1"
 # Should match arrow-flight's version of prost.
 prost = "0.14.1"

Cargo.toml

@@ -46,6 +46,7 @@ futures = { workspace = true }
 itertools = { workspace = true }
 log = { workspace = true }
 object_store = { workspace = true }
+percent-encoding = { workspace = true }
 
 [dev-dependencies]
 chrono = { workspace = true }

datafusion/catalog-listing/Cargo.toml

@@ -17,6 +17,7 @@
 
 //! Helper functions for the table implementation
 
+use std::borrow::Cow;
 use std::sync::Arc;
 
 use datafusion_catalog::Session;
@@ -43,6 +44,10 @@ use datafusion_expr::{Expr, Volatility};
 use datafusion_physical_expr::create_physical_expr;
 use object_store::path::Path;
 use object_store::{ObjectMeta, ObjectStore};
+use percent_encoding::{AsciiSet, CONTROLS, percent_decode_str, utf8_percent_encode};
+
+const PARTITION_VALUE_ENCODE_SET: &AsciiSet =
+    &CONTROLS.add(b' ').add(b'%').add(b'/').add(b'?').add(b'#');
 
 /// Check whether the given expression can be resolved using only the columns `col_names`.
 /// This means that if this function returns true:
@@ -272,7 +277,16 @@ pub fn evaluate_partition_prefix<'a>(
             Some(PartitionValue::Single(val)) => {
                 // if a partition only has a single literal value, then it can be added to the
                 // prefix
-                parts.push(format!("{p}={val}"));
+                let encoded = encode_partition_value(val);
+                if encoded != val.as_str() {
+                    // The same decoded value can be represented by both raw and
+                    // percent-encoded partition directories. Prefix pruning is
+                    // an optimization, so stop before this partition rather
+                    // than listing only one spelling and potentially skipping
+                    // valid rows.
+                    break;
+                }
+                parts.push(format!("{p}={encoded}"));
             }
             _ => {
                 // break on the first unconstrainted partition to create a common prefix
@@ -289,6 +303,10 @@ pub fn evaluate_partition_prefix<'a>(
     }
 }
 
+fn encode_partition_value(value: &str) -> Cow<'_, str> {
+    utf8_percent_encode(value, PARTITION_VALUE_ENCODE_SET).into()
+}
+
 pub fn filter_partitioned_file(
     pf: PartitionedFile,
     filters: &[Expr],
@@ -343,7 +361,7 @@ fn try_into_partitioned_file(
         .into_iter()
         .zip(partition_cols)
         .map(|(parsed, (_, datatype))| {
-            ScalarValue::try_from_string(parsed.to_string(), datatype)
+            ScalarValue::try_from_string(parsed.into_owned(), datatype)
         })
         .collect::<Result<Vec<_>>>()?;
 
@@ -435,12 +453,15 @@ fn object_meta_to_partitioned_file(
 }
 
 /// Extract the partition values for the given `file_path` (in the given `table_path`)
-/// associated to the partitions defined by `table_partition_cols`
+/// associated to the partitions defined by `table_partition_cols`.
+///
+/// Partition values are percent-decoded to match Hive-style object-store paths
+/// that encode special characters in path segments.
 pub fn parse_partitions_for_path<'a, I>(
     table_path: &ListingTableUrl,
     file_path: &'a Path,
     table_partition_cols: I,
-) -> Option<Vec<&'a str>>
+) -> Option<Vec<Cow<'a, str>>>
 where
     I: IntoIterator<Item = &'a str>,
 {
@@ -449,7 +470,13 @@ where
     let mut part_values = vec![];
     for (part, expected_partition) in subpath.zip(table_partition_cols) {
         match part.split_once('=') {
-            Some((name, val)) if name == expected_partition => part_values.push(val),
+            Some((name, val)) if name == expected_partition => {
+                // Preserve the original value if percent-decoding produces invalid UTF-8.
+                let decoded = percent_decode_str(val)
+                    .decode_utf8()
+                    .unwrap_or(Cow::Borrowed(val));
+                part_values.push(decoded);
+            }
             _ => {
                 debug!(
                     "Ignoring file: file_path='{file_path}', table_path='{table_path}', part='{part}', partition_col='{expected_partition}'",
@@ -525,7 +552,7 @@ mod tests {
     #[test]
     fn test_parse_partitions_for_path() {
         assert_eq!(
-            Some(vec![]),
+            Some(vec![] as Vec<Cow<'_, str>>),
             parse_partitions_for_path(
                 &ListingTableUrl::parse("file:///bucket/mytable").unwrap(),
                 &Path::from("bucket/mytable/file.csv"),
@@ -549,15 +576,51 @@ mod tests {
             )
         );
         assert_eq!(
-            Some(vec!["v1"]),
+            Some(vec![Cow::Borrowed("v1")]),
             parse_partitions_for_path(
                 &ListingTableUrl::parse("file:///bucket/mytable").unwrap(),
                 &Path::from("bucket/mytable/mypartition=v1/file.csv"),
                 vec!["mypartition"]
             )
         );
+        for (path, column, expected) in [
+            (
+                "bucket/mytable/mypartition=v%2F1/file.csv",
+                "mypartition",
+                "v/1",
+            ),
+            (
+                "bucket/mytable/name=John%20Doe/file.csv",
+                "name",
+                "John Doe",
+            ),
+            (
+                "bucket/mytable/mypartition=test%20dir%2Ffile/file.csv",
+                "mypartition",
+                "test dir/file",
+            ),
+            (
+                "bucket/mytable/mypartition=%C3%A9/file.csv",
+                "mypartition",
+                "é",
+            ),
+            (
+                "bucket/mytable/mypartition=%FF/file.csv",
+                "mypartition",
+                "%FF",
+            ),
+        ] {
+            assert_eq!(
+                Some(vec![Cow::Borrowed(expected)]),
+                parse_partitions_for_path(
+                    &ListingTableUrl::parse("file:///bucket/mytable").unwrap(),
+                    &Path::parse(path).unwrap(),
+                    vec![column]
+                )
+            );
+        }
         assert_eq!(
-            Some(vec!["v1"]),
+            Some(vec![Cow::Borrowed("v1")]),
             parse_partitions_for_path(
                 &ListingTableUrl::parse("file:///bucket/mytable/").unwrap(),
                 &Path::from("bucket/mytable/mypartition=v1/file.csv"),
@@ -574,15 +637,15 @@ mod tests {
             )
         );
         assert_eq!(
-            Some(vec!["v1", "v2"]),
+            Some(vec![Cow::Borrowed("v1"), Cow::Borrowed("v2")]),
             parse_partitions_for_path(
                 &ListingTableUrl::parse("file:///bucket/mytable").unwrap(),
                 &Path::from("bucket/mytable/mypartition=v1/otherpartition=v2/file.csv"),
                 vec!["mypartition", "otherpartition"]
             )
         );
         assert_eq!(
-            Some(vec!["v1"]),
+            Some(vec![Cow::Borrowed("v1")]),
             parse_partitions_for_path(
                 &ListingTableUrl::parse("file:///bucket/mytable").unwrap(),
                 &Path::from("bucket/mytable/mypartition=v1/otherpartition=v2/file.csv"),
@@ -614,6 +677,32 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_try_into_partitioned_file_decodes_partition_value() {
+        let table_path = ListingTableUrl::parse("file:///bucket/mytable").unwrap();
+        let partition_cols = vec![("category".to_string(), DataType::Utf8)];
+        let meta = ObjectMeta {
+            location: Path::parse(
+                "bucket/mytable/category=Electronics%2FComputers/data.parquet",
+            )
+            .unwrap(),
+            last_modified: chrono::Utc::now(),
+            size: 100,
+            e_tag: None,
+            version: None,
+        };
+
+        let result =
+            try_into_partitioned_file(meta, &partition_cols, &table_path).unwrap();
+        assert!(result.is_some());
+        let pf = result.unwrap();
+        assert_eq!(pf.partition_values.len(), 1);
+        assert_eq!(
+            pf.partition_values[0],
+            ScalarValue::Utf8(Some("Electronics/Computers".to_string()))
+        );
+    }
+
     #[test]
     fn test_try_into_partitioned_file_root_file_skipped() {
         // File in root directory (not inside any partition path) should be
@@ -768,6 +857,27 @@ mod tests {
             Some(Path::from("a=foo")),
         );
 
+        assert_eq!(
+            evaluate_partition_prefix(
+                partitions,
+                &[col("a").eq(lit("Electronics/Computers"))],
+            ),
+            None,
+        );
+
+        assert_eq!(
+            evaluate_partition_prefix(partitions, &[col("a").eq(lit("John Doe"))]),
+            None,
+        );
+
+        assert_eq!(
+            evaluate_partition_prefix(
+                partitions,
+                &[col("a").eq(lit("foo")).and(col("b").eq(lit("John Doe")))],
+            ),
+            Some(Path::from("a=foo")),
+        );
+
         assert_eq!(
             evaluate_partition_prefix(
                 partitions,