[xudong963]

Which issue does this PR close?

• Closes ParquetSink keeps every written file's ParquetMetaData in memory until the sink finishes #27026.
• Related: Expose per-file Parquet FileMetaData from write operations (ParquetSink) #23472 (exposing per-file metadata from ParquetSink).

Rationale for this change

A COPY ... TO ... STORED AS PARQUET or INSERT INTO that writes many files keeps growing in memory until the statement finishes, outside the memory pool. ParquetSink kept every written file's ParquetMetaData (all column chunk statistics plus page indexes) until the sink was dropped, in two places: the writer tasks returned (Path, ParquetMetaData) through a JoinSet that is only drained after the demux loop has announced the last file, and the join loop then moved each one into written: HashMap<Path, ParquetMetaData>.

Nothing on the COPY / INSERT path reads that map. ParquetSink::written() does have users who construct the sink themselves and read the metadata of the one or few files it wrote: InfluxDB 3 (parquet_file/src/serialize.rs), datafusion_iceberg (builds DataFiles from it) and LakeSoul (file_sink_writer.rs). Their behavior is unchanged. #27026 has the production numbers: a ~700-file partition write grew RSS by roughly 1 GiB per 10 GB written until it was OOM-killed.

What changes are included in this PR?

• ParquetSink::with_retain_written_metadata(bool), default true. With false, written() returns an empty map and no file metadata is kept.
• ParquetFormat::create_writer_physical_plan turns it off for the sinks it builds for COPY and INSERT, since nothing on that path reads the metadata.
• The writer tasks now return only their file's row and compressed byte counts. record_written_file runs inside the task: it stores the ParquetMetaData in the map when retention is on, or drops it right there when it is off. Either way nothing is buffered in the JoinSet until the last file is announced. The duplicate path check moves with it. Metrics are unchanged.
• One test for the switch (parquet_sink_write_without_retained_metadata), and an upgrade guide entry for the ParquetFormat behavior change.

The retention switch is not part of the plan's protobuf encoding, like the map itself; a sink that went through a round-trip is created with ParquetSink::new and retains metadata. A proto field could be added if we want COPY on remote executors to benefit too.

What is the testing strategy for this PR?

• parquet_sink_write_without_retained_metadata checks that a sink with retention off still writes and counts the rows and that written() is empty.
• The existing parquet_sink_* tests (default retention) cover written() for single, partitioned and parallel writes and are unchanged. cargo test -p datafusion --lib -- datasource::file_format::parquet::tests::parquet_sink passes (9 tests).
• cargo clippy --all-targets --all-features -D warnings on datafusion-datasource-parquet, cargo clippy --tests on datafusion and cargo check --tests on datafusion-proto are clean.

Are there any user-facing changes?

New builder method ParquetSink::with_retain_written_metadata. Sinks created by ParquetFormat no longer retain written file metadata; anyone downcasting the sink of such a plan to call written() afterwards would now get an empty map and should construct the ParquetSink themselves (documented in the upgrade guide). No existing API is removed.

[matthewmturner]

I am wondering if users of datafusion may have used that returned parquet metadata for creating external row group indexes so while internally its not used it could potentially be disruptive to users. Maybe it would be safer to keep existing functionality as is and add the ability to opt out of returning metadata?

[xudong963]

I am wondering if users of datafusion may have used that returned parquet metadata for creating external row group indexes so while internally its not used it could potentially be disruptive to users. Maybe it would be safer to keep existing functionality as is and add the ability to opt out of returning metadata?

Good call, thanks @matthewmturner! I let llm re-check all github repos, lol, and there are real users of written() after all: InfluxDB 3 (parquet_file/src/serialize.rs), datafusion_iceberg (builds DataFiles from it) and LakeSoul (file_sink_writer.rs). All construct ParquetSink themselves and write one or a few files per sink, so the retention is harmless for them.

I’ll switch this PR to keep written() as is and add an opt-out on the builder, turn it off in ParquetFormat::create_writer_physical_plan (nothing on the COPY/INSERT path reads it), and still fix the JoinSet buffering so that opting out retains nothing.