Refactor PartitionedFile: add ordering field and new_from_meta constructor (#19596)

## Which issue does this PR close?

Part of #19433

## Rationale for this change

In preparation for ordering inference from Parquet metadata, we need to
be able to store per-file ordering information on `PartitionedFile`.
This PR adds the necessary infrastructure.

## What changes are included in this PR?

- Add `ordering: Option<LexOrdering>` field to `PartitionedFile` struct
- Add `new_from_meta(ObjectMeta)` constructor for creating files from
metadata (cleaner than manually constructing)
- Add `with_ordering()` builder method to set ordering information
- Add `with_partition_values()` builder method for consistency
- Update all `PartitionedFile` constructors to initialize `ordering:
None`
- Update callsites in test_util, proto, and substrait to use
`new_from_meta`

## Are these changes tested?

Yes, existing tests pass. The new field is currently always `None` so no
new tests are needed yet. Tests for ordering inference will come in a
follow-up PR.

## Are there any user-facing changes?

No user-facing API changes. The `ordering` field is public but users
typically construct `PartitionedFile` via the provided constructors
which handle this automatically.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.5 <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: 3 people

datafusion/core/src/datasource/file_format/mod.rs

@@ -73,17 +73,7 @@ pub(crate) mod test_util {
             .infer_stats(state, &store, file_schema.clone(), &meta)
             .await?;
 
-        let file_groups = vec![
-            vec![PartitionedFile {
-                object_meta: meta,
-                partition_values: vec![],
-                range: None,
-                statistics: None,
-                extensions: None,
-                metadata_size_hint: None,
-            }]
-            .into(),
-        ];
+        let file_groups = vec![vec![PartitionedFile::new_from_meta(meta)].into()];
 
         let exec = format
             .create_physical_plan(

datafusion/core/src/datasource/mod.rs

@@ -113,14 +113,7 @@ mod tests {
             version: None,
         };
 
-        let partitioned_file = PartitionedFile {
-            object_meta: meta,
-            partition_values: vec![],
-            range: None,
-            statistics: None,
-            extensions: None,
-            metadata_size_hint: None,
-        };
+        let partitioned_file = PartitionedFile::new_from_meta(meta);
 
         let f1 = Field::new("id", DataType::Int32, true);
         let f2 = Field::new("extra_column", DataType::Utf8, true);

datafusion/core/src/datasource/physical_plan/parquet.rs

@@ -54,7 +54,7 @@ mod tests {
     use datafusion_datasource::source::DataSourceExec;
 
     use datafusion_datasource::file::FileSource;
-    use datafusion_datasource::{FileRange, PartitionedFile, TableSchema};
+    use datafusion_datasource::{PartitionedFile, TableSchema};
     use datafusion_datasource_parquet::source::ParquetSource;
     use datafusion_datasource_parquet::{
         DefaultParquetFileReaderFactory, ParquetFileReaderFactory, ParquetFormat,
@@ -1527,14 +1527,7 @@ mod tests {
     #[tokio::test]
     async fn parquet_exec_with_range() -> Result<()> {
         fn file_range(meta: &ObjectMeta, start: i64, end: i64) -> PartitionedFile {
-            PartitionedFile {
-                object_meta: meta.clone(),
-                partition_values: vec![],
-                range: Some(FileRange { start, end }),
-                statistics: None,
-                extensions: None,
-                metadata_size_hint: None,
-            }
+            PartitionedFile::new_from_meta(meta.clone()).with_range(start, end)
         }
 
         async fn assert_parquet_read(
@@ -1616,21 +1609,15 @@ mod tests {
             .await
             .unwrap();
 
-        let partitioned_file = PartitionedFile {
-            object_meta: meta,
-            partition_values: vec![
+        let partitioned_file = PartitionedFile::new_from_meta(meta)
+            .with_partition_values(vec![
                 ScalarValue::from("2021"),
                 ScalarValue::UInt8(Some(10)),
                 ScalarValue::Dictionary(
                     Box::new(DataType::UInt16),
                     Box::new(ScalarValue::from("26")),
                 ),
-            ],
-            range: None,
-            statistics: None,
-            extensions: None,
-            metadata_size_hint: None,
-        };
+            ]);
 
         let expected_schema = Schema::new(vec![
             Field::new("id", DataType::Int32, true),
@@ -1711,20 +1698,13 @@ mod tests {
             .unwrap()
             .child("invalid.parquet");
 
-        let partitioned_file = PartitionedFile {
-            object_meta: ObjectMeta {
-                location,
-                last_modified: Utc.timestamp_nanos(0),
-                size: 1337,
-                e_tag: None,
-                version: None,
-            },
-            partition_values: vec![],
-            range: None,
-            statistics: None,
-            extensions: None,
-            metadata_size_hint: None,
-        };
+        let partitioned_file = PartitionedFile::new_from_meta(ObjectMeta {
+            location,
+            last_modified: Utc.timestamp_nanos(0),
+            size: 1337,
+            e_tag: None,
+            version: None,
+        });
 
         let file_schema = Arc::new(Schema::empty());
         let config = FileScanConfigBuilder::new(
@@ -2376,36 +2356,22 @@ mod tests {
         );
         let config = FileScanConfigBuilder::new(store_url, source)
             .with_file(
-                PartitionedFile {
-                    object_meta: ObjectMeta {
-                        location: Path::from(name_1),
-                        last_modified: Utc::now(),
-                        size: total_size_1,
-                        e_tag: None,
-                        version: None,
-                    },
-                    partition_values: vec![],
-                    range: None,
-                    statistics: None,
-                    extensions: None,
-                    metadata_size_hint: None,
-                }
-                .with_metadata_size_hint(123),
-            )
-            .with_file(PartitionedFile {
-                object_meta: ObjectMeta {
-                    location: Path::from(name_2),
+                PartitionedFile::new_from_meta(ObjectMeta {
+                    location: Path::from(name_1),
                     last_modified: Utc::now(),
-                    size: total_size_2,
+                    size: total_size_1,
                     e_tag: None,
                     version: None,
-                },
-                partition_values: vec![],
-                range: None,
-                statistics: None,
-                extensions: None,
-                metadata_size_hint: None,
-            })
+                })
+                .with_metadata_size_hint(123),
+            )
+            .with_file(PartitionedFile::new_from_meta(ObjectMeta {
+                location: Path::from(name_2),
+                last_modified: Utc::now(),
+                size: total_size_2,
+                e_tag: None,
+                version: None,
+            }))
             .build();
 
         let exec = DataSourceExec::from_data_source(config);

datafusion/core/src/test_util/parquet.rs

@@ -160,16 +160,8 @@ impl TestParquetFile {
                 .with_table_parquet_options(parquet_options.clone()),
         );
         let scan_config_builder =
-            FileScanConfigBuilder::new(self.object_store_url.clone(), source).with_file(
-                PartitionedFile {
-                    object_meta: self.object_meta.clone(),
-                    partition_values: vec![],
-                    range: None,
-                    statistics: None,
-                    extensions: None,
-                    metadata_size_hint: None,
-                },
-            );
+            FileScanConfigBuilder::new(self.object_store_url.clone(), source)
+                .with_file(PartitionedFile::new_from_meta(self.object_meta.clone()));
 
         let df_schema = Arc::clone(&self.schema).to_dfschema_ref()?;
 

datafusion/core/tests/parquet/custom_reader.rs

@@ -69,13 +69,9 @@ async fn route_data_access_ops_to_parquet_file_reader_factory() {
         store_parquet_in_memory(vec![batch]).await;
     let file_group = parquet_files_meta
         .into_iter()
-        .map(|meta| PartitionedFile {
-            object_meta: meta,
-            partition_values: vec![],
-            range: None,
-            statistics: None,
-            extensions: Some(Arc::new(String::from(EXPECTED_USER_DEFINED_METADATA))),
-            metadata_size_hint: None,
+        .map(|meta| {
+            PartitionedFile::new_from_meta(meta)
+                .with_extensions(Arc::new(String::from(EXPECTED_USER_DEFINED_METADATA)))
         })
         .collect();
 

datafusion/core/tests/parquet/page_pruning.rs

@@ -67,14 +67,7 @@ async fn get_parquet_exec(
         .await
         .unwrap();
 
-    let partitioned_file = PartitionedFile {
-        object_meta: meta,
-        partition_values: vec![],
-        range: None,
-        statistics: None,
-        extensions: None,
-        metadata_size_hint: None,
-    };
+    let partitioned_file = PartitionedFile::new_from_meta(meta);
 
     let df_schema = schema.clone().to_dfschema().unwrap();
     let execution_props = ExecutionProps::new();

datafusion/datasource-parquet/src/row_group_filter.rs

@@ -1559,14 +1559,7 @@ mod tests {
             ParquetObjectReader::new(Arc::new(in_memory), object_meta.location.clone())
                 .with_file_size(object_meta.size);
 
-        let partitioned_file = PartitionedFile {
-            object_meta,
-            partition_values: vec![],
-            range: None,
-            statistics: None,
-            extensions: None,
-            metadata_size_hint: None,
-        };
+        let partitioned_file = PartitionedFile::new_from_meta(object_meta);
 
         let reader = ParquetFileReader {
             inner,

datafusion/datasource/src/display.rs

@@ -287,13 +287,6 @@ mod tests {
             version: None,
         };
 
-        PartitionedFile {
-            object_meta,
-            partition_values: vec![],
-            range: None,
-            statistics: None,
-            extensions: None,
-            metadata_size_hint: None,
-        }
+        PartitionedFile::new_from_meta(object_meta)
     }
 }

datafusion/datasource/src/file_scan_config.rs

@@ -1661,43 +1661,40 @@ mod tests {
 
         impl From<File> for PartitionedFile {
             fn from(file: File) -> Self {
-                PartitionedFile {
-                    object_meta: ObjectMeta {
-                        location: Path::from(format!(
-                            "data/date={}/{}.parquet",
-                            file.date, file.name
-                        )),
-                        last_modified: chrono::Utc.timestamp_nanos(0),
-                        size: 0,
-                        e_tag: None,
-                        version: None,
-                    },
-                    partition_values: vec![ScalarValue::from(file.date)],
-                    range: None,
-                    statistics: Some(Arc::new(Statistics {
-                        num_rows: Precision::Absent,
-                        total_byte_size: Precision::Absent,
-                        column_statistics: file
-                            .statistics
-                            .into_iter()
-                            .map(|stats| {
-                                stats
-                                    .map(|(min, max)| ColumnStatistics {
-                                        min_value: Precision::Exact(
-                                            ScalarValue::Float64(min),
-                                        ),
-                                        max_value: Precision::Exact(
-                                            ScalarValue::Float64(max),
-                                        ),
-                                        ..Default::default()
-                                    })
-                                    .unwrap_or_default()
-                            })
-                            .collect::<Vec<_>>(),
-                    })),
-                    extensions: None,
-                    metadata_size_hint: None,
-                }
+                let object_meta = ObjectMeta {
+                    location: Path::from(format!(
+                        "data/date={}/{}.parquet",
+                        file.date, file.name
+                    )),
+                    last_modified: chrono::Utc.timestamp_nanos(0),
+                    size: 0,
+                    e_tag: None,
+                    version: None,
+                };
+                let statistics = Arc::new(Statistics {
+                    num_rows: Precision::Absent,
+                    total_byte_size: Precision::Absent,
+                    column_statistics: file
+                        .statistics
+                        .into_iter()
+                        .map(|stats| {
+                            stats
+                                .map(|(min, max)| ColumnStatistics {
+                                    min_value: Precision::Exact(ScalarValue::Float64(
+                                        min,
+                                    )),
+                                    max_value: Precision::Exact(ScalarValue::Float64(
+                                        max,
+                                    )),
+                                    ..Default::default()
+                                })
+                                .unwrap_or_default()
+                        })
+                        .collect::<Vec<_>>(),
+                });
+                PartitionedFile::new_from_meta(object_meta)
+                    .with_partition_values(vec![ScalarValue::from(file.date)])
+                    .with_statistics(statistics)
             }
         }
     }