apache / apache/datafusion

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

Open
#23,472 0 comments 1 reaction 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.