@@ -29,7 +29,7 @@ use datafusion_datasource::{FileExtensions, ListingTableUrl};
 use datafusion_expr::{BinaryExpr, Operator, lit, utils};
 
 use arrow::{
-    array::AsArray,
+    array::{Array, AsArray},
     datatypes::{DataType, Field},
     record_batch::RecordBatch,
 };
@@ -45,6 +45,8 @@ use datafusion_physical_expr::create_physical_expr;
 use object_store::path::Path;
 use object_store::{ObjectMeta, ObjectStore};
 
+const HIVE_DEFAULT_PARTITION_NAME: &str = "__HIVE_DEFAULT_PARTITION__";
+
 /// Check whether the given expression can be resolved using only the columns `col_names`.
 /// This means that if this function returns true:
 /// - the table provider can filter the table partition values with this expression
@@ -349,19 +351,22 @@ fn filter_partitions(
     let expr = create_physical_expr(&filter, df_schema, &props)?;
 
     // Since we're only operating on a single file, our batch and resulting "array" holds only one
-    // value indicating if the input file matches the provided filters
+    // value indicating if the input file matches the provided filters. A NULL is not a match:
+    // a null partition value makes that reachable, as `part = 1` is NULL for a null partition.
     let matches = expr.evaluate(&batch)?.into_array(1)?;
-    if matches.as_boolean().value(0) {
+    let matches = matches.as_boolean();
+    if matches.is_valid(0) && matches.value(0) {
         return Ok(Some(pf));
     }
 
     Ok(None)
 }
 
 /// Returns `Ok(None)` when the file is not inside a valid partition path
-/// (e.g. a stale file in the table root directory). Such files are skipped
-/// because hive-style partition values are never null and there is no valid
-/// value to assign for non-partitioned files.
+/// (e.g. a stale file in the table root directory), as there is no value to
+/// assign for non-partitioned files.
+///
+/// A partition value named `__HIVE_DEFAULT_PARTITION__` is read as NULL.
 fn try_into_partitioned_file(
     object_meta: ObjectMeta,
     partition_cols: &[(String, DataType)],
@@ -379,6 +384,9 @@ fn try_into_partitioned_file(
         .into_iter()
         .zip(partition_cols)
         .map(|(parsed, (_, datatype))| {
+            if parsed == HIVE_DEFAULT_PARTITION_NAME {
+                return ScalarValue::try_new_null(datatype);
+            }
             ScalarValue::try_from_string(parsed.to_string(), datatype)
         })
         .collect::<Result<Vec<_>>>()?;
@@ -439,7 +447,7 @@ pub async fn pruned_partition_list<'a>(
         )?;
 
         Ok(objects
-            .try_filter_map(|object_meta| {
+            .try_filter_map(move |object_meta| {
                 futures::future::ready(try_into_partitioned_file(
                     object_meta,
                     partition_cols,
@@ -649,6 +657,50 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_try_into_partitioned_file_null_partition() {
+        let table_path = ListingTableUrl::parse("file:///bucket/mytable").unwrap();
+        // A type the placeholder cannot be parsed as, which is how this used to fail.
+        let partition_cols = vec![("day".to_string(), DataType::Date32)];
+        let meta = ObjectMeta {
+            location: Path::from(
+                "bucket/mytable/day=__HIVE_DEFAULT_PARTITION__/data.parquet",
+            ),
+            last_modified: chrono::Utc::now(),
+            size: 100,
+            e_tag: None,
+            version: None,
+        };
+
+        let pf = try_into_partitioned_file(meta, &partition_cols, &table_path)
+            .unwrap()
+            .expect("file should be listed");
+
+        assert_eq!(pf.partition_values, vec![ScalarValue::Date32(None)]);
+    }
+
+    #[test]
+    fn test_try_into_partitioned_file_other_partition_name_is_literal() {
+        let table_path = ListingTableUrl::parse("file:///bucket/mytable").unwrap();
+        let partition_cols = vec![("category".to_string(), DataType::Utf8)];
+        let meta = ObjectMeta {
+            location: Path::from("bucket/mytable/category=MY_NULL/data.parquet"),
+            last_modified: chrono::Utc::now(),
+            size: 100,
+            e_tag: None,
+            version: None,
+        };
+
+        let pf = try_into_partitioned_file(meta, &partition_cols, &table_path)
+            .unwrap()
+            .expect("file should be listed");
+
+        assert_eq!(
+            pf.partition_values,
+            vec![ScalarValue::Utf8(Some("MY_NULL".to_string()))]
+        );
+    }
+
     #[test]
     fn test_try_into_partitioned_file_root_file_skipped() {
         // File in root directory (not inside any partition path) should be
 
@@ -211,10 +211,11 @@ impl ListingTable {
             .options
             .ok_or_else(|| internal_datafusion_err!("No ListingOptions provided"))?;
 
-        // Add the partition columns to the file schema
+        // Add the partition columns to the file schema. They are nullable because a Hive
+        // compliant writer uses `__HIVE_DEFAULT_PARTITION__` for a null partition value.
         let mut builder = SchemaBuilder::from(file_schema.as_ref().to_owned());
         for (part_col_name, part_col_type) in &options.table_partition_cols {
-            builder.push(Field::new(part_col_name, part_col_type.clone(), false));
+            builder.push(Field::new(part_col_name, part_col_type.clone(), true));
         }
 
         let table_schema = Arc::new(
@@ -341,7 +342,8 @@ impl ListingTable {
             self.options
                 .table_partition_cols
                 .iter()
-                .map(|(col, field)| Arc::new(Field::new(col, field.clone(), false)))
+                // Nullable to match the table schema built in `try_new`.
+                .map(|(col, field)| Arc::new(Field::new(col, field.clone(), true)))
                 .collect(),
         );
 
@@ -792,7 +794,7 @@ impl ListingTable {
 
         // Note: Statistics already include both file columns and partition columns.
         // PartitionedFile::with_statistics automatically appends exact partition column
-        // statistics (min=max=partition_value, null_count=0, distinct_count=1) computed
+        // statistics (e.g. min=max=partition_value, null_count=0, distinct_count=1) computed
         // from partition_values.
         Ok(ListFilesResult {
             file_groups,
 
@@ -233,6 +233,111 @@ async fn test_list_partition() {
     );
 }
 
+/// A partition directory named `__HIVE_DEFAULT_PARTITION__` holds the rows whose partition
+/// value is null, so it is read as NULL rather than parsed as the column's type.
+#[tokio::test]
+async fn test_pruned_partition_list_null_partition() {
+    let (store, state) = make_test_store_and_state(&[
+        ("tablepath/mypartition=3/file.parquet", 100),
+        (
+            "tablepath/mypartition=__HIVE_DEFAULT_PARTITION__/file.parquet",
+            100,
+        ),
+    ]);
+
+    let pruned = pruned_partition_list(
+        state.as_ref(),
+        store.as_ref(),
+        &ListingTableUrl::parse("file:///tablepath/").unwrap(),
+        &[],
+        ".parquet",
+        &[(String::from("mypartition"), DataType::Int32)],
+    )
+    .await
+    .expect("partition pruning failed")
+    .try_collect::<Vec<_>>()
+    .await
+    .unwrap();
+
+    assert_eq!(pruned.len(), 2);
+    assert_eq!(&pruned[0].partition_values, &[ScalarValue::Int32(Some(3))]);
+    assert_eq!(&pruned[1].partition_values, &[ScalarValue::Int32(None)]);
+}
+
+#[tokio::test]
+async fn test_pruned_partition_list_prunes_null_partition() {
+    let files: &[(&str, u64)] = &[
+        ("tablepath/mypartition=3/file.parquet", 100),
+        (
+            "tablepath/mypartition=__HIVE_DEFAULT_PARTITION__/file.parquet",
+            100,
+        ),
+    ];
+    let partition_cols = &[(String::from("mypartition"), DataType::Int32)];
+
+    let pruned_with = |filter: Expr| async move {
+        let (store, state) = make_test_store_and_state(files);
+        pruned_partition_list(
+            state.as_ref(),
+            store.as_ref(),
+            &ListingTableUrl::parse("file:///tablepath/").unwrap(),
+            &[filter],
+            ".parquet",
+            partition_cols,
+        )
+        .await
+        .expect("partition pruning failed")
+        .try_collect::<Vec<_>>()
+        .await
+        .unwrap()
+    };
+
+    // The null partition does not match an equality filter: `mypartition = 3` is NULL there.
+    let pruned = pruned_with(Expr::eq(col("mypartition"), lit(3i32))).await;
+    assert_eq!(pruned.len(), 1);
+    assert_eq!(
+        pruned[0].object_meta.location.as_ref(),
+        "tablepath/mypartition=3/file.parquet"
+    );
+
+    // ... and only the null partition matches `IS NULL`.
+    let pruned = pruned_with(col("mypartition").is_null()).await;
+    assert_eq!(pruned.len(), 1);
+    assert_eq!(
+        pruned[0].object_meta.location.as_ref(),
+        "tablepath/mypartition=__HIVE_DEFAULT_PARTITION__/file.parquet"
+    );
+}
+
+#[tokio::test]
+async fn test_pruned_partition_list_hive_null_partition_name() {
+    let (store, state) = make_test_store_and_state(&[
+        ("tablepath/mypartition=MY_NULL/file.parquet", 100),
+        (
+            "tablepath/mypartition=__HIVE_DEFAULT_PARTITION__/file.parquet",
+            100,
+        ),
+    ]);
+    let pruned = pruned_partition_list(
+        state.as_ref(),
+        store.as_ref(),
+        &ListingTableUrl::parse("file:///tablepath/").unwrap(),
+        &[],
+        ".parquet",
+        &[(String::from("mypartition"), DataType::Utf8)],
+    )
+    .await
+    .expect("partition pruning failed")
+    .try_collect::<Vec<_>>()
+    .await
+    .unwrap();
+
+    // Only the fixed Hive marker is read as NULL; other names remain literal values.
+    assert_eq!(pruned.len(), 2);
+    assert_eq!(&pruned[0].partition_values, &[ScalarValue::from("MY_NULL")]);
+    assert_eq!(&pruned[1].partition_values, &[ScalarValue::Utf8(None)]);
+}
+
 pub fn make_test_store_and_state(
     files: &[(&str, u64)],
 ) -> (Arc<InMemory>, Arc<dyn Session>) {
 
@@ -371,9 +371,11 @@ impl FileGroupPartitioner {
 /// in the group for the **full table schema** (file columns + partition columns).
 ///
 /// Partition column statistics are derived from the individual file partition values:
-/// - `min` = minimum partition value across all files in the group
-/// - `max` = maximum partition value across all files in the group
-/// - `null_count` = 0 (partition values are never null)
+/// - `min` = minimum partition value across all files in the group, or absent if any
+///   file in the group has a null partition value (see #26748)
+/// - `max` = maximum partition value across all files in the group, or absent if any
+///   file in the group has a null partition value
+/// - `null_count` = the number of rows in files with a null partition value
 ///
 /// This allows query optimizers to prune entire file groups based on partition bounds.
 #[derive(Debug, Clone)]
 
@@ -117,9 +117,12 @@ impl FileRange {
 /// computed from [`Self::partition_values`]:
 ///
 /// - `min = max = partition_value` (all rows in a file share the same partition value)
-/// - `null_count = 0` (partition values extracted from paths are never null)
+/// - `null_count = 0` (every row has the file's partition value)
 /// - `distinct_count = 1` (single distinct value per file for each partition column)
 ///
+/// A null partition value instead gives `null_count = num_rows`, `distinct_count = 0`
+/// and absent `min`/`max`.
+///
 /// This enables query optimizers to use partition column bounds for pruning and planning.
 pub struct PartitionedFile {
     /// Path for the file (e.g. URL, filesystem path, etc)
@@ -316,8 +319,11 @@ impl PartitionedFile {
     /// This method will automatically append exact statistics for partition columns
     /// based on `partition_values`:
     /// - `min = max = partition_value` (all rows have the same value)
-    /// - `null_count = 0` (partition values from paths are never null)
+    /// - `null_count = 0` (all rows have a non-null value)
     /// - `distinct_count = 1` (all rows have the same partition value)
+    ///
+    /// A null partition value (e.g. a Hive default partition) instead gives
+    /// `null_count = num_rows`, `distinct_count = 0` and absent `min`/`max`.
     pub fn with_statistics(mut self, file_statistics: Arc<Statistics>) -> Self {
         if self.partition_values.is_empty() {
             // No partition columns, use stats as-is
@@ -326,11 +332,20 @@ impl PartitionedFile {
             // Extend stats with exact partition column statistics
             let mut stats = Arc::unwrap_or_clone(file_statistics);
             for partition_value in &self.partition_values {
+                let (null_count, min_max, distinct_count) = if partition_value.is_null() {
+                    (stats.num_rows, Precision::Absent, Precision::Exact(0))
+                } else {
+                    (
+                        Precision::Exact(0),
+                        Precision::Exact(partition_value.clone()),
+                        Precision::Exact(1),
+                    )
+                };
                 let col_stats = ColumnStatistics {
-                    null_count: Precision::Exact(0),
-                    max_value: Precision::Exact(partition_value.clone()),
-                    min_value: Precision::Exact(partition_value.clone()),
-                    distinct_count: Precision::Exact(1),
+                    null_count,
+                    max_value: min_max.clone(),
+                    min_value: min_max,
+                    distinct_count,
                     sum_value: Precision::Absent,
                     byte_size: partition_value
                         .data_type()
@@ -767,6 +782,29 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_with_statistics_null_partition_value() {
+        use crate::PartitionedFile;
+        use datafusion_common::stats::Precision;
+        use datafusion_common::{ScalarValue, Statistics};
+
+        let mut pf = PartitionedFile::new("test.parquet", 100);
+        pf.partition_values = vec![ScalarValue::Date32(None)];
+
+        let file_stats = Arc::new(Statistics {
+            num_rows: Precision::Exact(2),
+            total_byte_size: Precision::Exact(16),
+            column_statistics: vec![],
+        });
+
+        let stats = pf.with_statistics(file_stats).statistics.unwrap();
+        let partition_col_stats = &stats.column_statistics[0];
+        assert_eq!(partition_col_stats.null_count, Precision::Exact(2));
+        assert_eq!(partition_col_stats.min_value, Precision::Absent);
+        assert_eq!(partition_col_stats.max_value, Precision::Absent);
+        assert_eq!(partition_col_stats.distinct_count, Precision::Exact(0));
+    }
+
     #[test]
     fn test_url_contains() {
         let url = ListingTableUrl::parse("file:///var/data/mytable/").unwrap();