xudong963 opened a new issue, #27026: URL: https://github.com/apache/datafusion/issues/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` (https://github.com/apache/datafusion/blob/d38fde44039b4d8028105506a510f4c522001a9c/datafusion/datasource-parquet/src/sink.rs#L273-L275), but the `JoinSet` is only drained (https://github.com/apache/datafusion/blob/d38fde44039b4d8028105506a510f4c522001a9c/datafusion/datasource-parquet/src/sink.rs#L357) after the demux loop has announced the last file (https://github.com/apache/datafusion/blob/d38fde44039b4d8028105506a510f4c522001a9c/datafusion/datasource-parquet/src/sink.rs#L287). Completed tasks' metadata sits in the `JoinSet` until then. 2. The join loop then moves each `ParquetMetaData` into `written: HashMap<Path, ParquetMetaData>` (https://github.com/apache/datafusion/blob/d38fde44039b4d8028105506a510f4c522001a9c/datafusion/datasource-parquet/src/sink.rs#L72, https://github.com/apache/datafusion/blob/d38fde44039b4d8028105506a510f4c522001a9c/datafusion/datasource-parquet/src/sink.rs#L371), 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`: ```sql 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 - #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()`: 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 #23472. I can open a PR if there is agreement on the direction. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
