Expose per-file Parquet FileMetaData from write operations (ParquetSink)
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### 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:
```rust
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:
```python
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](https://github.com/apache/iceberg-rust) — needs `DataFile` metadata for Iceberg commits
- [apache/iceberg-python (pyiceberg)](https://github.com/apache/iceberg-python) — same, via datafusion-python
- [delta-io/delta-rs](https://github.com/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:** https://github.com/apache/datafusion-python/issues/1624 (per-session object store config — a prerequisite for efficient streaming writes to cloud storage)
Contributor guide
Research direction
Start by reading ParquetSink and DataSink::write_all, then trace DataFrame.write_parquet and COPY INTO to see where FileMetaData is discarded. Compare the proposed output-batch, callback, and metadata-query approaches against the existing count result and downstream needs. Done means an agreed API exposes per-file metadata without breaking existing write behavior, with coverage for the selected interface.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python, rust
- Domain
- backend-api-design, data-engineering
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 42/100