Repository navigation
Conversation
ParquetSink kept each written file's ParquetMetaData in a HashMap for the lifetime of the sink, and its writer tasks also handed the metadata back through a JoinSet that is only drained once the last file has been announced. Nothing in the engine read either; only the datafusion-core unit tests used ParquetSink::written(). For Atlas this meant a 5-day options NBBO weekly rollup (~700 output files) grew ~1 GiB of RSS per 10 GB written outside the memory pool until the 64 GiB container limit killed it. Writer tasks now reduce the closed file's metadata to its row count and compressed byte count for the rows_written / bytes_written metrics and drop the ParquetMetaData inside the task. The written map, ParquetSink::written() and the parking_lot dependency go away. The core tests read the footers back from a temporary local object store instead.
xudong963
force-pushed
the
xudong963/parquet-sink-drop-written-metadata
branch
from
November 3, 2026 00:21
86104be to
f5e5f51
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
nbbo_quotes_weekly_v1partition (~680 GiB, ~700 output files) grew the task pod's RSS by roughly 1 GiB per 10 GB written, outside the DataFusion memory pool, until the 64 GiB container limit killed it. Heap profiles put the retained memory on each written file'sParquetMetaData(polygon-io/rust-app-atlas#6549).ParquetSink; Parquet writer: Decimal/Float16 column chunk statistics alias the Arrow writer's value arena, keeping whole batches alive in ParquetMetaData apache/arrow-rs#12027 (fix in Fix: handle NULL input for regex match operations apache/datafusion#12028) is why a single file's metadata is so large (Decimal column statistics pin 128 KiB per-batch arenas).Rationale for this change
ParquetSinkkept every written file'sParquetMetaDatainwritten: HashMap<Path, ParquetMetaData>for the lifetime of the sink, i.e. for the whole partition write. Nothing in the engine reads that map; only theParquetSinkunit tests indatafusion-coredid, throughParquetSink::written().The retention also did not stop at the map: the writer tasks returned
(Path, ParquetMetaData)through aJoinSetthat is only drained after the demux loop has announced the last file, so completed tasks' metadata sat in theJoinSetfor just as long.What changes are included in this PR?
(rows, compressed bytes)(written_file_counts) and drop theParquetMetaDatainside the task. The sink keeps no per-file state after a file is closed; therows_written/bytes_writtenmetrics are unchanged.writtenmap,ParquetSink::written()and the duplicate-path check that rode on the map are removed. Output paths are generated by the demuxer (<write_id>_<n>), so the check never had a way to fire.parking_lotis dropped fromdatafusion-datasource-parquet; the mutex was its only use.datafusion-coreParquetSinktests get aSinkTestStorefixture (temp dir +LocalFileSystem) and read each written file's footer back from the store instead of asking the sink. As a side effect the temp directory now lives for the whole test instead of being deleted right after creation and silently recreated by the object store on the first write.Are these changes tested?
cargo test -p datafusion --lib -- datasource::file_format::parquet::tests::parquet_sink(8 tests: single/partitioned/parallel writes, arrow-schema metadata, extensions, memory reservation) passes.cargo clippy -p datafusion-datasource-parquet --all-targets -- -D warnings,cargo fmt --allandtaplo format --checkare clean.Are there any user-facing changes?
ParquetSink::written()is removed. Atlas does not call it; per-file metadata after a write is the subject of apache#23472 and would need a different, opt-in design.