Expose per-file Parquet FileMetaData from write operations (ParquetSink)

[qzyu999]

Is your feature request related to a problem or challenge?

Lakehouse table formats (Apache Iceberg, Delta Lake, Apache Hudi) require detailed per-file Parquet metadata after writing data files. Specifically, they need the Parquet FileMetaData thrift structure which contains:

• Per-column compressed/uncompressed sizes
• Per-column value counts and null counts
• Per-column min/max statistics (lower_bounds / upper_bounds)
• Row group split offsets
• Total file size and record count

This metadata is used to construct manifest entries (Iceberg's DataFile, Delta's AddFile) that enable scan planning optimizations like partition pruning and column statistics-based row group skipping.

Currently, when using DataFusion's DataFrame.write_parquet() or COPY INTO ... FORMAT PARQUET, the ParquetSink writes the file correctly but only returns a count of rows written to the caller. The detailed FileMetaData is computed internally (it's in the Parquet footer) but is discarded — not exposed to the Python/Rust caller.

This forces lakehouse integrations (iceberg-rust, pyiceberg, delta-rs) to implement their own Parquet writers (wrapping arrow-rs ArrowWriter directly) to capture this metadata, bypassing DataFusion's write path entirely. This means they cannot benefit from DataFusion's bounded-memory write capabilities (spill-to-disk during large writes, partitioned output, etc.).

Describe the solution you'd like

Expose parquet::file::metadata::FileMetaData (or a subset of it) from ParquetSink after write completion. Concretely:

Option A (preferred): Return metadata in the output RecordBatch

The DataSink::write_all() currently returns a RecordBatch with a single count column. Extend this to optionally include per-file metadata columns:

count	path	file_size	column_sizes	null_counts	lower_bounds	upper_bounds
50000	s3://bucket/part-0.pq	4194304	{0: 2097152, ...}	{0: 0, 1: 42}	{0: 0x01...}	{0: 0xff...}

This could be opt-in via a session config flag (e.g., datafusion.execution.parquet.return_file_metadata = true) to avoid breaking existing behavior.

Option B: Callback/hook on ParquetSink

Allow users to provide a callback that receives FileMetaData for each written file:

let sink = ParquetSink::new(...)
    .with_metadata_callback(|path, metadata| {
        // Caller captures per-file metadata here
    });

Option C: Separate metadata query after write

After write_parquet(), allow querying the written file's metadata without re-reading the full file:

df.write_parquet('output/')
metadata = ctx.parquet_metadata('output/part-0.parquet')

Describe alternatives you've considered

1. Bypass DataFusion's write path entirely (current workaround): Use arrow-rs ArrowWriter directly with a custom metadata collector. This works but loses DataFusion's partitioned writes, memory management, and parallelism.

2. Re-read the Parquet footer after writing: Open the written file, seek to the footer, parse FileMetaData. This works but adds an extra I/O round-trip per file and is inefficient for cloud storage (requires a separate range-request to read the last N bytes).

3. Compute statistics from the Arrow data before writing: Traverse the Arrow arrays to compute min/max/null counts before calling write. This duplicates work that the Parquet writer already does internally and doesn't capture compressed sizes or split offsets (which are only known after encoding).

Additional context

Downstream consumers who would benefit:

• apache/iceberg-rust — needs DataFile metadata for Iceberg commits
• apache/iceberg-python (pyiceberg) — same, via datafusion-python
• delta-io/delta-rs — needs AddFile metadata for Delta commits
• Any DataFusion-based lakehouse integration

The specific fields needed (from Parquet's FileMetaData thrift):

Parquet Field	Iceberg Usage
row_groups[*].columns[*].meta_data.total_compressed_size	column_sizes
row_groups[*].columns[*].meta_data.num_values	value_counts
row_groups[*].columns[*].meta_data.statistics.null_count	null_value_counts
row_groups[*].columns[*].meta_data.statistics.min_value	lower_bounds
row_groups[*].columns[*].meta_data.statistics.max_value	upper_bounds
row_groups[*].columns[0].meta_data.data_page_offset	split_offsets
Total file size in bytes	file_size_in_bytes
num_rows (already returned)	record_count

Related upstream issue: apache/datafusion-python#1624 (per-session object store config — a prerequisite for efficient streaming writes to cloud storage)