Skip to content

Datasource: Add file_metadata() to DataSink for per-file write metadata#23656

Open
qzyu999 wants to merge 1 commit into
apache:mainfrom
qzyu999:datasink-write-metadata
Open

Datasource: Add file_metadata() to DataSink for per-file write metadata#23656
qzyu999 wants to merge 1 commit into
apache:mainfrom
qzyu999:datasink-write-metadata

Conversation

@qzyu999

@qzyu999 qzyu999 commented Jul 17, 2026

Copy link
Copy Markdown

Which issue does this PR close?

Closes #23472.

Rationale for this change

External table formats (Apache Iceberg, Delta Lake, Apache Hudi) require per-file column statistics (column_sizes, null_counts, lower/upper bounds, split_offsets) when committing written files to their catalogs. Currently, ParquetSink computes and stores this metadata internally (self.written) but the DataSink trait only returns a row count forcing external consumers to either:

  1. Bypass DataFusion's write pipeline entirely (use PyArrow/parquet-rs directly)
  2. Re-read Parquet footers after writing (additional I/O round-trips)
  3. Downcast DataSinkExec.sink() to ParquetSink and call written() (fragile, undocumented)

This PR adds a standard, non-breaking mechanism to expose per-file metadata through the DataSink trait.

What changes are included in this PR?

  1. New type in datafusion-datasource:

    • FileWriteMetadata: path + row_count + byte_size + optional format-specific metadata bytes
  2. Trait extension (non-breaking, default implementation):

    • DataSink::file_metadata() Vec<FileWriteMetadata> (default: empty)
  3. ParquetSink implementation:

    • Overrides file_metadata() using the existing self.written HashMap that already collects ParquetMetaData during writes zero additional I/O
  4. DataSinkExec convenience:

    • file_metadata() method that delegates to the inner sink, avoiding the downcast dance

The typed ParquetMetaData remains accessible via the existing ParquetSink::written() method for Rust consumers who need full column-level statistics directly. The format_metadata: Option<Bytes> field on FileWriteMetadata is reserved for future Thrift serialization when FFI consumers (e.g. Python via PyO3) need it.

Are these changes tested?

Yes 14 new tests across both crates:

datafusion-datasource (7 tests):

  • FileWriteMetadata equality and clone behavior
  • Default file_metadata() returns empty after write_all
  • Overridden file_metadata() returns correct entries
  • Idempotent behavior (calling twice returns same result)
  • DataSinkExec::file_metadata() delegates correctly

datafusion-datasource-parquet (7 tests):

  • Empty before write, populated after write
  • Row count consistency (write_all count == sum of per-file row_counts)
  • Consistent with existing ParquetSink::written() accessor
  • Idempotent across multiple calls
  • Partitioned writes produce one entry per partition
  • E2E test through DataSinkExec wrapper with Arc<dyn DataSink> dispatch

All existing parquet_sink_write* tests continue to pass unchanged.

Are there any user-facing changes?

New public API surface (additive only, no breaking changes):

  • DataSink::file_metadata() default method
  • FileWriteMetadata struct
  • DataSinkExec::file_metadata() convenience method

@codecov-commenter

codecov-commenter commented Jul 17, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 95.79580% with 14 lines in your changes missing coverage. Please review.
✅ Project coverage is 80.68%. Comparing base (7ca6e54) to head (f9c560a).
⚠️ Report is 5 commits behind head on main.

Files with missing lines Patch % Lines
datafusion/datasource/src/sink.rs 88.13% 14 Missing ⚠️
Additional details and impacted files
@@            Coverage Diff             @@
##             main   #23656      +/-   ##
==========================================
+ Coverage   80.66%   80.68%   +0.01%     
==========================================
  Files        1086     1086              
  Lines      366694   367027     +333     
  Branches   366694   367027     +333     
==========================================
+ Hits       295806   296121     +315     
- Misses      53265    53277      +12     
- Partials    17623    17629       +6     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
  • 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Add FileWriteMetadata struct to datafusion-datasource along with a new
default method on the DataSink trait:

- file_metadata(): Post-hoc accessor returning per-file path, row count,
  byte size, and optional format-specific metadata bytes after write_all
  completes.

The default implementation returns an empty Vec, so existing DataSink
implementations are unaffected.

ParquetSink overrides file_metadata() to expose path, row_count, and
byte_size for each written file using the existing self.written HashMap
that already collects ParquetMetaData during writes. The typed
ParquetMetaData remains accessible via ParquetSink::written() for Rust
consumers who need column-level statistics directly.

DataSinkExec gains a file_metadata() convenience accessor that delegates
to the underlying sink.

This enables external table formats (Iceberg, Delta Lake, Hudi) to obtain
per-file metadata after a DataFusion write without re-reading file footers
or bypassing the write pipeline entirely.

Closes apache#23472
@qzyu999
qzyu999 force-pushed the datasink-write-metadata branch from a4d415e to f9c560a Compare July 17, 2026 09:43
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

datasource Changes to the datasource crate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

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

2 participants