Datasource: Add file_metadata() to DataSink for per-file write metadata by qzyu999 · Pull Request #23656 · apache/datafusion

qzyu999 · 2026-07-17T01:01:36Z

Which issue does this PR close?

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:

Bypass DataFusion's write pipeline entirely (use PyArrow/parquet-rs directly)
Re-read Parquet footers after writing (additional I/O round-trips)
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?

New type in datafusion-datasource:
- FileWriteMetadata: path + row_count + byte_size + optional format-specific metadata bytes
Trait extension (non-breaking, default implementation):
- DataSink::file_metadata() Vec<FileWriteMetadata> (default: empty)
ParquetSink implementation:
- Overrides file_metadata() using the existing self.written HashMap that already collects ParquetMetaData during writes zero additional I/O
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 · 2026-07-17T04:16:36Z

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

Copilot

Pull request overview

Note

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Adds a standardized way for DataSink implementations to expose per-file write metadata (e.g., file path, row count, size, and optional format-specific bytes) so external consumers (Iceberg/Delta/Hudi) can commit written files without re-reading footers or downcasting to ParquetSink.

Changes:

Introduces FileWriteMetadata and adds a default DataSink::file_metadata() accessor (returns empty by default).
Implements file_metadata() for ParquetSink using already-collected written metadata (no extra I/O).
Adds a DataSinkExec::file_metadata() convenience method and new unit tests in both crates.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.

File	Description
datafusion/datasource/src/sink.rs	Adds `FileWriteMetadata`, extends `DataSink`/`DataSinkExec` API, and adds trait-level tests.
datafusion/datasource-parquet/src/sink.rs	Implements `ParquetSink::file_metadata()` and adds parquet-specific tests validating behavior.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

+    fn build_test_ctx(store_url: &ObjectStoreUrl) -> Arc<TaskContext> {
+        let tmp_dir = tempfile::TempDir::new().unwrap();
+        let local = Arc::new(
+            LocalFileSystem::new_with_prefix(&tmp_dir)
+                .expect("should create object store"),
+        );
+
+        let session = SessionConfig::default();
+        let runtime = RuntimeEnv::default();
+        runtime
+            .object_store_registry
+            .register_store(store_url.as_ref(), local);
+
+        Arc::new(
+            TaskContext::default()
+                .with_session_config(session)
+                .with_runtime(Arc::new(runtime)),
+        )
+    }


+        let file_sink_config = FileSinkConfig {
+            original_url: String::default(),
+            object_store_url: object_store_url.clone(),
+            file_group: FileGroup::new(vec![PartitionedFile::new("/tmp".to_string(), 1)]),
+            table_paths: vec![ListingTableUrl::parse("file:///tmp/test/").unwrap()],
+            output_schema: Arc::clone(&schema),
+            table_partition_cols: vec![],
+            insert_op: InsertOp::Overwrite,
+            keep_partition_by_columns: false,
+            file_extension: "parquet".into(),
+            file_output_mode: FileOutputMode::Automatic,
+        };


+                let row_count = parquet_meta.file_metadata().num_rows() as u64;
+                let byte_size: u64 = parquet_meta
+                    .row_groups()
+                    .iter()
+                    .map(|rg| rg.compressed_size() as u64)
+                    .sum();


+    /// Sum of compressed row group sizes in bytes.
+    ///
+    /// Note: this may differ slightly from the actual on-disk file size as it
+    /// excludes the Parquet footer, page indexes, and other metadata overhead.
+    pub byte_size: u64,


timsaucer

This looks like a good addition. I have a couple of suggestions and a question. Also I think the copilot suggestions look like they should be addressed.

timsaucer · 2026-07-28T19:18:37Z

+    fn make_test_batch(schema: &SchemaRef) -> RecordBatch {
+        let col_a: ArrayRef = Arc::new(StringArray::from(vec!["foo", "bar", "baz"]));
+        let col_b: ArrayRef = Arc::new(StringArray::from(vec!["one", "two", "three"]));
+        RecordBatch::try_new(Arc::clone(schema), vec![col_a, col_b]).unwrap()


There is a record_batch! macro that makes this a little more pleasant.

timsaucer · 2026-07-28T19:19:43Z

+        let batch =
+            RecordBatch::try_new(Arc::clone(&schema), vec![col_a, col_b]).unwrap();


Same comment as above about reusing the record_batch! macro

timsaucer · 2026-07-28T19:21:50Z

+    /// Sum of compressed row group sizes in bytes.
+    ///
+    /// Note: this may differ slightly from the actual on-disk file size as it
+    /// excludes the Parquet footer, page indexes, and other metadata overhead.


FileWriteMetada should be format agnostic, right? Here we're talking about row group sizes, which isn't a concept in all of the datasources. I think we just need a comment update here, but if this is specific to parquet then it probably needs to go into the format_metadata.

timsaucer · 2026-07-28T19:23:30Z

+    /// reconstruct column statistics without re-reading the file.
+    ///
+    /// For formats that do not produce file-level metadata this is `None`.
+    pub format_metadata: Option<Bytes>,


I suppose the other option, which matches some metadata approaches is that these be key/value pairs. That has the advantage of being human readable but the downside of potential data loss on conversion, such as f64 -> String. Here we have to pay a serialization/deserialization cost so parsing strings is likely not too much less performant. It's probably not a hot path anyway.

github-actions Bot added the datasource Changes to the datasource crate label Jul 17, 2026

qzyu999 mentioned this pull request Jul 17, 2026

Expose per-file write metadata from DataFrame.write_parquet() apache/datafusion-python#1637

Open

qzyu999 force-pushed the datasink-write-metadata branch from ed75a23 to a4d415e Compare July 17, 2026 02:24

qzyu999 force-pushed the datasink-write-metadata branch from a4d415e to f9c560a Compare July 17, 2026 09:43

timsaucer requested review from Copilot and timsaucer July 28, 2026 18:34

Copilot AI reviewed Jul 28, 2026

View reviewed changes

Copilot started reviewing on behalf of timsaucer July 28, 2026 19:24 View session

timsaucer reviewed Jul 28, 2026

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

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

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

qzyu999 commented Jul 17, 2026

Uh oh!

codecov-commenter commented Jul 17, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

timsaucer left a comment

Uh oh!

timsaucer Jul 28, 2026

Uh oh!

timsaucer Jul 28, 2026

Uh oh!

timsaucer Jul 28, 2026

Uh oh!

timsaucer Jul 28, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

		let batch =
		RecordBatch::try_new(Arc::clone(&schema), vec![col_a, col_b]).unwrap();

Uh oh!

Conversation

qzyu999 commented Jul 17, 2026

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Uh oh!

codecov-commenter commented Jul 17, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

timsaucer left a comment

Choose a reason for hiding this comment

Uh oh!

timsaucer Jul 28, 2026

Choose a reason for hiding this comment

Uh oh!

timsaucer Jul 28, 2026

Choose a reason for hiding this comment

Uh oh!

timsaucer Jul 28, 2026

Choose a reason for hiding this comment

Uh oh!

timsaucer Jul 28, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

codecov-commenter commented Jul 17, 2026 •

edited

Loading