ParquetSink keeps every written file's ParquetMetaData in memory until the sink finishes #27026

Describe the bug

ParquetSink keeps the ParquetMetaData of every file it has written for as long as the sink lives, i.e. until the whole DataSinkExec partition finishes. Two places hold it:

1. Each writer task returns (Path, ParquetMetaData) through a JoinSet (

   datafusion/datafusion/datasource-parquet/src/sink.rs

   Lines 273 to 275 in d38fde4

   	let mut file_write_tasks: JoinSet<
   	std::result::Result<(Path, ParquetMetaData), DataFusionError>,
   	> = JoinSet::new();

   ), but the JoinSet is only drained (

   datafusion/datafusion/datasource-parquet/src/sink.rs

   Line 357 in d38fde4

   while let Some(result) = file_write_tasks.join_next().await {

   ) after the demux loop has announced the last file (

   datafusion/datafusion/datasource-parquet/src/sink.rs

   Line 287 in d38fde4

   while let Some((file_metadata, mut rx)) = file_stream_rx.recv().await {

   ). Completed tasks' metadata sits in the JoinSet until then.
2. The join loop then moves each ParquetMetaData into written: HashMap<Path, ParquetMetaData> (

   datafusion/datafusion/datasource-parquet/src/sink.rs

   Line 72 in d38fde4

   written: Arc<parking_lot::Mutex<HashMap<Path, ParquetMetaData>>>,

   ,

   datafusion/datafusion/datasource-parquet/src/sink.rs

   Line 371 in d38fde4

   .try_insert(path.clone(), parquet_meta_data)

   ), where it stays until the sink is dropped.

Nothing in the engine reads that map. The only caller of ParquetSink::written() is the unit tests in datafusion/core/src/datasource/file_format/parquet.rs. The join loop itself only needs two numbers per file (num_rows and the sum of compressed row group sizes) for the rows_written / bytes_written metrics.

The retained memory scales with files written × metadata size. The metadata carries every column chunk's statistics, and the column/offset indexes when page indexes are enabled, so for wide tables it is not small. With parquet ≥ 59 it can also be far larger than it looks: Decimal128 min/max statistics are slices into the writer's 128 KiB per-batch arena and keep the whole arena alive (apache/arrow-rs#12027, fix in apache/arrow-rs#12028).

What we saw in production: a single DataSinkExec partition writing ~680 GiB as ~700 files (1 GiB each, Decimal price columns, page index on) grew the process RSS by roughly 1 GiB per 10 GB written, entirely outside the memory pool, until the 64 GiB container limit OOM-killed it. Heap profiles attributed the retained bytes to the per-file ParquetMetaData held by the sink. Disabling statistics on the decimal columns made RSS flat at ~8 GiB for the same job, which confirms where the memory was going.

To Reproduce

Any COPY ... TO ... STORED AS PARQUET (or INSERT INTO a listing table) that produces many files from one statement. Memory is released only when the statement finishes, not as files close. A sketch with datafusion-cli:

SET datafusion.execution.soft_max_rows_per_output_file = 100000;
COPY (SELECT CAST(value AS DECIMAL(18, 4)) AS price, value
      FROM generate_series(1, 100000000))
TO '/tmp/sink_out/' STORED AS PARQUET;

Watch the process RSS during the COPY: it keeps climbing with the number of files written (~1,000 here) and only drops once the statement returns. The effect is much larger with wider schemas, page indexes and decimal columns, as above.

Expected behavior

The sink should hold no per-file state once a file is closed. Memory of a write should be bounded by the files currently being written, not by how many files have been written so far.

Additional context

• Expose per-file Parquet FileMetaData from write operations (ParquetSink) #23472 asks to expose per-file metadata from ParquetSink for Iceberg/Delta commits. That is a reason to offer an opt-in way to receive each file's metadata, not to retain all of it by default. written() as it stands is both the wrong default and awkward for that use (it clones the whole map).
• We fixed this in our fork by having the writer tasks return only (rows, bytes), dropping the ParquetMetaData inside the task, and removing the map and written(): Stop ParquetSink from retaining every written file's ParquetMetaData massive-com/arrow-datafusion#88.
• For upstream written() is public, so a plausible path is: stop retaining by default, deprecate written(), and add an opt-in hook that receives (Path, ParquetMetaData) as each file closes, which would also cover Expose per-file Parquet FileMetaData from write operations (ParquetSink) #23472. I can open a PR if there is agreement on the direction.